StreamVisitor

def visit(
    self,
    *,
    sub: "LogicSubscriber",
    channel: str | None = None,
    list: str | None = None,
    stream: str | None = None,
) -> str | None:
    if stream is None or not isinstance(sub, _StreamHandlerMixin):
        return None

    if stream == sub.stream_sub.name:
        return stream

    return None

def get_message(  # type: ignore[override]
    self,
    channel: str,
    body: Any,
    sub: "_StreamHandlerMixin",
) -> Any:
    if sub.stream_sub.batch:
        return BatchStreamMessage(
            type="bstream",
            channel=channel,
            data=[{bDATA_KEY: body}],
            message_ids=[],
        )

    return DefaultStreamMessage(
        type="stream",
        channel=channel,
        data={bDATA_KEY: body},
        message_ids=[],
    )