StreamBatchSubscriber

@override
async def start(self) -> None:
    client = self._client

    self.extra_watcher_options.update(
        redis=client,
        group=self.stream_sub.group,
    )

    stream = self.stream_sub

    read: ReadCallable

    if stream.group and stream.consumer:
        group_create_id = "$" if self.last_id == ">" else self.last_id
        try:
            await client.xgroup_create(
                name=stream.name,
                id=group_create_id,
                groupname=stream.group,
                mkstream=stream.declare,
            )
        except ResponseError as e:
            if "already exists" not in str(e):
                raise
        else:
            self.read_id = ">"

        self.last_id = self.read_id

        if stream.min_idle_time is None:

            def read(_: str) -> Awaitable[ReadResponse]:
                return self._xreadgroup(
                    count=stream.max_records,
                    block=stream.polling_interval,
                    noack=stream.no_ack,
                )

        else:

            async def read(_: str) -> ReadResponse:
                stream_message = await client.xautoclaim(
                    name=self.stream_sub.name,
                    groupname=self.stream_sub.group,
                    consumername=self.stream_sub.consumer,
                    min_idle_time=self.min_idle_time,
                    start_id=self.autoclaim_start_id,
                    count=1,
                )
                stream_name = self.stream_sub.name.encode()
                (next_id, messages, *_) = stream_message

                # Update start_id for next call
                self.autoclaim_start_id = next_id

                if next_id == b"0-0" and not messages:
                    await asyncio.sleep(stream.polling_interval / 1000)  # ms to s
                    return ()

                return ((stream_name, messages),)

    else:

        def read(
            last_id: str,
        ) -> Awaitable[ReadResponse]:
            return client.xread(
                {stream.name: last_id},
                block=stream.polling_interval,
                count=stream.max_records,
            )

    await super().start(read)

async def stop(self) -> None:
    """Clean up handler subscription, cancel consume task in graceful mode."""
    await super().stop()

    for task in self.tasks:
        if not task.done():
            task.cancel()

    self.tasks.clear()

def add_call(
    self,
    *,
    parser_: Optional["CustomCallable"],
    decoder_: Optional["CustomCallable"],
    dependencies_: Sequence["Dependant"],
    codec_: Optional["CodecProto"] = None,
) -> Self:
    self._call_options = _CallOptions(
        parser=parser_,
        decoder=decoder_,
        dependencies=dependencies_,
        codec=codec_,
    )
    return self

async def consume(self, msg: MsgType) -> Any:
    """Consume a message asynchronously."""
    if not self.running:
        return None

    try:
        return await self.process_message(msg)

    except StopConsume:
        # Stop handler at StopConsume exception
        await self.stop()

    except SystemExit:
        # Stop handler at `exit()` call
        await self.stop()

        if app := self._outer_config.context.get("app"):
            app.exit()

    except Exception:  # nosec B110  # noqa: S110
        # All other exceptions were logged by CriticalLogMiddleware
        pass

async def process_message(self, msg: MsgType) -> "Response":
    """Execute all message processing stages."""
    context = self._outer_config.context
    logger_state = self._outer_config.logger

    async with AsyncExitStack() as stack:
        stack.enter_context(self.lock)

        # Enter context before middlewares
        stack.enter_context(
            context.scopes(
                (
                    ("handler_", self),
                    ("logger", logger_state.logger.logger),
                    *self._outer_config.extra_context.items(),
                ),
            ),
        )

        # enter all middlewares
        middlewares: list[BaseMiddleware] = []
        for base_m in self.__build__middlewares_stack():
            middleware = base_m(msg, context=context)
            middlewares.append(middleware)
            await middleware.__aenter__()

        cache: dict[Any, Any] = {}
        parsing_error: Exception | None = None
        for h in self.calls:
            try:
                message = await h.is_suitable(msg, cache)
            except Exception as e:
                parsing_error = e
                break

            if message is not None:
                stack.enter_context(
                    context.scopes(
                        (
                            ("log_context", self.get_log_context(message)),
                            ("message", message),
                        ),
                    ),
                )

                # Middlewares should be exited before scope release
                for m in middlewares:
                    stack.push_async_exit(m.__aexit__)

                result_msg = ensure_response(
                    await h.call(
                        message=message,
                        # consumer middlewares
                        _extra_middlewares=(
                            m.consume_scope for m in middlewares[::-1]
                        ),
                    ),
                )

                if not result_msg.correlation_id:
                    result_msg.correlation_id = message.correlation_id

                for p in chain(
                    self.__get_response_publisher(message),
                    h.handler._publishers,
                ):
                    await p._publish(
                        result_msg.as_publish_command(),
                        _extra_middlewares=(
                            m.publish_scope for m in middlewares[::-1]
                        ),
                    )

                # Return data for tests
                return result_msg

        # Suitable handler was not found or
        # parsing/decoding exception occurred
        for m in middlewares:
            stack.push_async_exit(m.__aexit__)

        # Reraise it to catch in tests
        if parsing_error:
            raise parsing_error

        error_msg = f"There is no suitable handler for {msg=}"
        raise SubscriberNotFound(error_msg)

    # An error was raised and processed by some middleware
    return ensure_response(None)

