Initial commit, pt. 63

cmd/streamcli/commands/commands.go

@@ -75,6 +75,16 @@ var (
 		Run:  variablesSet,
 	}
 
+	Config = &cobra.Command{
+		Use: "config",
+	}
+
+	ConfigGet = &cobra.Command{
+		Use:  "get",
+		Args: cobra.ExactArgs(0),
+		Run:  configGet,
+	}
+
 	LoggerLevel = logger.LevelWarning
 )
 
@@ -88,6 +98,9 @@ func init() {
 	Variables.AddCommand(VariablesGetHash)
 	Variables.AddCommand(VariablesSet)
 
+	Root.AddCommand(Config)
+	Config.AddCommand(ConfigGet)
+
 	Root.PersistentFlags().Var(&LoggerLevel, "log-level", "")
 	Root.PersistentFlags().String("remote-addr", "localhost:3594", "the path to the config file")
 	StreamSetup.PersistentFlags().String("title", "", "stream title")
@@ -212,3 +225,16 @@ func variablesSet(cmd *cobra.Command, args []string) {
 	err = streamD.SetVariable(ctx, consts.VarKey(variableKey), value)
 	assertNoError(ctx, err)
 }
+
+func configGet(cmd *cobra.Command, args []string) {
+	ctx := cmd.Context()
+
+	remoteAddr, err := cmd.Flags().GetString("remote-addr")
+	assertNoError(ctx, err)
+	streamD := client.New(remoteAddr)
+
+	cfg, err := streamD.GetConfig(ctx)
+	assertNoError(ctx, err)
+
+	cfg.WriteTo(os.Stdout)
+}

pkg/streamd/server/grpc.go

@@ -1224,7 +1224,7 @@ func (grpc *GRPCServer) SubscribeToConfigChanges(
 		srv,
 	)
 }
-func (grpc *GRPCServer) SubscribeToStreamChanges(
+func (grpc *GRPCServer) SubscribeToStreamsChanges(
 	req *streamd_grpc.SubscribeToStreamsChangesRequest,
 	srv streamd_grpc.StreamD_SubscribeToStreamsChangesServer,
 ) error {

pkg/streamd/streamd.go

@@ -1482,6 +1482,7 @@ func eventSubToChan[T any](
 	go func() {
 		<-ctx.Done()
 		d.EventBus.Unsubscribe(topic, callback)
+		d.EventBus.WaitAsync()
 		close(r)
 	}()
 	return r, nil

pkg/streampanel/panel.go

@@ -1405,7 +1405,8 @@ func (p *Panel) getUpdatedStatus_startStopStreamButton(ctx context.Context) {
 	p.streamMutex.Lock()
 	defer p.streamMutex.Unlock()
 
-	if isEnabled, _ := p.StreamD.IsBackendEnabled(ctx, obs.ID); isEnabled {
+	obsIsEnabled, _ := p.StreamD.IsBackendEnabled(ctx, obs.ID)
+	if obsIsEnabled {
 		obsStreamStatus, err := p.StreamD.GetStreamStatus(ctx, obs.ID)
 		if err != nil {
 			logger.Error(ctx, fmt.Errorf("unable to get stream status from OBS: %w", err))
@@ -1428,7 +1429,6 @@ func (p *Panel) getUpdatedStatus_startStopStreamButton(ctx context.Context) {
 			}
 			return
 		}
-
 	}
 
 	if p.updateTimerHandler != nil {
@@ -1446,7 +1446,9 @@ func (p *Panel) getUpdatedStatus_startStopStreamButton(ctx context.Context) {
 	}
 
 	if !ytIsEnabled || !p.youtubeCheck.Checked {
-		p.startStopButton.Enable()
+		if obsIsEnabled {
+			p.startStopButton.Enable()
+		}
 		return
 	}
 
@@ -1763,24 +1765,38 @@ func (p *Panel) initMainWindow(
 	p.profilesListWidget = profilesList
 
 	if _, ok := p.StreamD.(*client.Client); ok {
-		go func() {
-			p.getUpdatedStatus(ctx)
-
-			t := time.NewTicker(1 * time.Second)
-			for {
-				select {
-				case <-ctx.Done():
-					t.Stop()
-					return
-				case <-t.C:
-				}
-
-				p.getUpdatedStatus(ctx)
-			}
-		}()
+		p.subscribeUpdateControlPage(ctx)
 	}
 }
 
+func (p *Panel) subscribeUpdateControlPage(ctx context.Context) {
+	go func() {
+		logger.Debugf(ctx, "subscribe to streams and config changes")
+		defer logger.Debugf(ctx, "/subscribe to streams and config changes")
+
+		chStreams, err := p.StreamD.SubscribeToStreamsChanges(ctx)
+		if err != nil {
+			p.DisplayError(err)
+			//return
+		}
+		chConfigs, err := p.StreamD.SubscribeToConfigChanges(ctx)
+		if err != nil {
+			p.DisplayError(err)
+			//return
+		}
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-chStreams:
+			case <-chConfigs:
+			}
+			p.getUpdatedStatus(ctx)
+		}
+	}()
+}
+
 func (p *Panel) getSelectedProfile() Profile {
 	return getProfile(p.configCache, *p.selectedProfileName)
 }