FakeProducer

@override
async def publish(self, cmd: "RedisPublishCommand") -> int | bytes:
    body = await build_message(
        message=cmd.body,
        reply_to=cmd.reply_to,
        correlation_id=cmd.correlation_id or self.broker.config.id_generator(),
        headers=cmd.headers,
        message_format=cmd.message_format,
        serializer=self.broker.config.fd_config._serializer,
        codec=self.codec,
    )
    destination = _make_destination_kwargs(cmd)
    visitors = (ChannelVisitor(), ListVisitor(), StreamVisitor())
    session_id = uuid.uuid4()
    for visitor, visited_ch, handler in self._find_handlers(
        destination=destination,
        visitors=visitors,
        cmd=cmd,
        session_id=session_id,
    ):
        msg = visitor.get_message(
            visited_ch,
            body,
            handler,
        )
        self._put_pel(msg=msg, cmd=cmd, handler=handler, session_id=session_id)
        await self._execute_handler(msg, handler, session_id=session_id)

    return 0

@override
async def request(self, cmd: "RedisPublishCommand") -> "PubSubMessage":
    body = await build_message(
        message=cmd.body,
        correlation_id=cmd.correlation_id or self.broker.config.id_generator(),
        headers=cmd.headers,
        message_format=cmd.message_format,
        serializer=self.broker.config.fd_config._serializer,
        codec=self.codec,
    )

    destination = _make_destination_kwargs(cmd)
    visitors = (ChannelVisitor(), ListVisitor(), StreamVisitor())
    session_id = uuid.uuid4()

    for visitor, visited_ch, handler in self._find_handlers(
        destination=destination,
        visitors=visitors,
        cmd=cmd,
        session_id=session_id,
    ):
        msg = visitor.get_message(
            visited_ch,
            body,
            handler,
        )
        self._put_pel(msg=msg, cmd=cmd, handler=handler, session_id=session_id)
        with anyio.fail_after(cmd.timeout):
            return await self._execute_handler(msg, handler, session_id=session_id)

    raise SubscriberNotFound

@override
async def publish_batch(self, cmd: "RedisPublishCommand") -> int:
    data_to_send = [
        await build_message(
            m,
            correlation_id=cmd.correlation_id or self.broker.config.id_generator(),
            headers=cmd.headers,
            message_format=cmd.message_format,
            serializer=self.broker.config.fd_config._serializer,
            codec=self.codec,
        )
        for m in cmd.batch_bodies
    ]
    session_id = uuid.uuid4()

    for visitor, visited_ch, handler in self._find_handlers(
        {"list": cmd.destination},
        (ListVisitor(),),
        cmd=cmd,
        session_id=session_id,
    ):
        casted_handler = cast("_ListHandlerMixin", handler)

        if casted_handler.list_sub.batch:
            msg = visitor.get_message(
                visited_ch,
                data_to_send,
                casted_handler,
            )
            self._put_pel(msg=msg, cmd=cmd, handler=handler, session_id=session_id)
            await self._execute_handler(msg, handler, session_id=session_id)

    return 0

def connect(
    self,
    serializer: Optional["SerializerProto"] = None,
    codec: Optional["CodecProto"] = None,
) -> None:
    self.serializer = serializer
    if codec is not None:
        self.codec = codec