diff --git a/internal/api/api.go b/internal/api/api.go index b62b143..efd1e71 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -61,5 +61,6 @@ func (manager *ApiManagerCtx) Mount(r chi.Router) { r.Post("/recreate", manager.roomRecreate) }) + r.Get("/events", manager.events) r.Get("/docker-compose.yaml", manager.dockerCompose) } diff --git a/internal/api/rooms.go b/internal/api/rooms.go index 612e101..60aaa44 100644 --- a/internal/api/rooms.go +++ b/internal/api/rooms.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "net/http" "strconv" @@ -250,6 +251,38 @@ func (manager *ApiManagerCtx) roomGenericAction(action func(ctx context.Context, } } +func (manager *ApiManagerCtx) events(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + + 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 + } + + fmt.Fprintf(w, "data: %s\n\n", jsonData) + flusher.Flush() + } + } +} + func (manager *ApiManagerCtx) dockerCompose(w http.ResponseWriter, r *http.Request) { response, err := manager.rooms.ExportAsDockerCompose(r.Context()) if err != nil { diff --git a/internal/room/events.go b/internal/room/events.go new file mode 100644 index 0000000..e01bd08 --- /dev/null +++ b/internal/room/events.go @@ -0,0 +1,112 @@ +package room + +import ( + "context" + "fmt" + + "github.com/m1k1o/neko-rooms/internal/types" + + dockerTypes "github.com/docker/docker/api/types" + "github.com/docker/docker/api/types/filters" +) + +func (manager *RoomManagerCtx) EventsLoopStart() { + ctx, cancel := context.WithCancel(context.Background()) + manager.eventsLoopCtx = ctx + manager.eventsLoopCancel = cancel + + msgs, errs := manager.client.Events(ctx, dockerTypes.EventsOptions{ + Filters: filters.NewArgs( + filters.Arg("type", "container"), + filters.Arg("label", fmt.Sprintf("m1k1o.neko_rooms.instance=%s", manager.config.InstanceName)), + filters.Arg("event", "create"), + filters.Arg("event", "start"), + //filters.Arg("event", "health_status"), + filters.Arg("event", "stop"), + filters.Arg("event", "destroy"), + ), + }) + + manager.eventsWg.Add(1) + go func() { + defer manager.eventsWg.Done() + + for { + select { + case err, ok := <-errs: + if !ok { + manager.logger.Fatal().Msg("docker event error channel closed") + return + } + + manager.logger.Err(err).Msg("got docker event error") + return + case msg, ok := <-msgs: + if !ok { + manager.logger.Fatal().Msg("docker event message channel closed") + return + } + + e := types.RoomEvent{ + ID: msg.Actor.ID[:12], + Action: msg.Action, + Labels: msg.Actor.Attributes, + } + + manager.logger.Info(). + Str("id", e.ID). + Str("action", e.Action). + Msg("got docker event") + + // broadcast event + manager.eventsMu.Lock() + for _, listener := range manager.eventsListeners { + listener <- e + } + manager.eventsMu.Unlock() + } + } + }() +} + +func (manager *RoomManagerCtx) EventsLoopStop() error { + manager.eventsLoopCancel() + manager.eventsWg.Wait() + return nil +} + +func (manager *RoomManagerCtx) Events(ctx context.Context) (<-chan types.RoomEvent, <-chan error) { + messages := make(chan types.RoomEvent) + errs := make(chan error, 1) + + // add listener + manager.eventsMu.Lock() + manager.eventsListeners = append(manager.eventsListeners, messages) + manager.eventsMu.Unlock() + + manager.eventsWg.Add(1) + go func() { + defer manager.eventsWg.Done() + defer close(errs) + + select { + case <-ctx.Done(): + errs <- ctx.Err() + case <-manager.eventsLoopCtx.Done(): + errs <- fmt.Errorf("room manager shutdown") + return + } + + // remove listener + manager.eventsMu.Lock() + for i, listener := range manager.eventsListeners { + if listener == messages { + manager.eventsListeners = append(manager.eventsListeners[:i], manager.eventsListeners[i+1:]...) + break + } + } + manager.eventsMu.Unlock() + }() + + return messages, errs +} diff --git a/internal/room/manager.go b/internal/room/manager.go index 18eed7f..34d6569 100644 --- a/internal/room/manager.go +++ b/internal/room/manager.go @@ -12,6 +12,7 @@ import ( "path/filepath" "strconv" "strings" + "sync" "time" "github.com/docker/cli/opts" @@ -55,6 +56,12 @@ type RoomManagerCtx struct { logger zerolog.Logger config *config.Room client *dockerClient.Client + + eventsMu sync.Mutex + eventsWg sync.WaitGroup + eventsListeners []chan types.RoomEvent + eventsLoopCtx context.Context + eventsLoopCancel context.CancelFunc } func (manager *RoomManagerCtx) Config() types.RoomsConfig { diff --git a/internal/types/room.go b/internal/types/room.go index fb9f103..2365b9f 100644 --- a/internal/types/room.go +++ b/internal/types/room.go @@ -138,6 +138,12 @@ type RoomMember struct { Muted bool `json:"muted"` } +type RoomEvent struct { + ID string `json:"id"` + Action string `json:"action"` + Labels map[string]string `json:"-"` +} + var ErrRoomNotFound = fmt.Errorf("room not found") type RoomManager interface { @@ -155,4 +161,8 @@ type RoomManager interface { Start(ctx context.Context, id string) error Stop(ctx context.Context, id string) error Restart(ctx context.Context, id string) error + + EventsLoopStart() + EventsLoopStop() error + Events(ctx context.Context) (<-chan RoomEvent, <-chan error) } diff --git a/neko.go b/neko.go index f4fe711..75fca59 100644 --- a/neko.go +++ b/neko.go @@ -131,6 +131,7 @@ func (main *MainCtx) Start() { client, main.Configs.Room, ) + main.roomManager.EventsLoopStart() main.pullManager = pull.New( client, @@ -169,6 +170,9 @@ func (main *MainCtx) Shutdown() { err = main.pullManager.Shutdown() main.logger.Err(err).Msg("pull manager shutdown") + + err = main.roomManager.EventsLoopStop() + main.logger.Err(err).Msg("room events loop shutdown") } func (main *MainCtx) ServeCommand(cmd *cobra.Command, args []string) {