StreamVisitor

def visit(
    self,
    *,
    sub: "LogicSubscriber",
    channel: Optional[str] = None,
    list: Optional[str] = None,
    stream: Optional[str] = None,
) -> Optional[str]:
    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=[],
        )

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