@override
async def get_one(
    self,
    *,
    timeout: float = 5.0,
) -> "RedisStreamMessage | None":
    assert not self.calls, (
        "You can't use `get_one` method if subscriber has registered handlers."
    )
    claim_meta: ClaimMeta | None = None

    if self.stream_sub.group and self.stream_sub.consumer:
        if self.min_idle_time is None:
            stream_message = await self._xreadgroup(
                count=1,
                block=math.ceil(timeout * 1000),
            )
            if not stream_message:
                return None

            ((stream_name, (entry,)),) = stream_message
            message_id, raw_message, claim_meta = self._parse_stream_entry(entry)
        else:
            stream_message = await self._client.xautoclaim(
                name=self.stream_sub.name,
                groupname=self.stream_sub.group,
                consumername=self.stream_sub.consumer,
                min_idle_time=self.min_idle_time,
                start_id=self.autoclaim_start_id,
                count=1,
            )
            (next_id, messages, *_) = stream_message
            # Update start_id for next call
            self.autoclaim_start_id = next_id
            if not messages:
                return None
            stream_name = self.stream_sub.name.encode()
            ((message_id, raw_message),) = messages
    else:
        stream_message = await self._client.xread(
            {self.stream_sub.name: self.last_id},
            block=math.ceil(timeout * 1000),
            count=1,
        )
        if not stream_message:
            return None

        ((stream_name, ((message_id, raw_message),)),) = stream_message

    self.last_id = message_id.decode()

    redis_incoming_msg = DefaultStreamMessage(
        type="stream",
        channel=stream_name.decode(),
        message_ids=[message_id],
        data=raw_message,
    )
    _attach_claim_metadata(redis_incoming_msg, [claim_meta])

    context = self._outer_config.context
    async_parser, async_decoder = self._get_parser_and_decoder()

    msg: RedisStreamMessage = await process_msg(  # type: ignore[assignment]
        msg=redis_incoming_msg,
        middlewares=(
            m(redis_incoming_msg, context=context) for m in self._broker_middlewares
        ),
        parser=async_parser,
        decoder=async_decoder,
    )
    return msg

def get_log_context(
    self,
    message: Optional["BrokerStreamMessage[Any]"],
) -> dict[str, str]:
    return self.build_log_context(
        message=message,
        channel=self.stream_sub.name,
    )

def schema(self) -> dict[str, "SubscriberSpec"]:
    self._build_fastdepends_model()
    return self.specification.get_schema()

def add_task(
    self,
    func: Callable[..., Coroutine[Any, Any, Any]],
    func_args: tuple[Any, ...] | None = None,
    func_kwargs: dict[str, Any] | None = None,
    *,
    restart_on_failure: bool = True,
) -> None:
    args = func_args or ()
    kwargs = func_kwargs or {}
    task = asyncio.create_task(func(*args, **kwargs))
    callback = TaskCallbackSupervisor(
        func,
        func_args,
        func_kwargs,
        self,
        restart_on_failure=restart_on_failure,
    )
    task.add_done_callback(callback)
    self.tasks.append(task)

@staticmethod
def build_log_context(
    message: Optional["BrokerStreamMessage[Any]"],
    channel: str = "",
) -> dict[str, str]:
    return {
        "channel": channel,
        "message_id": getattr(message, "message_id", ""),
    }

async def consume_one(self, msg: Any) -> None:
    await self.consume(msg)