[Brook] Added Topic prefix filters to Stream reads.

scripts/_proto/temporal/api/streamservice/v1/stream_service.proto

@@ -179,6 +179,9 @@ message ReadRecordsRequest {
     // End the read with an IDLE Record when no new Record comes for this long. Empty never
     // ends on idle. The facade counts it across reads, since one read waits at most 60 s.
     google.protobuf.Duration idle_timeout = 9;
+    // Also read the Topics that start with one of these. A Record passes when its Topic is in
+    // topics or starts with a prefix. Both empty reads every Topic.
+    repeated string topic_prefixes = 10;
 }
 
 message ReadRecordsResponse {

scripts/_wit/stream-service.wit

@@ -243,8 +243,10 @@ interface stream-service {
     from-offset: option<s64>,
     /// @nexus.doc "Where to start when from-offset is empty. Unspecified is the beginning."
     start: option<read-start>,
-    /// @nexus.doc "Empty reads every Topic."
+    /// @nexus.doc "Topics to read, by exact match. With topic-prefixes, empty reads every Topic."
     topics: list<string>,
+    /// @nexus.doc "Also read the Topics that start with one of these."
+    topic-prefixes: list<string>,
     /// @nexus.doc "At most this many Records. 1 to 1000."
     max-records: option<s32>,
     /// @nexus.doc "How long to wait for a new Record before an empty answer. At most 60 s."

temporalio/api/streamservice/v1/stream_service_pb2.pyi

@@ -730,6 +730,7 @@ class ReadRecordsRequest(google.protobuf.message.Message):
     WAIT_FIELD_NUMBER: builtins.int
     CREATE_WAIT_FIELD_NUMBER: builtins.int
     IDLE_TIMEOUT_FIELD_NUMBER: builtins.int
+    TOPIC_PREFIXES_FIELD_NUMBER: builtins.int
     namespace: builtins.str
     @property
     def stream(self) -> global___StreamRef: ...
@@ -754,6 +755,13 @@ class ReadRecordsRequest(google.protobuf.message.Message):
         """End the read with an IDLE Record when no new Record comes for this long. Empty never
         ends on idle. The facade counts it across reads, since one read waits at most 60 s.
         """
+    @property
+    def topic_prefixes(
+        self,
+    ) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]:
+        """Also read the Topics that start with one of these. A Record passes when its Topic is in
+        topics or starts with a prefix. Both empty reads every Topic.
+        """
     def __init__(
         self,
         *,
@@ -766,6 +774,7 @@ class ReadRecordsRequest(google.protobuf.message.Message):
         wait: google.protobuf.duration_pb2.Duration | None = ...,
         create_wait: google.protobuf.duration_pb2.Duration | None = ...,
         idle_timeout: google.protobuf.duration_pb2.Duration | None = ...,
+        topic_prefixes: collections.abc.Iterable[builtins.str] | None = ...,
     ) -> None: ...
     def HasField(
         self,
@@ -797,6 +806,8 @@ class ReadRecordsRequest(google.protobuf.message.Message):
             b"start",
             "stream",
             b"stream",
+            "topic_prefixes",
+            b"topic_prefixes",
             "topics",
             b"topics",
             "wait",

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 = "0981f62b1454e6501e3173d2115d94cb25eb9036"
+rev = "1c8c01f5ce87ca4ebb3fba155430920ff0a95c5c"
 default-features = false
 optional = true
 

temporalio/contrib/streams/_outside.py

@@ -99,6 +99,7 @@ def subscribe(
         self,
         *,
         topics: Sequence[Topic[Any]] | None = None,
+        topic_prefixes: Sequence[str] = (),
         from_offset: int | None = None,
         wait_new_records: bool = True,
         idle_timeout: timedelta | None = None,
@@ -111,7 +112,9 @@ def subscribe(
         a failed attempt, and it passes every Topic filter.
 
         Args:
-            topics: Read only these Topics. Defaults to every Topic.
+            topics: Read these Topics, by exact match.
+            topic_prefixes: Also read the Topics that start with one of these. With no
+                ``topics`` and no prefixes, the read gives every Topic.
             from_offset: Start at this offset, inclusive. Defaults to the first Record.
             wait_new_records: Wait for new Records at the end of the Stream.
             idle_timeout: End with an ``Idle`` item when no new Record comes for this long.
@@ -120,6 +123,7 @@ def subscribe(
         """
         return self._read(
             topics=[topic.name for topic in topics] if topics else None,
+            topic_prefixes=topic_prefixes,
             types={
                 **self._types,
                 **{topic.name: topic.type for topic in topics or []},
@@ -134,6 +138,7 @@ async def _read(
         self,
         *,
         topics: list[str] | None,
+        topic_prefixes: Sequence[str],
         types: dict[str, Any],
         default_type: Any,
         from_offset: int | None,
@@ -145,6 +150,7 @@ async def _read(
             self._converter,
             await self._stream_id(),
             topics=topics,
+            topic_prefixes=topic_prefixes,
             types=types,
             default_type=default_type,
             from_offset=from_offset,

temporalio/contrib/streams/_reader.py

@@ -30,6 +30,7 @@ async def read(
     stream_id: str,
     *,
     topics: Sequence[str] | None,
+    topic_prefixes: Sequence[str] = (),
     types: Mapping[str, Any],
     default_type: Any = None,
     from_offset: int | None,
@@ -61,6 +62,7 @@ async def read(
             from_offset=offset,
             start=None if offset is not None else ReadStart.READ_START_BEGINNING,
             topics=names,
+            topic_prefixes=list(topic_prefixes),
             idle_timeout=idle_timeout,
             create_wait=min(create_left, create_step) if create_left else None,
         )