third-party-mirrors/navidrome
1
0
Fork 0
mirror of https://github.com/navidrome/navidrome.git synced 2025-04-27 17:02:17 +03:00
Code Issues Packages Projects Releases Wiki Activity

Use channels for EventStream instead of diodes

Browse Source
This commit is contained in:
Deluan 2023-04-03 10:51:24 -04:00
parent fea2de8f90
commit b22d0366d5
1 changed files with 28 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
server/events/sse.go
View File

@ -12,7 +12,7 @@ import (
"github.com/navidrome/navidrome/consts" "github.com/navidrome/navidrome/consts"
"github.com/navidrome/navidrome/log" "github.com/navidrome/navidrome/log"
"github.com/navidrome/navidrome/model/request" "github.com/navidrome/navidrome/model/request"
"github.com/navidrome/navidrome/utils/diodes" "github.com/navidrome/navidrome/utils/pl"
"github.com/navidrome/navidrome/utils/singleton" "github.com/navidrome/navidrome/utils/singleton"
) )
@ -24,6 +24,7 @@ type Broker interface {
const ( const (
keepAliveFrequency = 15 * time.Second keepAliveFrequency = 15 * time.Second
writeTimeOut = 5 * time.Second writeTimeOut = 5 * time.Second
bufferSize = 1
) )
type ( type (
@ -41,7 +42,7 @@ type (
username string username string
userAgent string userAgent string
clientUniqueId string clientUniqueId string
diode *diodes.Diode[message] msgC chan message
} }
) )
@ -80,7 +81,7 @@ func GetBroker() Broker {
func (b *broker) SendMessage(ctx context.Context, evt Event) { func (b *broker) SendMessage(ctx context.Context, evt Event) {
msg := b.prepareMessage(ctx, evt) msg := b.prepareMessage(ctx, evt)
log.Trace("Broker received new event", "event", msg) log.Trace("Broker received new event", "type", msg.event, "data", msg.data)
b.publish <- msg b.publish <- msg
} }
@ -147,21 +148,18 @@ func (b *broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// Each connection registers its own message channel with the Broker's connections registry // Each connection registers its own message channel with the Broker's connections registry
c := b.subscribe(r) c := b.subscribe(r)
defer b.unsubscribe(c) defer b.unsubscribe(c)
log.Debug(ctx, "New broker client", "client", c.String()) log.Debug(ctx, "Started new EventStream connection", "client", c.String())
for { for event := range pl.ReadOrDone(ctx, c.msgC) {
event := c.diode.Next() log.Trace(ctx, "Sending event to client", "event", event, "client", c.String())
if event == nil { err := writeEvent(ctx, w, event, writeTimeOut)
log.Trace(ctx, "Client closed the EventStream connection", "client", c.String())
return
}
log.Trace(ctx, "Sending event to client", "event", *event, "client", c.String())
err := writeEvent(ctx, w, *event, writeTimeOut)
if err != nil { if err != nil {
log.Debug(ctx, "Error sending event to client. Closing connection", "event", *event, "client", c.String(), err) log.Debug(ctx, "Error sending event to client. Closing connection", "event", event, "client", c.String(), err)
return return
} }
} }
log.Trace(ctx, "Client EventStream connection closed", "client", c.String())
return
} }
func (b *broker) subscribe(r *http.Request) client { func (b *broker) subscribe(r *http.Request) client {
@ -175,9 +173,7 @@ func (b *broker) subscribe(r *http.Request) client {
userAgent: r.UserAgent(), userAgent: r.UserAgent(),
clientUniqueId: clientUniqueId, clientUniqueId: clientUniqueId,
} }
c.diode = diodes.New[message](ctx, 1024, diodes.AlertFunc(func(missed int) { c.msgC = make(chan message, bufferSize)
log.Debug("Dropped SSE events", "client", c.String(), "missed", missed)
}))
// Signal the broker that we have a new client // Signal the broker that we have a new client
b.subscribing <- c b.subscribing <- c
@ -220,18 +216,19 @@ func (b *broker) listen() {
// A new client has connected. // A new client has connected.
// Register their message channel // Register their message channel
clients[c] = struct{}{} clients[c] = struct{}{}
log.Debug("Client added to event broker", "numClients", len(clients), "newClient", c.String()) log.Debug("Client added to EventStream broker", "numActiveClients", len(clients), "newClient", c.String())
// Send a serverStart event to new client // Send a serverStart event to new client
msg := b.prepareMessage(context.Background(), msg := b.prepareMessage(context.Background(),
&ServerStart{StartTime: consts.ServerStart, Version: consts.Version}) &ServerStart{StartTime: consts.ServerStart, Version: consts.Version})
c.diode.Put(msg) sendOrDrop(c, msg)
case c := <-b.unsubscribing: case c := <-b.unsubscribing:
// A client has detached, and we want to // A client has detached, and we want to
// stop sending them messages. // stop sending them messages.
close(c.msgC)
delete(clients, c) delete(clients, c)
log.Debug("Removed client from event broker", "numClients", len(clients), "client", c.String()) log.Debug("Removed client from EventStream broker", "numActiveClients", len(clients), "client", c.String())
case msg := <-b.publish: case msg := <-b.publish:
msg.id = getNextEventId() msg.id = getNextEventId()
@ -241,7 +238,7 @@ func (b *broker) listen() {
for c := range clients { for c := range clients {
if b.shouldSend(msg, c) { if b.shouldSend(msg, c) {
log.Trace("Putting event on client's queue", "client", c.String(), "event", msg) log.Trace("Putting event on client's queue", "client", c.String(), "event", msg)
c.diode.Put(msg) sendOrDrop(c, msg)
} }
} }
@ -254,8 +251,18 @@ func (b *broker) listen() {
msg.id = getNextEventId() msg.id = getNextEventId()
for c := range clients { for c := range clients {
log.Trace("Putting a keepalive event on client's queue", "client", c.String(), "event", msg) log.Trace("Putting a keepalive event on client's queue", "client", c.String(), "event", msg)
c.diode.Put(msg) sendOrDrop(c, msg)
} }
} }
} }
} }
func sendOrDrop(client client, msg message) {
select {
case client.msgC <- msg:
default:
if log.CurrentLevel() >= log.LevelTrace {
log.Trace("Event dropped because client's channel is full", "event", msg, "client", client.String())
}
}
}