[Brook] Removed the old Stream API.

temporalio/contrib/streams/__init__.py

@@ -1,16 +1,24 @@
 """Streams of Records that Workflows, Activities and clients write and read.
 
-Workflow code writes with :py:func:`workflow_stream` and :py:func:`create_workflow_stream`.
-Activities use :py:func:`activity_stream`. Clients read with :py:func:`for_handle` or :py:func:`for_ref` and create
-standalone Streams with :py:func:`create_stream`. A Stream plugin, such as
-:py:class:`temporalio.contrib.streams.redis.RedisStreamPlugin`, carries the calls.
+A Stream definition, a class with ``@streams.defn``, gives the Stream name and the type of each
+Topic. Workflow code creates a Stream with
+:py:func:`temporalio.contrib.streams.workflow.create_stream`. A client creates and gets Streams
+with :py:class:`StreamClient`. A Stream is data: a Workflow passes it to an Activity, a child
+Workflow or the next run as an argument, and the parameter's type gives the handle class of
+that context, such as :py:class:`ActivityStream`.
+
+A Stream plugin carries the calls and registers a backend name with ``name=``, such as
+:py:class:`temporalio.contrib.streams.redis.RedisStreamPlugin` (``redis``) and
+:py:class:`temporalio.contrib.streams.testing.MemoryStreamPlugin` (``memory``). A handle
+carries the name of its backend. A client may have two plugins, and ``default=True`` marks the
+one that a create with no ``backend=`` uses. For now a Worker has one Stream plugin, so it
+serves one backend.
 
 .. warning::
     This package is experimental.
 """
 
 from temporalio.contrib.streams import workflow
-from temporalio.contrib.streams._activity import activity_stream
 from temporalio.contrib.streams._activity_stream import (
     ActivityStream,
     ActivityTopic,
@@ -19,11 +27,6 @@
 )
 from temporalio.contrib.streams._backend import StreamBackend
 from temporalio.contrib.streams._buffered import BufferedPublisher
-from temporalio.contrib.streams._client import (
-    create_stream,
-    for_handle,
-    for_ref,
-)
 from temporalio.contrib.streams._defn import (
     StreamDefinition,
     TopicDefinition,
@@ -45,18 +48,12 @@
 )
 from temporalio.contrib.streams._handle import StreamHandle
 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,
-    workflow_stream,
-)
+from temporalio.contrib.streams.workflow import WorkflowStream, WorkflowTopic
 
 __all__ = [
     "ActivityStream",
@@ -65,37 +62,29 @@
     "ClientStream",
     "ClientTopic",
     "Closed",
-    "OutsideStream",
     "ReadOnlyContextError",
     "Record",
     "StreamBackend",
     "StreamBufferFullError",
     "StreamClient",
-    "StreamDefinition",
     "StreamClosedError",
+    "StreamDefinition",
     "StreamError",
     "StreamExpiredError",
     "StreamFencedError",
     "StreamHandle",
     "StreamNotFoundError",
     "StreamOffsetError",
-    "StreamRef",
     "StreamStorageError",
     "StreamTransportMissingError",
     "StreamUnsupportedError",
     "Superseded",
     "SyncActivityStream",
     "SyncActivityTopic",
-    "Topic",
     "TopicDefinition",
     "WorkflowStream",
-    "activity_stream",
-    "create_stream",
-    "create_workflow_stream",
+    "WorkflowTopic",
     "defn",
-    "for_handle",
-    "for_ref",
     "topic",
     "workflow",
-    "workflow_stream",
 ]

temporalio/contrib/streams/_access.py

@@ -74,7 +74,7 @@ async def _read(
             types=types,
             from_offset=from_offset,
             wait_new_records=wait_new_records,
-            wait_for_create=create_wait,
+            create_wait=create_wait,
         )
         async for item in items:
             yield item

temporalio/contrib/streams/_activity.py

@@ -1,59 +1,46 @@
-"""Streams in Activity code."""
+"""The Stream state of one Activity attempt."""
 
 from __future__ import annotations
 
 import asyncio
 import contextvars
-from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
-from datetime import timedelta
 from typing import Any
 
 import temporalio.api.streamservice.v1 as stream_pb
 import temporalio.converter
-from temporalio.contrib.streams import _errors
 from temporalio.contrib.streams._backend import StreamBackend
-from temporalio.contrib.streams._outside import OutsideStream, batch_options
-from temporalio.contrib.streams._stream import StreamRef
-from temporalio.contrib.streams._writer import BatchOptions, StreamWriter
+from temporalio.contrib.streams._writer import StreamWriter
 
 
 @dataclass
 class ActivityStreams:
-    """What a Stream plugin gives the facade for one Activity attempt.
-
-    .. warning::
-        This API is experimental.
-    """
+    """What a Stream plugin gives the Activity handles for one Activity attempt."""
 
     backend: StreamBackend
     converter: temporalio.converter.DataConverter
     namespace: str
     producer: stream_pb.Producer
-    default_ref: StreamRef | None
     loop: asyncio.AbstractEventLoop
     backend_name: str = ""
-    batch: BatchOptions = BatchOptions()
     writers: dict[str, StreamWriter] = field(default_factory=dict)
     # The attempt's BufferedPublishers, which the end of the attempt flushes.
     publishers: list[Any] = field(default_factory=list)
 
-    def writer(self, stream_id: str, batch: BatchOptions | None = None) -> StreamWriter:
+    def writer(self, stream_id: str) -> StreamWriter:
         """The attempt's writer for a Stream, one per Stream."""
         writer = self.writers.get(stream_id)
         if writer is None:
