RabbitBrokerConfig

faststream.rabbit.configs.RabbitBrokerConfig dataclass

RabbitBrokerConfig(
    *,
    prefix="",
    include_in_schema=True,
    broker_middlewares=(),
    broker_parser=None,
    broker_decoder=None,
    producer=FakeAioPikaFastProducer(),
    logger=LoggerState(),
    fd_config=FastDependsConfig(),
    broker_dependencies=(),
    graceful_timeout=None,
    extra_context=dict(),
    channel_manager=FakeChannelManager(),
    declarer=FakeRabbitDeclarer(),
    virtual_host="",
    app_id=None,
)

Bases: BrokerConfig

prefix class-attribute instance-attribute

prefix = ''

include_in_schema class-attribute instance-attribute

include_in_schema = True

broker_middlewares class-attribute instance-attribute

broker_middlewares = ()

broker_parser class-attribute instance-attribute

broker_parser = None

broker_decoder class-attribute instance-attribute

broker_decoder = None

logger class-attribute instance-attribute

logger = field(default_factory=LoggerState)

fd_config class-attribute instance-attribute

fd_config = field(default_factory=FastDependsConfig)

broker_dependencies class-attribute instance-attribute

broker_dependencies = ()

graceful_timeout class-attribute instance-attribute

graceful_timeout = None

extra_context class-attribute instance-attribute

extra_context = field(default_factory=dict)

channel_manager class-attribute instance-attribute

channel_manager = field(default_factory=FakeChannelManager)

declarer class-attribute instance-attribute

declarer = field(default_factory=FakeRabbitDeclarer)

producer class-attribute instance-attribute

producer = field(default_factory=FakeAioPikaFastProducer)

virtual_host class-attribute instance-attribute

virtual_host = ''

app_id class-attribute instance-attribute

app_id = None

add_middleware

add_middleware(middleware)

Source code in faststream/_internal/configs/broker.py

def add_middleware(self, middleware: "BrokerMiddleware[Any]") -> None:
    self.broker_middlewares = (*self.broker_middlewares, middleware)

insert_middleware

insert_middleware(middleware)

Source code in faststream/_internal/configs/broker.py

def insert_middleware(self, middleware: "BrokerMiddleware[Any]") -> None:
    self.broker_middlewares = (middleware, *self.broker_middlewares)

connect

connect(connection)

Source code in faststream/rabbit/configs/broker.py

def connect(self, connection: "RobustConnection") -> None:
    self.channel_manager.connect(connection)
    self.producer.connect(serializer=self.fd_config._serializer)

disconnect

disconnect()

Source code in faststream/rabbit/configs/broker.py

def disconnect(self) -> None:
    self.channel_manager.disconnect()
    self.declarer.disconnect()
    self.producer.disconnect()