stream events

Author: adrianlyjak

src/app/workflow.py

@@ -1,8 +1,10 @@
 import asyncio
+import time
 
 from workflows import Context, Workflow, step
-from workflows.events import StartEvent, StopEvent
+from workflows.events import StartEvent, StopEvent, Event
 import logging
+from datetime import datetime
 
 logger = logging.getLogger(__name__)
 
@@ -11,15 +13,30 @@ class PingEvent(StartEvent):
     message: str
 
 
-class PongEvent(StopEvent):
+class PongEvent(Event):
     message: str
 
 
+class WorkflowCompletedEvent(StopEvent):
+    timestamp: str
+
+
 class DefaultWorkflow(Workflow):
     @step
-    async def start(self, event: PingEvent, context: Context) -> PongEvent:
+    async def start(self, event: PingEvent, context: Context) -> WorkflowCompletedEvent:
+        start = time.monotonic()
         logger.info(f"Received message: {event.message}")
-        return PongEvent(message=f"PONG for '{event.message}'")
+        for i in range(5):
+            logger.info(f"Processing message: {event.message} {i}")
+            elapsed = (time.monotonic() - start) * 1000
+            context.write_event_to_stream(
+                PongEvent(message=f"+{elapsed:.0f}ms PONG {i + 1}/5 ")
+            )
+            await asyncio.sleep(0.2)
+        return WorkflowCompletedEvent(
+            timestamp="workflow completed at "
+            + datetime.now().isoformat(timespec="seconds")
+        )
 
 
 workflow = DefaultWorkflow()

ui/src/index.css

@@ -83,3 +83,15 @@ body {
     filter: blur(28px) saturate(110%);
   }
 }
+
+/* Simple fade-in-up animation for streamed items */
+@keyframes fade-in-left {
+  from {
+    opacity: 0;
+    transform: translateX(36px);
+  }
+  to {
+    opacity: 1;
+    transform: translateX(0);
+  }
+}

ui/src/pages/Home.tsx

@@ -13,9 +13,9 @@ export default function Home() {
           llama-ui to call the workflow. Customize this app with your own
           workflow and UI.
         </div>
-        <div className="flex flex-row gap-4 items-center justify-center w-full">
+        <div className="flex flex-row gap-4 items-start justify-center w-full">
           {taskId ? (
-            <HandlerOuput handlerId={taskId} />
+            <HandlerOutput handlerId={taskId} />
           ) : (
             <Output>
               <span className="text-black/60 dark:text-white/60">
@@ -71,14 +71,35 @@ function RunButton({
   );
 }
 
-function HandlerOuput({ handlerId }: { handlerId: string }) {
+function HandlerOutput({ handlerId }: { handlerId: string }) {
+  // stream events and result from the workflow
   const taskData = useWorkflowTask(handlerId);
 
-  const result = taskData.events.find((event) =>
+  // read workflow events here
+  const pongs = taskData.events.filter((event) =>
     event.type.match(/PongEvent$/),
-  ) as { type: string; data: { message: string } } | undefined;
+  ) as { type: string; data: { message: string } }[];
+  const completed = taskData.events.find((event) =>
+    event.type.match(/WorkflowCompletedEvent$/),
+  ) as { type: string; data: { timestamp: string } } | undefined;
 
-  return <Output>{result ? result.data.message : "Running... "}</Output>;
+  return (
+    <div className="flex flex-col gap-4 w-full min-h-60">
+      <Output>{completed ? completed.data.timestamp : "Running... "}</Output>
+      {pongs.map((pong, index) => (
+        <span
+          className="text-black/60 dark:text-white/60 text-sm m-0"
+          key={pong.data.message}
+          style={{
+            animation: "fade-in-left 80ms ease-out both",
+            willChange: "opacity, transform",
+          }}
+        >
+          {pong.data.message}
+        </span>
+      ))}
+    </div>
+  );
 }
 
 function Output({ children }: { children: React.ReactNode }) {