switch from websockets to SSE

3 files changed, 41 insertions, 52 deletions

diff --git a/main.go b/main.go
index 0ea10dd..53abc2f 100644
--- a/main.go
+++ b/main.go
@@ -14,8 +14,6 @@ import (
 	"muhq.space/go/wrms/llog"
 
 	"github.com/google/uuid"
-
-	"nhooyr.io/websocket"
 )
 
 var wrms *Wrms
@@ -188,30 +186,29 @@ func nextHandler(w http.ResponseWriter, r *http.Request) {
 	genericControlHandler(w, r, "next")
 }
 
-func wsEndpoint(w http.ResponseWriter, r *http.Request) {
+func eventsEndpoint(w http.ResponseWriter, r *http.Request) {
 	connId, err := getConnId(w, r)
 	if err != nil {
 		return
 	}
 
-	ws, err := websocket.Accept(w, r, nil)
-	if err != nil {
-		llog.Warning("Accepting the websocket failed with %s", err)
-		return
-	}
+	// prepare the header
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	w.Header().Set("Access-Control-Allow-Origin", "*")
 
 	ctx, cancel := context.WithCancel(r.Context())
-	ctx = ws.CloseRead(ctx)
+
+	conn := newConnection(connId, w, ctx, cancel)
 
 	defer func() {
 		llog.Info("cancel context of connection %v", connId)
 		cancel()
+		conn.Close()
 	}()
 
-	conn := newConnection(connId, ws, ctx, cancel)
-	defer conn.Close()
-
-	llog.Info("New websocket connection with id %v", connId)
+	llog.Info("New SSE connection with id %v", connId)
 	err = wrms.initConn(conn)
 	if err != nil {
 		return
@@ -230,7 +227,7 @@ func setupRoutes() {
 	http.HandleFunc("/delete", deleteHandler)
 	http.HandleFunc("/next", nextHandler)
 	http.HandleFunc("/playpause", playPauseHandler)
-	http.HandleFunc("/ws", wsEndpoint)
+	http.HandleFunc("/events", eventsEndpoint)
 }
 
 func main() {
diff --git a/web/client.html b/web/client.html
index 41e224f..a95bbb7 100644
--- a/web/client.html
+++ b/web/client.html
@@ -225,20 +225,10 @@
         playing.appendChild(songLabel);
       }
 
-      let socket_url = "ws://"+location.hostname + (location.port ? ":" + location.port: "") + "/ws";
-      let socket = new WebSocket(socket_url);
-      console.log("Attempting Connection...");
+      let events = new EventSource("/events");
 
-      socket.onopen = () => {
-        console.log("Successfully Connected");
-      };
-      
-      socket.onclose = event => {
-        console.log("Socket Closed Connection: ", event);
-      };
-
-      socket.onmessage = message => {
-        console.log("Socket Received Message: ", message);
+      events.onmessage = message => {
+        console.log("Received Event: ", message);
         const cmd = JSON.parse(message.data);
         switch (cmd.cmd) {
           case "add":
@@ -277,10 +267,6 @@
         }
       };
 
-      socket.onerror = error => {
-        console.log("Socket Error: ", error);
-      };
-
       function submitSearch() {
         resetSearch();
 
diff --git a/wrms.go b/wrms.go
index 1f74957..96bb43d 100644
--- a/wrms.go
+++ b/wrms.go
@@ -3,14 +3,14 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+	"net/http"
 	"sync"
 	"sync/atomic"
 
 	"muhq.space/go/wrms/llog"
 
 	"github.com/google/uuid"
-
-	"nhooyr.io/websocket"
 )
 
 type Song struct {
@@ -60,7 +60,8 @@ type Connection struct {
 	closing   atomic.Bool
 	refs      atomic.Int64
 	Events    chan Event
-	ws        *websocket.Conn
+	w         http.ResponseWriter
+	flusher   http.Flusher
 	ctx       context.Context
 	cancel    func()
 	nextEvent uint64
@@ -68,14 +69,20 @@ type Connection struct {
 
 const EVENT_BUFFER_SIZE = 3
 
-func newConnection(id uuid.UUID, ws *websocket.Conn, ctx context.Context, cncl func()) *Connection {
+func 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")
+	}
 
 	return &Connection{
-		Id:     id,
-		Events: make(chan Event, EVENT_BUFFER_SIZE),
-		ws:     ws,
-		ctx:    ctx,
-		cancel: cncl,
+		Id:      id,
+		Events:  make(chan Event, EVENT_BUFFER_SIZE),
+		w:       w,
+		flusher: flusher,
+		ctx:     ctx,
+		cancel:  cncl,
 	}
 }
 
@@ -105,7 +112,11 @@ func (conn *Connection) serve() {
 			delete(toSend, conn.nextEvent)
 			// We have not received the next event yet -> receive a new event
 		} else {
-			ev = <-conn.Events
+			select {
+			case ev = <-conn.Events:
+			case <-conn.ctx.Done():
+				return
+			}
 			// We received a future object -> store it and continue
 			if ev.Id > conn.nextEvent {
 				toSend[ev.Id] = &ev
@@ -121,12 +132,10 @@ func (conn *Connection) serve() {
 			return
 		}
 
-		llog.Debug("Sending ev %s to %s", string(data), conn.Id)
-		err = conn.ws.Write(conn.ctx, websocket.MessageText, data)
-		if err != nil {
-			llog.Warning("Sending the ev %s to %d failed with %s", string(data), conn.Id, err)
-			return
-		}
+		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()
 	}
 }
 
@@ -226,11 +235,8 @@ func (wrms *Wrms) initConn(conn *Connection) error {
 			return err
 		}
 
-		err = conn.ws.Write(conn.ctx, websocket.MessageText, data)
-		if err != nil {
-			llog.Warning("Sending the initial commands failed with %s", err)
-			return err
-		}
+		fmt.Fprintf(conn.w, "data: %s\n\n", string(data))
+		conn.flusher.Flush()
 	}
 
 	return nil