grail/gralias
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Chore: styles

Browse Source
This commit is contained in:
Grail Finder
2025-07-13 12:28:52 +03:00
parent 155aa1b2cb
commit ad44dc0642
4 changed files with 17 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
broker/sse.go
View File

@@ -91,7 +91,9 @@ func (broker *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// Listen for new notifications and redistribute them to clients
func (broker *Broker) Listen() {
slog.Info("Broker listener started")
for {
slog.Info("Broker waiting for event")
select {
case s := <-broker.newClients:
// A new client has connected.
@@ -104,16 +106,21 @@ func (broker *Broker) Listen() {
delete(broker.clients, s)
slog.Info("Client removed", "clients listening", len(broker.clients))
case event := <-broker.Notifier:
slog.Info("Received new event", "event", event.EventName, "payload", event.Payload)
// We got a new event from the outside!
// Send event to all connected clients
slog.Info("Broadcasting event to clients", "client_count", len(broker.clients))
for clientMessageChan := range broker.clients {
slog.Info("Sending event to client", "client", clientMessageChan)
select {
case clientMessageChan <- event:
slog.Info("Successfully sent event to client", "client", clientMessageChan)
case <-time.After(patience):
delete(broker.clients, clientMessageChan)
slog.Info("Client was removed", "clients listening", len(broker.clients))
slog.Warn("Client timed out, removed", "client", clientMessageChan, "clients listening", len(broker.clients))
}
}
slog.Info("Finished broadcasting event")
}
}
}