Serialized, non-ordered handling of n messages

Author: dandavison

Diff for: signals_and_updates/serialized_handling_of_n_messages.py

@@ -0,0 +1,65 @@
+import asyncio
+import logging
+import random
+from typing import Optional
+
+from temporalio import common, workflow
+from temporalio.client import Client, WorkflowHandle
+from temporalio.worker import Worker
+
+Payload = str
+
+
+@workflow.defn
+class MessageProcessor:
+    def __init__(self) -> None:
+        self.queue = asyncio.Queue()
+
+    @workflow.run
+    async def run(self, serialized_queue_state: Optional[SerializedQueueState] = None):
+        # Initialize workflow state after CAN. Note that handler is sync, so it cannot wait for
+        # workflow initialization.
+        if serialized_queue_state:
+            self.queue.update_from_serialized_state(serialized_queue_state)
+        while True:
+            workflow.logger.info(f"waiting for msg {self.queue.head + 1}")
+            payload = await self.queue.next()
+            workflow.logger.info(payload)
+            if workflow.info().is_continue_as_new_suggested():
+                workflow.logger.info("CAN")
+                workflow.continue_as_new(args=[self.queue.serialize()])
+
+    # Note: sync handler
+    @workflow.update
+    def process_message(self, sequence_number: int, payload: Payload):
+        self.queue.add(payload, sequence_number)
+
+
+async def app(wf: WorkflowHandle):
+    for i in range(100):
+        print(f"sending update {i}")
+        await wf.execute_update(
+            MessageProcessor.process_message, args=[i, f"payload {i}"]
+        )
+
+
+async def main():
+    client = await Client.connect("localhost:7233")
+
+    async with Worker(
+        client,
+        task_queue="tq",
+        workflows=[MessageProcessor],
+    ):
+        wf = await client.start_workflow(
+            MessageProcessor.run,
+            id="wid",
+            task_queue="tq",
+            id_reuse_policy=common.WorkflowIDReusePolicy.TERMINATE_IF_RUNNING,
+        )
+        await asyncio.gather(app(wf), wf.result())
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    asyncio.run(main())