[FROZEN][Jetty] Read a stream in Workflow code and cut a busy reader's task.

temporalio/contrib/streams/__init__.py

@@ -110,6 +110,13 @@
     track_supersession,
 )
 from temporalio.contrib.streams._topic import DEFAULT_TOPIC, StreamTopic, topic
+from temporalio.contrib.streams._workflow import (
+    StreamBatch,
+    StreamPosition,
+    TaskCutoff,
+    WorkflowStreamReader,
+    workflow_reader,
+)
 from temporalio.contrib.streams._workflow_producer import (
     WorkflowStreamProducer,
     stream_producer,
@@ -128,6 +135,7 @@
     "LinkStateKind",
     "RecordKind",
     "StoreCapabilities",
+    "StreamBatch",
     "StreamClosedError",
     "StreamCursorError",
     "StreamError",
@@ -137,6 +145,7 @@
     "StreamNotFoundError",
     "StreamOutcomeUnknownError",
     "StreamOwnerKind",
+    "StreamPosition",
     "StreamProducer",
     "StreamProducerError",
     "StreamRecord",
@@ -151,10 +160,13 @@
     "StreamUnsupportedError",
     "Supersession",
     "SupersessionTracker",
+    "TaskCutoff",
+    "WorkflowStreamReader",
     "WorkflowStreamProducer",
     "activity_handle",
     "stream_handle",
     "stream_producer",
     "topic",
     "track_supersession",
+    "workflow_reader",
 ]

temporalio/contrib/streams/_plugin.py

@@ -124,7 +124,7 @@ async def connect_service_client(
         return service_client
 
     def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
-        """Check that the Worker's client carries this store.
+        """Check that the Worker's client carries this store, and add the built-in Activities.
 
         Also adds the interceptor that stamps each Local Activity call, so a
         Local Activity can publish.
@@ -148,7 +148,14 @@ def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
             )
         # Imported here, since the Activity reaches the store through this module.
         from temporalio.contrib.streams._local_activity import StreamsInterceptor
-
+        from temporalio.contrib.streams._workflow import read_activity
+
+        activities = list(config.get("activities") or [])
+        # The same plugin on the client and the Worker configures the Worker twice.
+        for builtin in (read_activity,):
+            if builtin not in activities:
+                activities.append(builtin)
+        config["activities"] = activities
         interceptors = list(config.get("interceptors") or [])
         # Two would number each Local Activity call twice.
         if not any(isinstance(i, StreamsInterceptor) for i in interceptors):