diff options
| -rw-r--r-- | connection.go | 142 | ||||
| -rw-r--r-- | song.go | 53 | ||||
| -rw-r--r-- | wrms.go | 177 |
3 files changed, 195 insertions, 177 deletions
diff --git a/connection.go b/connection.go new file mode 100644 index 0000000..4901fdd --- /dev/null +++ b/connection.go @@ -0,0 +1,142 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "sync/atomic" + + "muhq.space/go/wrms/llog" + + "github.com/google/uuid" +) + +type Connection struct { + wrms *Wrms + Id uuid.UUID + nr uint64 + closing atomic.Bool + refs atomic.Int64 + Events chan Event + w http.ResponseWriter + flusher http.Flusher + ctx context.Context + cancel func() + nextEvent uint64 +} + +const EVENT_BUFFER_SIZE = 3 + +func (wrms *Wrms) newConnection(id uuid.UUID, + w http.ResponseWriter, + ctx context.Context, + cncl func()) *Connection { + // prepare the flusher + flusher, ok := w.(http.Flusher) + if !ok { + llog.Fatal("Flusher type assertion failed") + } + + conn := &Connection{ + wrms: wrms, + Id: id, + nr: wrms.nextConnNr.Add(1), + Events: make(chan Event, EVENT_BUFFER_SIZE), + w: w, + flusher: flusher, + ctx: ctx, + cancel: cncl, + } + + wrms.addConn(conn) + return conn +} + +func (c *Connection) Send(ev Event) { + llog.DDebug("Sending %v to %v", ev, c.Id) + // Connection is closing -> not write to it + if c.closing.Load() { + return + } + // Register us as sender + c.refs.Add(1) + // Send the event + c.Events <- ev + // Deregister us as sender + c.refs.Add(-1) +} + +func (conn *Connection) _send(evs []interface{}) error { + for _, ev := range evs { + data, err := json.Marshal(ev) + if err != nil { + llog.Warning("Encoding the initial command %v failed with %s", ev, err) + return err + } + + sdata := string(data) + llog.Debug("Sending ev %s to %s", sdata, conn.Id) + fmt.Fprintf(conn.w, "data: %s\n\n", sdata) + conn.flusher.Flush() + } + + return nil +} + +func (conn *Connection) _sendEv(ev Event) error { + return conn._send([]interface{}{ev}) +} + +func (conn *Connection) serve() { + // Map to store future events to send them in order + toSend := make(map[uint64]*Event) + + for { + var ev Event + // We have the next event -> send it + if toSend[conn.nextEvent] != nil { + ev = *toSend[conn.nextEvent] + delete(toSend, conn.nextEvent) + // We have not received the next event yet -> receive a new event + } else { + llog.DDebug("%v: Awaiting event %d", conn.Id, conn.nextEvent) + select { + case ev = <-conn.Events: + case <-conn.ctx.Done(): + return + } + // We received a future object -> store it and continue + if ev.Id > conn.nextEvent { + llog.DDebug("%v: Received future event %d", conn.Id, ev.Id) + toSend[ev.Id] = &ev + continue + } + } + + // increment the expected event id + if ev.Id == conn.nextEvent { + conn.nextEvent++ + } + + conn._sendEv(ev) + } +} + +func (c *Connection) Close() { + llog.Info("Closing connection %s", c.Id) + // Remove the closing connection from the map + wrms.delConn(c) + // Announce that the connection is going to be closed + c.closing.Store(true) + + // Consume all events from the registered senders + for c.refs.Load() > 0 { + select { + case <-c.Events: + default: + } + } + + close(c.Events) +} @@ -0,0 +1,53 @@ +package main + +import ( + "encoding/json" + + "muhq.space/go/wrms/llog" + + "github.com/google/uuid" +) + +type Song struct { + Title string `json:"title"` + Artist string `json:"artist"` + Source string `json:"source"` + Uri string `json:"uri"` + Weight float64 `json:"weight"` + Album string `json:"album"` + Year int `json:"year"` + index int `json:"-"` // used by heap.Interface + Upvotes map[uuid.UUID]struct{} `json:"upvotes"` + Downvotes map[uuid.UUID]struct{} `json:"downvotes"` +} + +func NewSong(title, artist, source, uri string) *Song { + return &Song{ + Title: title, + Artist: artist, + Source: source, + Uri: uri, + Upvotes: map[uuid.UUID]struct{}{}, + Downvotes: map[uuid.UUID]struct{}{}, + } +} + +func NewDetailedSong(title, artist, source, uri, album string, year int) *Song { + s := NewSong(title, artist, source, uri) + s.Album = album + s.Year = year + return s +} + +func NewSongFromJson(data []byte) (*Song, error) { + var s Song + err := json.Unmarshal(data, &s) + if err != nil { + llog.Error("Failed to parse song data %s with %s", string(data), err) + return nil, err + } + + s.Upvotes = map[uuid.UUID]struct{}{} + s.Downvotes = map[uuid.UUID]struct{}{} + return &s, nil +} @@ -1,10 +1,6 @@ package main import ( - "context" - "encoding/json" - "fmt" - "net/http" "sync" "sync/atomic" @@ -13,50 +9,6 @@ import ( "github.com/google/uuid" ) -type Song struct { - Title string `json:"title"` - Artist string `json:"artist"` - Source string `json:"source"` - Uri string `json:"uri"` - Weight float64 `json:"weight"` - Album string `json:"album"` - Year int `json:"year"` - index int `json:"-"` // used by heap.Interface - Upvotes map[uuid.UUID]struct{} `json:"upvotes"` - Downvotes map[uuid.UUID]struct{} `json:"downvotes"` -} - -func NewSong(title, artist, source, uri string) *Song { - return &Song{ - Title: title, - Artist: artist, - Source: source, - Uri: uri, - Upvotes: map[uuid.UUID]struct{}{}, - Downvotes: map[uuid.UUID]struct{}{}, - } -} - -func NewDetailedSong(title, artist, source, uri, album string, year int) *Song { - s := NewSong(title, artist, source, uri) - s.Album = album - s.Year = year - return s -} - -func NewSongFromJson(data []byte) (*Song, error) { - var s Song - err := json.Unmarshal(data, &s) - if err != nil { - llog.Error("Failed to parse song data %s with %s", string(data), err) - return nil, err - } - - s.Upvotes = map[uuid.UUID]struct{}{} - s.Downvotes = map[uuid.UUID]struct{}{} - return &s, nil -} - type Event struct { Event string `json:"cmd"` Id uint64 `json:"id"` @@ -84,135 +36,6 @@ func (wrms *Wrms) newNotification(notification string) Event { return wrms.newEvent(notification, nil) } -type Connection struct { - wrms *Wrms - Id uuid.UUID - nr uint64 - closing atomic.Bool - refs atomic.Int64 - Events chan Event - w http.ResponseWriter - flusher http.Flusher - ctx context.Context - cancel func() - nextEvent uint64 -} - -const EVENT_BUFFER_SIZE = 3 - -func (wrms *Wrms) newConnection(id uuid.UUID, - w http.ResponseWriter, - ctx context.Context, - cncl func()) *Connection { - // prepare the flusher - flusher, ok := w.(http.Flusher) - if !ok { - llog.Fatal("Flusher type assertion failed") - } - - conn := &Connection{ - wrms: wrms, - Id: id, - nr: wrms.nextConnNr.Add(1), - Events: make(chan Event, EVENT_BUFFER_SIZE), - w: w, - flusher: flusher, - ctx: ctx, - cancel: cncl, - } - - wrms.addConn(conn) - return conn -} - -func (c *Connection) Send(ev Event) { - llog.DDebug("Sending %v to %v", ev, c.Id) - // Connection is closing -> not write to it - if c.closing.Load() { - return - } - // Register us as sender - c.refs.Add(1) - // Send the event - c.Events <- ev - // Deregister us as sender - c.refs.Add(-1) -} - -func (conn *Connection) _send(evs []interface{}) error { - for _, ev := range evs { - data, err := json.Marshal(ev) - if err != nil { - llog.Warning("Encoding the initial command %v failed with %s", ev, err) - return err - } - - sdata := string(data) - llog.Debug("Sending ev %s to %s", sdata, conn.Id) - fmt.Fprintf(conn.w, "data: %s\n\n", sdata) - conn.flusher.Flush() - } - - return nil -} - -func (conn *Connection) _sendEv(ev Event) error { - return conn._send([]interface{}{ev}) -} - -func (conn *Connection) serve() { - // Map to store future events to send them in order - toSend := make(map[uint64]*Event) - - for { - var ev Event - // We have the next event -> send it - if toSend[conn.nextEvent] != nil { - ev = *toSend[conn.nextEvent] - delete(toSend, conn.nextEvent) - // We have not received the next event yet -> receive a new event - } else { - llog.DDebug("%v: Awaiting event %d", conn.Id, conn.nextEvent) - select { - case ev = <-conn.Events: - case <-conn.ctx.Done(): - return - } - // We received a future object -> store it and continue - if ev.Id > conn.nextEvent { - llog.DDebug("%v: Received future event %d", conn.Id, ev.Id) - toSend[ev.Id] = &ev - continue - } - } - - // increment the expected event id - if ev.Id == conn.nextEvent { - conn.nextEvent++ - } - - conn._sendEv(ev) - } -} - -func (c *Connection) Close() { - llog.Info("Closing connection %s", c.Id) - // Remove the closing connection from the map - wrms.delConn(c) - // Announce that the connection is going to be closed - c.closing.Store(true) - - // Consume all events from the registered senders - for c.refs.Load() > 0 { - select { - case <-c.Events: - default: - } - } - - close(c.Events) -} - type Wrms struct { Connections sync.Map nextConnNr atomic.Uint64 |