+
+            async def known_id() -> str:
+                return stream_id
+
             writer = StreamWriter(
                 backend=self.backend,
                 converter=self.converter,
                 namespace=self.namespace,
-                stream_id=_known(stream_id),
+                stream_id=known_id,
                 producer=self.producer,
-                batch=batch or self.batch,
-                # The default Stream has no key until its first write, so its create goes
-                # first. A received Stream must exist already.
-                create_first=self.default_ref is not None
-                and stream_id == self.default_ref.id,
-                loop=self.loop,
             )
             self.writers[stream_id] = writer
         return writer
@@ -75,56 +62,6 @@ async def close(self) -> None:
             raise errors[0]
 
 
-def _known(stream_id: str) -> Callable[[], Awaitable[str]]:
-    async def find() -> str:
-        return stream_id
-
-    return find
-
-
 current: contextvars.ContextVar[ActivityStreams | None] = contextvars.ContextVar(
     "__temporal_activity_streams", default=None
 )
-
-
-def activity_stream(
-    ref: StreamRef | None = None,
-    *,
-    batch_interval: timedelta | None = None,
-    max_batch_size: int | None = None,
-    max_retry_duration: timedelta | None = None,
-) -> OutsideStream:
-    """A Stream for this Activity to write to and read.
-
-    Writes are buffered and flushed every ``batch_interval``. The plugin flushes before the
-    Activity completes, and a failed flush fails the attempt.
-
-    Args:
-        ref: A Stream that the Activity received. With none, the default Stream of the
-            Workflow that scheduled the Activity.
-        batch_interval: How often the writer flushes. Defaults to 100 ms.
-        max_batch_size: At most this many Records in one batch.
-        max_retry_duration: How long a flush retries a Store failure.
-    """
-    streams = current.get()
-    if streams is None:
-        raise _errors.StreamTransportMissingError(
-            "No Stream plugin runs this Activity, such as RedisStreamPlugin"
-        )
-    if ref is None:
-        if streams.default_ref is None:
-            raise _errors.StreamError(
-                "A Standalone Activity has no default Stream, so pass a StreamRef"
-            )
-        ref = streams.default_ref
-    stream_id = ref.id
-    batch = None
-    if batch_interval or max_batch_size or max_retry_duration:
-        batch = batch_options(batch_interval, max_batch_size, max_retry_duration)
-    return OutsideStream(
-        backend=streams.backend,
-        converter=streams.converter,
-        namespace=streams.namespace,
-        stream_id=stream_id,
-        writer=lambda: streams.writer(stream_id, batch),
-    )

temporalio/contrib/streams/_backend.py

@@ -2,9 +2,8 @@
 
 from __future__ import annotations
 
-import weakref
 from datetime import timedelta
-from typing import Any, TypeVar
+from typing import TypeVar
 
 import google.protobuf.message
 
@@ -48,8 +47,6 @@ class StreamBackend:
     def __init__(self, extension: temporalio.worker.WorkerExtension) -> None:
         """Create a backend over an extension."""
         self._extension = extension
-        # Writers whose open or encode may still run on the loop, for close_writers.
-        self.writers: weakref.WeakSet[Any] = weakref.WeakSet()
 
     @property
     def extension(self) -> temporalio.worker.WorkerExtension:
@@ -84,23 +81,10 @@ async def open_writer(self, request: ext_pb.OpenWriterRequest) -> int:
         )
         return response.writer_id
 
-    def publish(
-        self, writer_id: int, records: list[stream_pb.AppendRecord], force_flush: bool
-    ) -> None:
-        """Add encoded Records to a writer's buffer."""
-        self.call(
-            "Publish",
-            ext_pb.PublishRequest(
-                writer_id=writer_id, records=records, force_flush=force_flush
-            ),
-        )
-
     async def write_records(
         self, writer_id: int, records: list[stream_pb.AppendRecord]
     ) -> int:
-        """Write Records at once for a writer, after its buffered ones. Returns the last
-        Record's offset.
-        """
+        """Write Records at once for a writer. Returns the last Record's offset."""
         response = await self.call_async(
             "WriteRecords",
             ext_pb.WriteRecordsRequest(writer_id=writer_id, records=records),
@@ -124,15 +108,6 @@ async def flush_run_buffer(
         )
         return response.records
 
-    async def flush_writer(self, writer_id: int) -> int:
-        """Wait until the writer's Records are in the Store."""
-        response = await self.call_async(
-            "FlushWriter",
-            ext_pb.FlushWriterRequest(writer_id=writer_id),
-            ext_pb.FlushWriterResponse,
-        )
-        return response.last_offset
-
     async def close_writer(self, writer_id: int) -> int:
         """Flush and drop a writer. The Stream stays open."""
         response = await self.call_async(
@@ -143,17 +118,11 @@ async def close_writer(self, writer_id: int) -> int:
         return response.last_offset
 
     async def close_writers(self) -> int:
-        """Flush and drop every open writer of the extension, and return how many failed."""
-        failed = 0
-        for writer in list(self.writers):
-            try:
-                await writer.settle()
-            except Exception:
-                failed += 1
+        """Drop every open writer of the extension, and return how many failed."""
         response = await self.call_async(
             "CloseWriters", ext_pb.CloseWritersRequest(), ext_pb.CloseWritersResponse
         )
-        return failed + response.failed
+        return response.failed
 
     async def drain_closes(self, timeout: timedelta) -> int:
         """Wait for the Workflow closes in flight, and return how many are left."""
@@ -194,17 +163,6 @@ async def call(
         )
 
 
-def backend_of_client(client: temporalio.client.Client) -> StreamBackend:
-    """The backend that a client's Stream plugin gives."""
-    for plugin in client.config()["plugins"]:
-        backend = getattr(plugin, "_temporal_stream_backend", None)
-        if isinstance(backend, StreamBackend):
-            return backend
-    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]: