feat: Re-introduce watch page reload on live state change (#1894)

Author: joschahenningsen

api/worker_grpc.go

@@ -235,7 +235,6 @@ func (s server) NotifyStreamFinished(ctx context.Context, request *pb.StreamFini
 	if err != nil {
 		logger.Error("Can't set stream not live", "err", err)
 	}
-	NotifyViewersLiveState(uint(request.StreamID), false)
 
 	return &pb.Status{Ok: true}, nil
 }
@@ -600,7 +599,6 @@ func (s server) NotifyStreamStarted(ctx context.Context, request *pb.StreamStart
 		default:
 			s.StreamsDao.SaveCOMBURL(&stream, request.HlsUrl)
 		}
-		NotifyViewersLiveState(stream.Model.ID, true)
 		NotifyLiveUpdateCourseWentLive(stream.Model.ID)
 	}()
 

cmd/tumlive/main.go

@@ -145,6 +145,7 @@ func run(ctx context.Context) error {
 	opts := []runner_manager.Option{
 		runner_manager.WithMassStorage(tools.Cfg.Paths.Mass),
 		runner_manager.WithCamService(camService),
+		runner_manager.WithLiveStateNotifier(api.NotifyViewersLiveState),
 	}
 	var subtitleClient pb.SubtitleGeneratorClient
 	if tools.Cfg.VoiceService.Host != "" {

pkg/runner_manager/manager.go

@@ -53,12 +53,20 @@ type Manager struct {
 	camsHandledLock sync.Mutex
 
 	lightLock sync.Mutex
+
+	liveStateNotifier LiveStateNotifier
+	liveNotified      map[uint]bool
+	liveNotifiedLock  sync.Mutex
 }
 
 type CamService interface {
 	For(address string, cameraType model.CameraType) (camera.Cam, error)
 }
 
+// LiveStateNotifier is called whenever a stream transitions into or out of the live state,
+// so that other parts of the application (e.g. the websocket hub) can react to it.
+type LiveStateNotifier func(streamID uint, live bool)
+
 // New returns a new instance of Manager with the given Options
 func New(dao dao.DaoWrapper, opts ...Option) *Manager {
 	m := Manager{
@@ -69,6 +77,7 @@ func New(dao dao.DaoWrapper, opts ...Option) *Manager {
 		})).With("service", "runner_manager"),
 		streamStartLock: sync.Mutex{},
 		camsHandled:     make(map[uint]bool),
+		liveNotified:    make(map[uint]bool),
 	}
 	m.applyOpts(opts)
 	return &m
@@ -157,6 +166,14 @@ func WithSubtitleClient(client pb.SubtitleGeneratorClient, auth string) Option {
 	}
 }
 
+// WithLiveStateNotifier registers a callback invoked when a stream starts or stops being live,
+// e.g. to notify viewers via websocket.
+func WithLiveStateNotifier(notifier LiveStateNotifier) Option {
+	return func(m *Manager) {
+		m.liveStateNotifier = notifier
+	}
+}
+
 func (m *Manager) applyOpts(opts []Option) {
 	for _, opt := range opts {
 		opt(m)
@@ -274,6 +291,8 @@ func (m *Manager) streamStarted(ctx context.Context, req *protobuf.StreamStartNo
 	}
 	m.streamStartLock.Unlock()
 
+	m.notifyLiveState(stream.ID, true)
+
 	err = m.handleCamera(ctx, stream)
 	if err != nil {
 		log.Error("failed to handle camera", "stream", stream.ID, "err", err)
@@ -287,6 +306,28 @@ func (m *Manager) streamStarted(ctx context.Context, req *protobuf.StreamStartNo
 	return nil
 }
 
+// notifyLiveState invokes the configured LiveStateNotifier, making sure viewers are only notified
+// once per stream when going live, even though streamStarted is called once per stream version.
+func (m *Manager) notifyLiveState(streamID uint, live bool) {
+	if m.liveStateNotifier == nil {
+		return
+	}
+	if live {
+		m.liveNotifiedLock.Lock()
+		alreadyNotified := m.liveNotified[streamID]
+		m.liveNotified[streamID] = true
+		m.liveNotifiedLock.Unlock()
+		if alreadyNotified {
+			return
+		}
+	} else {
+		m.liveNotifiedLock.Lock()
+		delete(m.liveNotified, streamID)
+		m.liveNotifiedLock.Unlock()
+	}
+	m.liveStateNotifier(streamID, live)
+}
+
 var errNotNoLectureSource = fmt.Errorf("no source configured for this lecture hall ip")
 
 func (m *Manager) requestStreamVersion(ctx context.Context, s model.Stream, client protobuf.RunnerServiceClient, lh model.LectureHall, version protobuf.StreamVersion) (*protobuf.StreamResponse, error) {
@@ -598,10 +639,12 @@ func (m *Manager) handleLightsOff(stream model.Stream) (err error) {
 
 func (m *Manager) streamEnded(ctx context.Context, notification *protobuf.StreamEndNotification) error {
 	m.logger.Debug("streamEnd", "payload", notification)
-	err := m.dao.StreamsDao.SetStreamNotLiveById(uint(notification.GetStream().GetId()))
+	streamID := uint(notification.GetStream().GetId())
+	err := m.dao.StreamsDao.SetStreamNotLiveById(streamID)
 	if err != nil {
 		return err
 	}
+	m.notifyLiveState(streamID, false)
 
 	stream, err := m.dao.StreamsDao.GetStreamByID(ctx, strconv.FormatUint(notification.GetStream().GetId(), 10))
 	if err != nil {

web/ts/components/video-information.ts

@@ -31,6 +31,8 @@ export function videoInformationContext(streamId: number): AlpineComponent {
                     this.handleViewersUpdate(data);
                 } else if ("description" in data) {
                     this.handleDescriptionUpdate(data);
+                } else if ("live" in data) {
+                    this.handleLiveState(data);
                 }
             };
             SocketConnections.ws.subscribe(handler);
@@ -44,6 +46,13 @@ export function videoInformationContext(streamId: number): AlpineComponent {
             this.less = upd.description.full.length > CUTOFFLENGTH;
             this.description = upd.description.full;
         },
+
+        handleLiveState(upd: { live: boolean }) {
+            if (upd.live) {
+                // stream just went live, reload the page to pick up the stream
+                window.location.reload();
+            }
+        },
     } as AlpineComponent;
 }