Skip to content

MQTTBrokerConfig

faststream.mqtt.broker.config.MQTTBrokerConfig dataclass #

MQTTBrokerConfig(
    *,
    prefix: str = "",
    include_in_schema: bool | None = True,
    broker_middlewares: Sequence[
        BrokerMiddleware[Any]
    ] = (),
    broker_parser: Optional[CustomCallable] = None,
    broker_decoder: Optional[CustomCallable] = None,
    broker_codec: Optional[CodecProto] = None,
    producer: ZmqttBaseProducer = ZmqttFakeProducer(),
    logger: LoggerState = LoggerState(),
    fd_config: FastDependsConfig = FastDependsConfig(),
    id_generator: IdGenerator = gen_cor_id,
    broker_dependencies: Sequence[Dependant] = (),
    graceful_timeout: float | None = 15.0,
    ack_policy: AckPolicy = (lambda: EMPTY)(),
    extra_context: dict[str, Any] = dict(),
    version: MQTTVersion | Literal["unset"] = "unset",
)

Bases: BrokerConfig

version class-attribute instance-attribute #

version: MQTTVersion | Literal['unset'] = 'unset'

producer class-attribute instance-attribute #

producer: ZmqttBaseProducer = field(
    default_factory=ZmqttFakeProducer
)

client property #

client: MQTTClient

prefix class-attribute instance-attribute #

prefix: str = ''

include_in_schema class-attribute instance-attribute #

include_in_schema: bool | None = True

broker_middlewares class-attribute instance-attribute #

broker_middlewares: Sequence[BrokerMiddleware[Any]] = ()

broker_parser class-attribute instance-attribute #

broker_parser: Optional[CustomCallable] = None

broker_decoder class-attribute instance-attribute #

broker_decoder: Optional[CustomCallable] = None

broker_codec class-attribute instance-attribute #

broker_codec: Optional[CodecProto] = None

logger class-attribute instance-attribute #

logger: LoggerState = field(default_factory=LoggerState)

fd_config class-attribute instance-attribute #

fd_config: FastDependsConfig = field(
    default_factory=FastDependsConfig
)

id_generator class-attribute instance-attribute #

id_generator: IdGenerator = gen_cor_id

broker_dependencies class-attribute instance-attribute #

broker_dependencies: Sequence[Dependant] = ()

graceful_timeout class-attribute instance-attribute #

graceful_timeout: float | None = 15.0

ack_policy class-attribute instance-attribute #

ack_policy: AckPolicy = field(default_factory=lambda: EMPTY)

extra_context class-attribute instance-attribute #

extra_context: dict[str, Any] = field(default_factory=dict)

context property #

context: ContextRepo

connect #

connect(client: MQTTClient) -> None
Source code in faststream/mqtt/broker/config.py 
44
45
46
47
48
def connect(self, client: "zmqtt.MQTTClient") -> None:
    self._client = client
    self.producer.connect(
        client, self.fd_config._serializer, codec=self.broker_codec or DefaultCodec()
    )

disconnect #

disconnect() -> None
Source code in faststream/mqtt/broker/config.py 
50
51
52
def disconnect(self) -> None:
    self._client = None
    self.producer.disconnect()

add_middleware #

add_middleware(middleware: BrokerMiddleware[Any]) -> None
Source code in faststream/mqtt/broker/config.py 
54
55
56
def add_middleware(self, middleware: "BrokerMiddleware[Any]") -> None:
    self._validate_middleware(middleware)
    return super().add_middleware(middleware)

insert_middleware #

insert_middleware(
    middleware: BrokerMiddleware[Any],
) -> None
Source code in faststream/mqtt/broker/config.py 
58
59
60
def insert_middleware(self, middleware: "BrokerMiddleware[Any]") -> None:
    self._validate_middleware(middleware)
    return super().insert_middleware(middleware)