[Brook] Added StreamClient.

temporalio/contrib/streams/__init__.py

@@ -20,7 +20,6 @@
 from temporalio.contrib.streams._backend import StreamBackend
 from temporalio.contrib.streams._buffered import BufferedPublisher
 from temporalio.contrib.streams._client import (
-    ClientStream,
     create_stream,
     for_handle,
     for_ref,
@@ -48,6 +47,11 @@
 from temporalio.contrib.streams._items import Closed, Record, Superseded
 from temporalio.contrib.streams._outside import OutsideStream
 from temporalio.contrib.streams._stream import StreamRef, Topic
+from temporalio.contrib.streams._stream_client import (
+    ClientStream,
+    ClientTopic,
+    StreamClient,
+)
 from temporalio.contrib.streams._workflow import (
     WorkflowStream,
     create_workflow_stream,
@@ -59,12 +63,14 @@
     "ActivityTopic",
     "BufferedPublisher",
     "ClientStream",
+    "ClientTopic",
     "Closed",
     "OutsideStream",
     "ReadOnlyContextError",
     "Record",
     "StreamBackend",
     "StreamBufferFullError",
+    "StreamClient",
     "StreamDefinition",
     "StreamClosedError",
     "StreamError",

temporalio/contrib/streams/_backend.py

@@ -203,3 +203,41 @@ def backend_of_client(client: temporalio.client.Client) -> StreamBackend:
     raise _errors.StreamTransportMissingError(
         "The client has no Stream plugin, such as RedisStreamPlugin"
     )
+
+
+def client_backend(
+    client: temporalio.client.Client, name: str | None
+) -> tuple[str, StreamBackend]:
+    """The name and the backend of a client's Stream plugin. None picks the plugin marked
+    as the default, or else the first one.
+
+    Raises:
+        StreamTransportMissingError: The client has no Stream plugin.
+        StreamUnsupportedError: The client has no Stream plugin of that name.
+    """
+    backends = [
+        (
+            getattr(plugin, "backend_name", ""),
+            backend,
+            getattr(plugin, "default", False),
+        )
+        for plugin in client.config()["plugins"]
+        if isinstance(
+            backend := getattr(plugin, "_temporal_stream_backend", None), StreamBackend
+        )
+    ]
+    if not backends:
+        raise _errors.StreamTransportMissingError(
+            "The client has no Stream plugin, such as RedisStreamPlugin"
+        )
+    if not name:
+        defaults = [entry for entry in backends if entry[2]] or backends
+        return defaults[0][0], defaults[0][1]
+    for backend_name, backend, _ in backends:
+        if backend_name == name:
+            return backend_name, backend
+    raise _errors.StreamUnsupportedError(
+        f"The client has no Stream backend {name}. It has "
+        f"{', '.join(entry[0] for entry in backends)}.",
+        reason="stream_unsupported",
+    )

temporalio/contrib/streams/_buffered.py

@@ -8,7 +8,8 @@
 from typing import Any, Generic, TypeVar
 
 from temporalio.contrib.streams import _activity
-from temporalio.contrib.streams._access import _AsyncTopic
+from temporalio.contrib.streams._activity_stream import ActivityTopic
+from temporalio.contrib.streams._stream_client import ClientTopic
 
 T = TypeVar("T")
 
@@ -29,7 +30,7 @@ class BufferedPublisher(Generic[T]):
 
     def __init__(
         self,
-        topic: _AsyncTopic[T],
+        topic: ActivityTopic[T] | ClientTopic[T],
         *,
         max_delay: timedelta = timedelta(milliseconds=100),
     ) -> None: