[Brook] Built the Stream handles on the generated Stream resource.

scripts/_wit/stream-service.wit

@@ -174,15 +174,6 @@ interface stream-service {
     stream-status-expired,
   }
 
-  /// @nexus.proto "temporal.api.streamservice.v1.CreateStreamResponse"
-  record create-stream-response {
-    %stream: option<stream-ref>,
-    /// @nexus.doc "False when the Stream existed with the same Retention."
-    created: option<bool>,
-    /// @nexus.doc "Closed when an earlier create and close used the id."
-    status: option<stream-status>,
-  }
-
   // The meta key of each Stream has a format version v. Lua refuses a write to a Stream with a
   // higher v with FAILED_PRECONDITION and the detail stream_format_too_new.
   /// @nexus.doc
@@ -290,8 +281,65 @@ interface stream-service {
     auto-close: option<duration>,
   }
 
+  // Its stream-ref is its serialized form, so a handle passes through Workflow input, Activity
+  // input and Signals as the ref. Each method fills in the ref of its request and takes the rest.
+  // The wire operations stay the same. A handle never changes: offsets are values that the
+  // methods return.
+  /// @nexus.doc "A handle on one Stream. Build it from a stream-ref, or get it from a create."
+  resource %stream {
+    constructor(%stream: stream-ref);
+
+    /// @nexus.doc "Append Records. The answer has the first and the last offset."
+    /// @nexus.operation "append-records"
+    append-records: func(
+      producer: option<producer>,
+      sequence: option<s64>,
+      records: list<append-record>,
+      previous-completed-task: option<workflow-task-epoch>,
+      failed-tasks: list<s64>,
+      withdraw-run: option<string>,
+    ) -> append-records-response;
+
+    /// @nexus.doc "Close the Stream. Readers drain what is left, then end."
+    /// @nexus.operation "close-stream"
+    close: func(
+      producer: option<producer>,
+      %result: option<payload>,
+      failure: option<failure>,
+      previous-completed-task: option<workflow-task-epoch>,
+      failed-tasks: list<s64>,
+    );
+
+    /// @nexus.doc "Read one page of Records. The iterator of the same name reads them all."
+    /// @nexus.operation "read-records"
+    read-records: func(
+      from-offset: option<s64>,
+      start: option<read-start>,
+      topics: list<string>,
+      topic-prefixes: list<string>,
+      max-records: option<s32>,
+      wait: option<duration>,
+      create-wait: option<duration>,
+    ) -> read-records-response;
+
+    /// @nexus.doc "Describe the Stream: its status, newest offset, Retention and auto-close."
+    /// @nexus.operation "describe-stream"
+    describe: func() -> describe-stream-response;
+  }
+
+  // The handle is built from the stream field of the response. The other fields stay beside
+  // it, since they describe this create, not the Stream.
+  /// @nexus.proto "temporal.api.streamservice.v1.CreateStreamResponse"
+  record create-stream-result {
+    %stream: own<%stream>,
+    /// @nexus.doc "False when the Stream existed with the same Retention."
+    created: option<bool>,
+    /// @nexus.doc "Closed when an earlier create and close used the id."
+    status: option<stream-status>,
+  }
+
   /// @nexus.operation name="CreateStream"
-  create-stream: func(request: create-stream-request) -> create-stream-response;
+  create-stream: func(request: create-stream-request) -> create-stream-result;
 
   /// @nexus.doc
   ///   "In Workflow code, Brook buffers this call in the run, and the facade drops the answer."
@@ -301,6 +349,8 @@ interface stream-service {
   /// @nexus.operation name="CloseStream"
   close-stream: func(request: close-stream-request) -> close-stream-response;
 
+  // The long poll moves from-offset, so nexgen also reads its pages as one async iterator. A WIT
+  // stream can't cross a Nexus call, so the read stays request and response on the wire.
   /// @nexus.doc "Not for Workflow code in Brook. A long poll. NOT_FOUND for an uncreated Stream."
   /// @nexus.operation name="ReadRecords"
   /// @nexus.long-poll

scripts/gen_stream_service.py

@@ -32,8 +32,8 @@
 output_dir = base_dir / "temporalio" / "contrib" / "streams" / "system"
 
 NEX_GEN_REPOSITORY = "https://github.com/moetemp/nexgen"
-# The head of moe/brook/v3/n7-long-poll-loop. A branch moves, so the pin is the commit.
-NEX_GEN_REVISION = "25b9c1be8f51245618297415aa51df6928c58504"
+# The head of moe/brook/v3/n8-stream-resource. A branch moves, so the pin is the commit.
+NEX_GEN_REVISION = "10fc35b40fcf48651c25e654a260bc71585f1429"
 nex_gen_root = base_dir / ".nexgen-streams" / NEX_GEN_REVISION
 
 

temporalio/bridge/Cargo.toml

@@ -94,7 +94,7 @@ url = "2.2"
 # crates, so the bridge sees one type of each.
 [dependencies.temporalio-streams]
 git = "https://github.com/moetemp/sdk-streams"
-rev = "daeb5fc2a13e1f3457a461f109e7f5b8eb31951c"
+rev = "2cc3eee3760a24b2feef30dc9471010cd49750e6"
 default-features = false
 optional = true
 

temporalio/contrib/streams/__init__.py

@@ -25,7 +25,6 @@
     SyncActivityStream,
     SyncActivityTopic,
 )
