RedisBroker

async def close(
    self,
    exc_type: Optional[Type[BaseException]] = None,
    exc_val: Optional[BaseException] = None,
    exec_tb: Optional[TracebackType] = None,
) -> None:
    """Closes the object.

    Args:
        exc_type: The type of the exception being handled, if any.
        exc_val: The exception instance being handled, if any.
        exec_tb: The traceback of the exception being handled, if any.

    Returns:
        None

    Raises:
        NotImplementedError: If the method is not implemented.

    """
    super()._abc_close(exc_type, exc_val, exec_tb)

    for h in self.handlers.values():
        await h.close()

    if self._connection is not None:
        await self._close(exc_type, exc_val, exec_tb)

async def connect(
    self,
    *args: Any,
    **kwargs: Any,
) -> "Redis[bytes]":
    """Connect to the Redis server.

    Args:
        args : additional positional arguments
        kwargs : additional keyword arguments
    """
    connection = await super().connect(*args, **kwargs)
    for p in self._publishers.values():
        p._producer = self._producer
    return connection

def include_router(self, router: BrokerRouter[Any, MsgType]) -> None:
    """Includes a router in the current object.

    Args:
        router: The router to be included.

    Returns:
        None

    """
    for r in router._handlers:
        self.subscriber(*r.args, **r.kwargs)(r.call)

    self._publishers = {**self._publishers, **router._publishers}

def include_routers(self, *routers: BrokerRouter[Any, MsgType]) -> None:
    """Includes routers in the current object.

    Args:
        *routers: Variable length argument list of routers to include.

    Returns:
        None

    """
    for r in routers:
        self.include_router(r)

@override
async def publish(  # type: ignore[override]
    self,
    *args: Any,
    **kwargs: Any,
) -> Optional[DecodedMessage]:
    assert self._producer, NOT_CONNECTED_YET  # nosec B101
    return await self._producer.publish(*args, **kwargs)

async def publish_batch(
    self,
    *args: Any,
    **kwargs: Any,
) -> None:
    assert self._producer, NOT_CONNECTED_YET  # nosec B101
    return await self._producer.publish_batch(*args, **kwargs)

@override
def publisher(  # type: ignore[override]
    self,
    channel: Union[Channel, PubSub, None] = None,
    list: Union[Channel, ListSub, None] = None,
    stream: Union[Channel, StreamSub, None] = None,
    headers: Optional[AnyDict] = None,
    reply_to: str = "",
    # AsyncAPI information
    title: Optional[str] = None,
    description: Optional[str] = None,
    schema: Optional[Any] = None,
    include_in_schema: bool = True,
) -> Publisher:
    channel = PubSub.validate(channel)
    list = ListSub.validate(list)
    stream = StreamSub.validate(stream)

    any_of = channel or list or stream
    if any_of is None:
        raise ValueError(INCORRECT_SETUP_MSG)

    key = Handler.get_routing_hash(any_of)
    publisher = self._publishers.get(
        key,
        Publisher(
            channel=channel,
            list=list,
            stream=stream,
            headers=headers,
            reply_to=reply_to,
            # AsyncAPI
            title=title,
            _description=description,
            _schema=schema,
            include_in_schema=include_in_schema,
        ),
    )
    super().publisher(key, publisher)
    if self._producer is not None:
        publisher._producer = self._producer
    return publisher

async def start(self) -> None:
    context.set_global(
        "default_log_context",
        self._get_log_context(None, ""),
    )

    await super().start()
    assert self._connection, NOT_CONNECTED_YET  # nosec B101

    for handler in self.handlers.values():
        if (stream := handler.stream_sub) is not None and stream.group:
            try:
                await self._connection.xgroup_create(
                    name=stream.name,
                    groupname=stream.group,
                    mkstream=True,
                )
            except ResponseError as e:
                if "already exists" not in str(e):
                    raise e

        c = self._get_log_context(None, handler.channel_name)
        self._log(f"`{handler.call_name}` waiting for messages", extra=c)
        await handler.start(self._connection)

@override
def subscriber(  # type: ignore[override]
    self,
    channel: Union[Channel, PubSub, None] = None,
    *,
    list: Union[Channel, ListSub, None] = None,
    stream: Union[Channel, StreamSub, None] = None,
    # broker arguments
    dependencies: Sequence[Depends] = (),
    parser: Optional[CustomParser[AnyRedisDict, RedisMessage]] = None,
    decoder: Optional[CustomDecoder[RedisMessage]] = None,
    middlewares: Optional[
        Sequence[Callable[[AnyRedisDict], BaseMiddleware]]
    ] = None,
    filter: Filter[RedisMessage] = default_filter,
    # AsyncAPI information
    title: Optional[str] = None,
    description: Optional[str] = None,
    include_in_schema: bool = True,
    **original_kwargs: Any,
) -> Callable[
    [Callable[P_HandlerParams, T_HandlerReturn]],
    HandlerCallWrapper[Any, P_HandlerParams, T_HandlerReturn],
]:
    channel = PubSub.validate(channel)
    list = ListSub.validate(list)
    stream = StreamSub.validate(stream)

    if (any_of := channel or list or stream) is None:
        raise ValueError(INCORRECT_SETUP_MSG)

    if all((channel, list)):
        raise ValueError("You can't use `PubSub` and `ListSub` both")
    elif all((channel, stream)):
        raise ValueError("You can't use `PubSub` and `StreamSub` both")
    elif all((list, stream)):
        raise ValueError("You can't use `ListSub` and `StreamSub` both")

    self._setup_log_context(channel=any_of.name)
    super().subscriber()

    key = Handler.get_routing_hash(any_of)
    handler = self.handlers[key] = self.handlers.get(
        key,
        Handler(  # type: ignore[abstract]
            log_context_builder=partial(
                self._get_log_context,
                channel=any_of.name,
            ),
            graceful_timeout=self.graceful_timeout,
            # Redis
            channel=channel,
            list=list,
            stream=stream,
            # AsyncAPI
            title=title,
            description=description,
            include_in_schema=include_in_schema,
        ),
    )

    def consumer_wrapper(
        func: Callable[P_HandlerParams, T_HandlerReturn],
    ) -> HandlerCallWrapper[
        AnyRedisDict,
        P_HandlerParams,
        T_HandlerReturn,
    ]:
        handler_call, dependant = self._wrap_handler(
            func,
            extra_dependencies=dependencies,
            **original_kwargs,
        )

        handler.add_call(
            handler=handler_call,
            filter=filter,
            middlewares=middlewares,
            parser=parser or self._global_parser,  # type: ignore[arg-type]
            decoder=decoder or self._global_decoder,  # type: ignore[arg-type]
            dependant=dependant,
        )

        return handler_call

    return consumer_wrapper