package broker import ( "context" "sync" "time" "github.com/rs/zerolog/log" "github.com/seu-usuario/go-react-web-tail/internal/models" ) // Broker gerencia clientes SSE e broadcasting de logs type Broker struct { clients map[chan models.LogEntry]struct{} mu sync.RWMutex register chan chan models.LogEntry unregister chan chan models.LogEntry broadcast chan models.LogEntry ctx context.Context cancel context.CancelFunc // Métricas totalBroadcasts uint64 activeClients int } // New cria uma nova instância do Broker func New(ctx context.Context) *Broker { ctx, cancel := context.WithCancel(ctx) b := &Broker{ clients: make(map[chan models.LogEntry]struct{}), register: make(chan chan models.LogEntry), unregister: make(chan chan models.LogEntry), broadcast: make(chan models.LogEntry, 1000), ctx: ctx, cancel: cancel, } go b.run() return b } // run é o loop principal do broker func (b *Broker) run() { ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() log.Info().Msg("Broker started") for { select { case <-b.ctx.Done(): log.Info().Msg("Broker shutting down") b.closeAllClients() return case client := <-b.register: b.addClient(client) case client := <-b.unregister: b.removeClient(client) case entry := <-b.broadcast: b.broadcastToClients(entry) case <-ticker.C: b.logMetrics() } } } // Subscribe registra um novo cliente para receber logs func (b *Broker) Subscribe() chan models.LogEntry { ch := make(chan models.LogEntry, 100) b.register <- ch return ch } // Unsubscribe remove um cliente func (b *Broker) Unsubscribe(ch chan models.LogEntry) { b.unregister <- ch } // Broadcast envia uma entrada de log para todos os clientes func (b *Broker) Broadcast(entry models.LogEntry) { select { case b.broadcast <- entry: case <-b.ctx.Done(): default: log.Warn().Msg("Broadcast channel full, dropping message") } } // Shutdown encerra o broker graciosamente func (b *Broker) Shutdown() { log.Info().Msg("Broker shutdown requested") b.cancel() } // addClient adiciona um novo cliente func (b *Broker) addClient(client chan models.LogEntry) { b.mu.Lock() defer b.mu.Unlock() b.clients[client] = struct{}{} b.activeClients = len(b.clients) log.Info().Int("active_clients", b.activeClients).Msg("Client subscribed") } // removeClient remove um cliente func (b *Broker) removeClient(client chan models.LogEntry) { b.mu.Lock() defer b.mu.Unlock() if _, ok := b.clients[client]; ok { delete(b.clients, client) close(client) b.activeClients = len(b.clients) log.Info().Int("active_clients", b.activeClients).Msg("Client unsubscribed") } } // broadcastToClients envia uma entrada para todos os clientes conectados func (b *Broker) broadcastToClients(entry models.LogEntry) { b.mu.RLock() defer b.mu.RUnlock() b.totalBroadcasts++ for clientCh := range b.clients { select { case clientCh <- entry: // Enviado com sucesso default: // Cliente está lento, remover go b.Unsubscribe(clientCh) log.Warn().Msg("Removed slow client") } } } // closeAllClients fecha todos os canais de clientes func (b *Broker) closeAllClients() { b.mu.Lock() defer b.mu.Unlock() for client := range b.clients { close(client) } b.clients = make(map[chan models.LogEntry]struct{}) log.Info().Msg("All clients disconnected") } // logMetrics registra métricas do broker func (b *Broker) logMetrics() { b.mu.RLock() defer b.mu.RUnlock() log.Debug(). Int("active_clients", b.activeClients). Uint64("total_broadcasts", b.totalBroadcasts). Int("broadcast_buffer", len(b.broadcast)). Msg("Broker metrics") } // GetMetrics retorna métricas atuais do broker func (b *Broker) GetMetrics() (activeClients int, totalBroadcasts uint64) { b.mu.RLock() defer b.mu.RUnlock() return b.activeClients, b.totalBroadcasts }