Merge branch for streaming

platform/src/bridge.ts

@@ -14,7 +14,8 @@ export const run = async (
   scriptName: string,
   port: number, // needed for self-calling services in pythonland
   args: JSON,
-  onLog?: (str: string) => void
+  onLog?: (str: string) => void,
+  onEvent?: (type: string, payload: any /* string or json tbh */) => void
 ) => {
   return new Promise<JSON | null>(async (resolve, reject) => {
     const id = crypto.randomUUID();
@@ -77,14 +78,24 @@ export const run = async (
       crlfDelay: Infinity,
     });
     rl.on("line", (line) => {
-      // First divert the log line locally
-      console.log(line);
-
       // Then divert any logs from a logger object to the websocket
       if (/^(INFO|DEBUG|ERROR|WARN)\:/.test(line)) {
+        // Divert the log line locally
+        console.log(line);
         // TODO I'd love to break the log line up in to JSON actually
         // { source, level, message }
         onLog?.(line);
+      } else if (/^(EVENT)\:/.test(line)) {
+        // TODO does the event encoding need to be any more complex than this?
+        // Nice that it stays human readable
+        const [_prefix, type, ...payload] = line.split(":");
+        let processedPayload = payload.join(":");
+        try {
+          processedPayload = JSON.parse(processedPayload);
+        } catch (e) {
+          // No json, no problem
+        }
+        onEvent?.(type, processedPayload);
       }
     });
 

platform/src/middleware/services.ts

@@ -12,11 +12,13 @@ const callService = (
   m: ModuleDescription,
   port: number,
   payload?: any,
-  onLog?: (str: string) => void
+  onLog?: (str: string) => void,
+  onEvent?: (evt: string, payload: any) => void
 ) => {
   if (m.type === "py") {
-    return run(m.name, port, payload as any, onLog);
+    return run(m.name, port, payload as any, onLog, onEvent);
   } else {
+    // TODO add event handling to ts services
     return m.handler!(port, payload as any, onLog);
   }
 };
@@ -63,8 +65,15 @@ export default async (app: Elysia, port: number) => {
                   data: log,
                 });
               };
+              const onEvent = (type: string, payload: any) => {
+                ws.send({
+                  event: "event",
+                  type,
+                  data: payload,
+                });
+              };
 
-              callService(m, port, message.data as any, onLog).then(
+              callService(m, port, message.data as any, onLog, onEvent).then(
                 (result) => {
                   ws.send({
                     event: "complete",

services/echo/echo.py

@@ -1,8 +1,11 @@
 from .log import log
+from util import EventLogger
 
+evt = EventLogger()
 
 # Simple python service to echo requests back to the caller
 # Used in test
 def main(x):
     log(x)
+    evt.send_status("{ msg: "+ str(x) +" }")
     return x

services/util.py

@@ -110,10 +110,12 @@ def send_status(self, message: str):
 
         :param message: Status message to log
         """
-        if self.logger:
-            self.logger.info(f"EVENT:STATUS:{message}")
-        else:
-            logging.info(f"EVENT:STATUS:{message}")
+        # if self.logger:
+        #     self.logger.info(f"EVENT:STATUS:{message}")
+        # else:
+        #     logging.info(f"EVENT:STATUS:{message}")
+        ## Don't use a logger for this because we don't want to prepend the service name stuff
+        print(f"EVENT:STATUS:{message}")
 
     def send_chunk(self, text: str):
         """