diff --git a/internal/api/api.go b/internal/api/api.go index efd1e71..4040c7a 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -61,6 +61,11 @@ func (manager *ApiManagerCtx) Mount(r chi.Router) { r.Post("/recreate", manager.roomRecreate) }) - r.Get("/events", manager.events) r.Get("/docker-compose.yaml", manager.dockerCompose) + + // + // events + // + + r.Get("/events", manager.events) } diff --git a/internal/api/events.go b/internal/api/events.go new file mode 100644 index 0000000..e6a124b --- /dev/null +++ b/internal/api/events.go @@ -0,0 +1,67 @@ +package api + +import ( + "encoding/json" + "fmt" + "net/http" + "time" +) + +func (manager *ApiManagerCtx) events(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + + sse := r.URL.Query().Has("sse") + if sse { + w.Header().Set("Content-Type", "text/event-stream") + } else { + w.Header().Set("Content-Type", "application/json") + } + + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "Connection does not support streaming", http.StatusBadRequest) + return + } + + var ping <-chan time.Time + if !sse { + // dummy channel, never ping + ping = make(<-chan time.Time) + } else { + // ping every 1 minute + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + ping = ticker.C + } + + // listen for room events + events, errs := manager.rooms.Events(r.Context()) + for { + select { + case <-ping: + fmt.Fprintf(w, ": ping\n\n") + flusher.Flush() + case _, ok := <-errs: + if !ok { + manager.logger.Debug().Msg("sse channel closed") + } + return + case e := <-events: + jsonData, err := json.Marshal(e) + if err != nil { + manager.logger.Err(err).Msg("failed to marshal event") + continue + } + + if sse { + fmt.Fprintf(w, "event: rooms\n") + fmt.Fprintf(w, "data: %s\n\n", jsonData) + } else { + fmt.Fprintf(w, "rooms\t%s\n", jsonData) + } + + flusher.Flush() + } + } +} diff --git a/internal/api/rooms.go b/internal/api/rooms.go index 056e92d..612e101 100644 --- a/internal/api/rooms.go +++ b/internal/api/rooms.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "errors" - "fmt" "io" "net/http" "strconv" @@ -251,49 +250,6 @@ func (manager *ApiManagerCtx) roomGenericAction(action func(ctx context.Context, } } -func (manager *ApiManagerCtx) events(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Cache-Control", "no-cache") - w.Header().Set("Connection", "keep-alive") - - sse := r.URL.Query().Has("sse") - if sse { - w.Header().Set("Content-Type", "text/event-stream") - } else { - w.Header().Set("Content-Type", "application/json") - } - - flusher, ok := w.(http.Flusher) - if !ok { - http.Error(w, "Connection does not support streaming", http.StatusBadRequest) - return - } - - events, errs := manager.rooms.Events(r.Context()) - for { - select { - case _, ok := <-errs: - if !ok { - manager.logger.Debug().Msg("sse channel closed") - } - return - case e := <-events: - jsonData, err := json.Marshal(e) - if err != nil { - http.Error(w, err.Error(), 500) - return - } - - if sse { - fmt.Fprintf(w, "data: %s\n\n", jsonData) - } else { - fmt.Fprintf(w, "%s\n", jsonData) - } - - flusher.Flush() - } - } -} - func (manager *ApiManagerCtx) dockerCompose(w http.ResponseWriter, r *http.Request) { response, err := manager.rooms.ExportAsDockerCompose(r.Context()) if err != nil {