feat: Implement initial Go backend and React frontend project structure for the gotail application.
This commit is contained in:
Luiz Costa committed
1 parent
dfe9b10bb8
commit
ef1784f8d4
189 files changed
+25220
-279
No files matched your search
@@ -0,0 +1,166 @@
|
||||
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
|
||||
}
|
||||
@@ -0,0 +1,202 @@
|
||||
package broker
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/seu-usuario/go-react-web-tail/internal/models"
|
||||
)
|
||||
|
||||
func TestBroker_SubscribeUnsubscribe(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
b := New(ctx)
|
||||
defer b.Shutdown()
|
||||
|
||||
// Subscribe
|
||||
ch := b.Subscribe()
|
||||
if ch == nil {
|
||||
t.Fatal("Subscribe returned nil channel")
|
||||
}
|
||||
|
||||
// Give broker time to process
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
activeClients, _ := b.GetMetrics()
|
||||
if activeClients != 1 {
|
||||
t.Errorf("Expected 1 active client, got %d", activeClients)
|
||||
}
|
||||
|
||||
// Unsubscribe
|
||||
b.Unsubscribe(ch)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
activeClients, _ = b.GetMetrics()
|
||||
if activeClients != 0 {
|
||||
t.Errorf("Expected 0 active clients after unsubscribe, got %d", activeClients)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBroker_Broadcast(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
b := New(ctx)
|
||||
defer b.Shutdown()
|
||||
|
||||
ch := b.Subscribe()
|
||||
defer b.Unsubscribe(ch)
|
||||
|
||||
entry := models.LogEntry{
|
||||
Filename: "test.log",
|
||||
Line: "Test message",
|
||||
}
|
||||
|
||||
// Broadcast
|
||||
b.Broadcast(entry)
|
||||
|
||||
// Receive
|
||||
select {
|
||||
case received := <-ch:
|
||||
if received.Filename != entry.Filename {
|
||||
t.Errorf("Expected filename %s, got %s", entry.Filename, received.Filename)
|
||||
}
|
||||
if received.Line != entry.Line {
|
||||
t.Errorf("Expected line %s, got %s", entry.Line, received.Line)
|
||||
}
|
||||
case <-time.After(1 * time.Second):
|
||||
t.Fatal("Timeout waiting for broadcast message")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBroker_MultipleClients(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
b := New(ctx)
|
||||
defer b.Shutdown()
|
||||
|
||||
// Subscribe multiple clients
|
||||
clients := make([]chan models.LogEntry, 3)
|
||||
for i := range clients {
|
||||
clients[i] = b.Subscribe()
|
||||
}
|
||||
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
activeClients, _ := b.GetMetrics()
|
||||
if activeClients != 3 {
|
||||
t.Errorf("Expected 3 active clients, got %d", activeClients)
|
||||
}
|
||||
|
||||
entry := models.LogEntry{
|
||||
Filename: "test.log",
|
||||
Line: "Broadcast to all",
|
||||
}
|
||||
|
||||
b.Broadcast(entry)
|
||||
|
||||
// All clients should receive
|
||||
for i, ch := range clients {
|
||||
select {
|
||||
case received := <-ch:
|
||||
if received.Line != entry.Line {
|
||||
t.Errorf("Client %d: expected line %s, got %s", i, entry.Line, received.Line)
|
||||
}
|
||||
case <-time.After(1 * time.Second):
|
||||
t.Errorf("Client %d: timeout waiting for message", i)
|
||||
}
|
||||
}
|
||||
|
||||
// Cleanup
|
||||
for _, ch := range clients {
|
||||
b.Unsubscribe(ch)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBroker_Shutdown(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
b := New(ctx)
|
||||
|
||||
ch := b.Subscribe()
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
b.Shutdown()
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
// Channel should be closed
|
||||
_, ok := <-ch
|
||||
if ok {
|
||||
t.Error("Expected channel to be closed after shutdown")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBroker_GetMetrics(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
b := New(ctx)
|
||||
defer b.Shutdown()
|
||||
|
||||
// Initial metrics
|
||||
activeClients, totalBroadcasts := b.GetMetrics()
|
||||
if activeClients != 0 {
|
||||
t.Errorf("Expected 0 active clients initially, got %d", activeClients)
|
||||
}
|
||||
if totalBroadcasts != 0 {
|
||||
t.Errorf("Expected 0 total broadcasts initially, got %d", totalBroadcasts)
|
||||
}
|
||||
|
||||
// Subscribe and broadcast
|
||||
ch := b.Subscribe()
|
||||
defer b.Unsubscribe(ch)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
b.Broadcast(models.LogEntry{
|
||||
Filename: "test.log",
|
||||
Line: "Test",
|
||||
})
|
||||
}
|
||||
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
activeClients, totalBroadcasts = b.GetMetrics()
|
||||
if activeClients != 1 {
|
||||
t.Errorf("Expected 1 active client, got %d", activeClients)
|
||||
}
|
||||
if totalBroadcasts != 5 {
|
||||
t.Errorf("Expected 5 total broadcasts, got %d", totalBroadcasts)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkBroker_Broadcast(b *testing.B) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
broker := New(ctx)
|
||||
defer broker.Shutdown()
|
||||
|
||||
ch := broker.Subscribe()
|
||||
defer broker.Unsubscribe(ch)
|
||||
|
||||
// Drain channel in background
|
||||
go func() {
|
||||
for range ch {
|
||||
}
|
||||
}()
|
||||
|
||||
entry := models.LogEntry{
|
||||
Filename: "test.log",
|
||||
Line: "Benchmark message",
|
||||
}
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
broker.Broadcast(entry)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user