log stream client

Author: ra-sun-god

src/daemon.ts

@@ -154,7 +154,7 @@ export default class Daemon {
     return this.server;
   }
 
-  async handleStreamMessage(msg: DaemonMessage, streamController: ReadableStreamController<any>, signal: AbortSignal) {
+  async handleStreamMessage(msg: DaemonMessage, streamController: ReadableStreamDefaultController, signal: AbortSignal) {
 
     if (!this.initialized) {
       await this.initialize();

src/index.ts

@@ -166,6 +166,52 @@ class BM2CLI {
       return { type: "error", error: "Fetch Error", success: false };
     }
   }
+  
+  async getDaemonStream<T>(data: DaemonMessage, callback: (data: T | null) => void) {
+    try {
+      
+      await this.startDaemon();
+     
+       const res = await fetch("http://localhost/command", {
+         unix: DAEMON_SOCKET,
+         method: "POST",
+         headers: { "Content-Type": "application/json" },
+         body: JSON.stringify(data)
+       })
+      
+       if (!res.body) {
+        console.error("No stream received");
+        process.exit(1);
+      }
+       
+      const reader = res.body.getReader();
+      const decoder = new TextDecoder();
+      
+      
+      let buffer = "";
+    
+      while (true) {
+        
+        const { value, done } = await reader.read();
+        
+        if (done) break;
+    
+        buffer += decoder.decode(value, { stream: true });
+    
+        // split SSE messages
+        const parts = buffer.split("\n\n");
+        buffer = parts.pop()!;
+    
+        for (const part of parts) {
+          console.log("part===>", part)
+        }
+      }
+      
+    } catch (e: any) {
+      console.log("callDaemonCmd:", e, e.stack);
+      return { type: "error", error: "Fetch Error", success: false };
+    }
+  }
 
   // -------------------------------------------------------------------------
   // Ecosystem config loader
@@ -570,6 +616,7 @@ class BM2CLI {
   }
 
   async cmdLogs(args: string[]) {
+    
     let target: string | number = "all";
     let lines = 20;
     let follow = false;
@@ -593,10 +640,6 @@ class BM2CLI {
       i++;
     }
 
-    console.log("target===>", target);
-    console.log("lines===>", lines);
-    console.log("follow===>", follow);
-    
     const renderLogs = (logs: LogItem[]) => {
       for (let log of logs) {
         let line;
@@ -612,10 +655,18 @@ class BM2CLI {
 
     if (follow) {
 
-      await this.sendToDaemon({
+      const opts: DaemonMessage = {
         type: "streamLogs",
         data: { target },
-      });
+        mode: "stream"
+      }
+      
+      const callback = (data: any) => {
+        console.log("data===>", data)
+      }
+      
+      const res = await this.getDaemonStream(opts, callback);
+      
 
     } else {
 

src/log-manager.ts

@@ -179,11 +179,16 @@ export class LogManager {
     return sortedLogs
   }
 
-  async tailLog(filePath: string, streamController: ReadableStreamController<any>, signal: any): Promise<void> {
+  async tailLog(
+    name: string,
+    id: number,
+    streamController: ReadableStreamDefaultController,
+    signal: any
+  ): Promise<void> {
 
     let lastSize = (await Bun.file(filePath).exists()) ? Bun.file(filePath).size : 0;
 
-    const watcher = watch(filePath, async () => {
+    const watcher = setInterval(async () => {
 
       const f = Bun.file(filePath);
 
@@ -192,12 +197,18 @@ export class LogManager {
       const chunk = await f.slice(lastSize, f.size).text();
       lastSize = f.size;
 
-      chunk.split("\n").filter(Boolean).forEach(streamController.enqueue);
+      chunk
+        .split("\n")
+        .filter(Boolean)
+        .forEach(line => {
+          const data = `data: ${JSON.stringify(line)}\n\n`
+          streamController.enqueue(data)
+        });
 
-    });
+    }, 2000);
 
     signal?.addEventListener("abort", () => {
-      watcher.close();
+      clearInterval(watcher)
     });
   }
 

src/process-manager.ts

@@ -320,7 +320,6 @@ import type { ReadableStreamController } from "bun";
       return logs.map((log) => ({ name: c.name, id: c.id, ...log }))
      }))).flat();
 
-     console.log("results===>", results)
 
      let sortedResults = results
        .sort((a, b) => (a.ts || "").localeCompare(b.ts || ""))
@@ -329,15 +328,14 @@ import type { ReadableStreamController } from "bun";
      return sortedResults;
    }
 
-   async streamLogs(target: string | number, streamController: ReadableStreamController<any>, signal: AbortSignal) {
+   async streamLogs(target: string | number, streamController: ReadableStreamDefaultController, signal: AbortSignal) {
 
      const containers = this.resolveTarget(target);
      const lm = this.logManager;
 
-     await Promise.all(containers.map(async (c) => {
-       const logPaths = lm.getLogPaths(c.name, c.id);
-       return lm.tailLog(logPaths.outFile, streamController, signal);
-     }))
+     await Promise.all(containers.map(async (c) => (
+      lm.tailLog(c.name, c.id, streamController, signal)
+     )))
 
    }