Initial commit

backend/internal/services/log_streamer.go

@@ -0,0 +1,81 @@
+package services
+
+import (
+	"bufio"
+	"io"
+	"log"
+	"sync"
+)
+
+type LogStreamer struct {
+	mu      sync.RWMutex
+	subs    map[string]map[string]chan string
+}
+
+func NewLogStreamer() *LogStreamer {
+	return &LogStreamer{
+		subs: make(map[string]map[string]chan string),
+	}
+}
+
+func (ls *LogStreamer) Subscribe(serverID, clientID string) chan string {
+	ls.mu.Lock()
+	defer ls.mu.Unlock()
+
+	if ls.subs[serverID] == nil {
+		ls.subs[serverID] = make(map[string]chan string)
+	}
+
+	ch := make(chan string, 256)
+	ls.subs[serverID][clientID] = ch
+	return ch
+}
+
+func (ls *LogStreamer) Unsubscribe(serverID, clientID string) {
+	ls.mu.Lock()
+	defer ls.mu.Unlock()
+
+	if ls.subs[serverID] != nil {
+		if ch, ok := ls.subs[serverID][clientID]; ok {
+			close(ch)
+			delete(ls.subs[serverID], clientID)
+		}
+		if len(ls.subs[serverID]) == 0 {
+			delete(ls.subs, serverID)
+		}
+	}
+}
+
+func (ls *LogStreamer) Stream(serverID string, reader io.Reader, closeMsg string) {
+	scanner := bufio.NewScanner(reader)
+	for scanner.Scan() {
+		line := scanner.Text()
+		ls.broadcast(serverID, line)
+	}
+	if err := scanner.Err(); err != nil {
+		log.Printf("log stream error for server %s: %v", serverID, err)
+	}
+	if closeMsg != "" {
+		ls.broadcast(serverID, closeMsg)
+	}
+}
+
+func (ls *LogStreamer) Broadcast(serverID, line string) {
+	ls.broadcast(serverID, line)
+}
+
+func (ls *LogStreamer) broadcast(serverID, line string) {
+	ls.mu.RLock()
+	defer ls.mu.RUnlock()
+
+	if ls.subs[serverID] == nil {
+		return
+	}
+
+	for _, ch := range ls.subs[serverID] {
+		select {
+		case ch <- line:
+		default:
+		}
+	}
+}