-from temporalio.contrib.streams._backend import StreamBackend
 from temporalio.contrib.streams._buffered import BufferedPublisher
 from temporalio.contrib.streams._defn import (
     StreamDefinition,
@@ -36,17 +35,18 @@
 from temporalio.contrib.streams._errors import (
     ReadOnlyContextError,
     StreamClosedError,
+    StreamConfigError,
     StreamError,
     StreamExpiredError,
     StreamFencedError,
     StreamNotFoundError,
     StreamOffsetError,
     StreamStorageError,
-    StreamTransportMissingError,
     StreamUnsupportedError,
 )
 from temporalio.contrib.streams._handle import StreamHandle
-from temporalio.contrib.streams._items import Closed, Record, Superseded
+from temporalio.contrib.streams._items import END, Closed, Record, Superseded
+from temporalio.contrib.streams._plugin import StreamPlugin
 from temporalio.contrib.streams._stream_client import (
     ClientStream,
     ClientTopic,
@@ -55,6 +55,7 @@
 from temporalio.contrib.streams.workflow import WorkflowStream, WorkflowTopic
 
 __all__ = [
+    "END",
     "ActivityStream",
     "ActivityTopic",
     "BufferedPublisher",
@@ -63,18 +64,18 @@
     "Closed",
     "ReadOnlyContextError",
     "Record",
-    "StreamBackend",
     "StreamClient",
     "StreamClosedError",
+    "StreamConfigError",
     "StreamDefinition",
     "StreamError",
     "StreamExpiredError",
     "StreamFencedError",
     "StreamHandle",
     "StreamNotFoundError",
     "StreamOffsetError",
+    "StreamPlugin",
     "StreamStorageError",
-    "StreamTransportMissingError",
     "StreamUnsupportedError",
     "Superseded",
     "SyncActivityStream",

temporalio/contrib/streams/_access.py

@@ -5,16 +5,18 @@
 from collections.abc import AsyncGenerator, Awaitable, Callable, Sequence
 from dataclasses import dataclass
 from datetime import timedelta
-from typing import Any, Generic, TypeVar
+from typing import Any, TypeVar
 
-import temporalio.api.streamservice.v1 as stream_pb
 import temporalio.converter
 from temporalio.contrib.streams import _convert, _reader
 from temporalio.contrib.streams._backend import StreamBackend
 from temporalio.contrib.streams._defn import TopicDefinition
-from temporalio.contrib.streams._handle import StreamHandle
+from temporalio.contrib.streams._handle import StreamHandle, TopicBase
+from temporalio.contrib.streams._items import End
 from temporalio.contrib.streams._reader import Item
 from temporalio.contrib.streams._writer import StreamWriter
+from temporalio.contrib.streams.system import client as service_client
+from temporalio.contrib.streams.system import models
 
 T = TypeVar("T")
 
@@ -41,7 +43,7 @@ def read(
         topics: list[str] | None,
         prefixes: Sequence[str],
         types: dict[str, Any],
-        from_offset: int | None,
+        from_offset: int | End | None,
         wait_new_records: bool,
         create_wait: timedelta | None,
     ) -> AsyncGenerator[Item, None]:
@@ -61,12 +63,12 @@ async def _read(
         topics: list[str] | None,
         prefixes: Sequence[str],
         types: dict[str, Any],
-        from_offset: int | None,
+        from_offset: int | End | None,
         wait_new_records: bool,
         create_wait: timedelta | None,
     ) -> AsyncGenerator[Item, None]:
         items = _reader.read(
-            self.backend.service(self.namespace),
+            self.backend.service(self.namespace, _convert.raw(self.converter)),
             self.converter,
             await self.stream_id(),
             topics=topics,
@@ -85,13 +87,10 @@ async def get_offset(self) -> int:
         It is 0 for a Stream with no Record, and ``from_offset=0`` reads from the start.
         Offsets are sparse and above zero, so 0 is never a Record's offset.
         """
-        response = await self.backend.call_async(
-            "DescribeStream",
-            stream_pb.DescribeStreamRequest(
-                namespace=self.namespace,
-                stream=stream_pb.StreamRef(id=await self.stream_id()),
-            ),
-            stream_pb.DescribeStreamResponse,
+        response = await self.backend.service(self.namespace).describe_stream(
+            models.DescribeStreamRequest(
+                stream=models.StreamRef(id=await self.stream_id())
+            )
         )
         # Offsets start above zero, so zero means the Stream has no Record.
         return response.last_offset + 1 if response.last_offset else 0
@@ -100,20 +99,14 @@ async def close(self, result: Any | None, failure: BaseException | None) -> None
         """Write this Producer's Records, then close the Stream."""
         writer = self.writer()
         await writer.close()
-        request = stream_pb.CloseStreamRequest(
-            namespace=self.namespace,
-            stream=stream_pb.StreamRef(id=await self.stream_id()),
-            producer=writer.producer,
+        stream = service_client.Stream(
+            self.backend.service(self.namespace, self.converter),
+            models.StreamRef(id=await self.stream_id()),
         )
-        if result is not None:
-            request.result.CopyFrom(
-                (await _convert.encode(self.converter, result)).payload
-            )
-        if failure is not None:
-            await self.converter.encode_failure(failure, request.failure)
-        await self.backend.call_async(
-            "CloseStream", request, stream_pb.CloseStreamResponse
+        producer = models._ProducerTransferTypeConverter().from_transfer_type(
+            writer.producer, models.Producer
         )
+        await stream.close(producer=producer, result=result, failure=failure)
 
 
 def read_spec(
@@ -140,34 +133,21 @@ def read_spec(
     return names or None, types
 
 
-class _AsyncTopic(Generic[T]):
+class _AsyncTopic(TopicBase[T]):
     # A Topic of an Activity or client handle. Each publish writes before it returns.
-    def __init__(self, stream: Any, name: str, type: Any) -> None:
-        self._stream = stream
-        self._name = name
-        self._type = type
-
-    @property
-    def name(self) -> str:
-        """Name of the Topic."""
-        return self._name
-
-    @property
-    def type(self) -> Any:
-        """Type that a read decodes the Topic's values with."""
-        return self._type
 
     async def publish(self, *values: T) -> int:
         """Write one Record for each value, in order, and return the last Record's offset.
 
         A list is one value.
         """
+        await self._stream._ready()
         return await self._stream._access().publish(self._name, values)
 
     def subscribe(
         self,
         *,
-        from_offset: int | None = None,
+        from_offset: int | End | None = None,
         wait_new_records: bool = True,
         create_wait: timedelta | None = None,
     ) -> AsyncGenerator[Item, None]:
@@ -180,21 +160,5 @@ def subscribe(
         )
 
 
-class SyncTopicBase(Generic[T]):
+class SyncTopicBase(TopicBase[T]):
     """Base of a Topic whose calls block, for sync Activities."""
-
-    def __init__(self, stream: Any, name: str, type: Any) -> None:
-        """Create a Topic. Use the Stream's ``topic`` instead."""
-        self._stream = stream
-        self._name = name
-        self._type = type
-
-    @property
-    def name(self) -> str:
-        """Name of the Topic."""
-        return self._name
-
-    @property
-    def type(self) -> Any:
-        """Type that a read decodes the Topic's values with."""
-        return self._type