Add support of Chat Notifications

go.mod

@@ -205,6 +205,7 @@ require (
 	github.com/dustin/go-humanize v1.0.1
 	github.com/ebitengine/oto/v3 v3.3.1
 	github.com/getsentry/sentry-go v0.28.1
+	github.com/go-andiamo/splitter v1.2.5
 	github.com/go-git/go-git/v5 v5.12.0
 	github.com/go-ng/xmath v0.0.0-20230704233441-028f5ea62335
 	github.com/go-yaml/yaml v2.1.0+incompatible

go.sum

@@ -227,6 +227,8 @@ github.com/gin-gonic/gin v1.10.0 h1:nTuyha1TYqgedzytsKYqna+DfLos46nTv2ygFy86HFU=
 github.com/gin-gonic/gin v1.10.0/go.mod h1:4PMNQiOhvDRa013RKVbsiNwoyezlm2rm0uX/T7kzp5Y=
 github.com/gliderlabs/ssh v0.3.7 h1:iV3Bqi942d9huXnzEF2Mt+CY9gLu8DNM4Obd+8bODRE=
 github.com/gliderlabs/ssh v0.3.7/go.mod h1:zpHEXBstFnQYtGnB8k8kQLol82umzn/2/snG7alWVD8=
+github.com/go-andiamo/splitter v1.2.5 h1:P3NovWMY2V14TJJSolXBvlOmGSZo3Uz+LtTl2bsV/eY=
+github.com/go-andiamo/splitter v1.2.5/go.mod h1:8WHU24t9hcMKU5FXDQb1hysSEC/GPuivIp0uKY1J8gw=
 github.com/go-errors/errors v1.4.2 h1:J6MZopCL4uSllY1OfXM374weqZFFItUbrImctkmUxIA=
 github.com/go-errors/errors v1.4.2/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og=
 github.com/go-git/gcfg v1.5.1-0.20230307220236-3a3c6141e376 h1:+zs/tPmkDkHx3U66DAb0lQFJrpS6731Oaa12ikc+DiI=

pkg/expression/eval.go

@@ -27,6 +27,10 @@ func Eval[T any](
 	if value == "" {
 		return result, nil
 	}
+
+	if v, ok := any(value).(T); ok {
+		return v, nil
+	}
 	_, err = fmt.Sscanf(value, "%v", &result)
 	if err != nil {
 		return result, fmt.Errorf("unable to scan value '%v' into %T: %w", value, result, err)

pkg/streampanel/chat.go

@@ -21,6 +21,7 @@ import (
 	"github.com/xaionaro-go/streamctl/pkg/streamcontrol/twitch"
 	"github.com/xaionaro-go/streamctl/pkg/streamcontrol/youtube"
 	"github.com/xaionaro-go/streamctl/pkg/streamd/api"
+	"github.com/xaionaro-go/streamctl/pkg/xsync"
 )
 
 type chatUI struct {
@@ -153,6 +154,26 @@ func (ui *chatUI) onReceiveMessage(
 		ui.List.Refresh()
 	})
 	observability.Go(ctx, func() {
+		notificationsEnabled := xsync.DoR1(ctx, &ui.Panel.configLocker, func() bool {
+			return ui.Panel.Config.Chat.NotificationsEnabled()
+		})
+		if !notificationsEnabled {
+			return
+		}
+		logger.Debugf(ctx, "SendNotification")
+		defer logger.Debugf(ctx, "/SendNotification")
+		ui.Panel.app.SendNotification(&fyne.Notification{
+			Title:   string(msg.Platform) + " chat message",
+			Content: msg.Username + ": " + msg.Message,
+		})
+	})
+	observability.Go(ctx, func() {
+		soundEnabled := xsync.DoR1(ctx, &ui.Panel.configLocker, func() bool {
+			return ui.Panel.Config.Chat.ReceiveMessageSoundAlarmEnabled()
+		})
+		if !soundEnabled {
+			return
+		}
 		concurrentCount := atomic.AddInt32(&ui.CurrentlyPlayingChatMessageSoundCount, 1)
 		defer atomic.AddInt32(&ui.CurrentlyPlayingChatMessageSoundCount, -1)
 		logger.Debugf(ctx, "PlayChatMessage (count: %d)", concurrentCount)
@@ -166,6 +187,18 @@ func (ui *chatUI) onReceiveMessage(
 			logger.Errorf(ctx, "unable to playback the chat message sound: %v", err)
 		}
 	})
+	observability.Go(ctx, func() {
+		commandTemplate := xsync.DoR1(ctx, &ui.Panel.configLocker, func() string {
+			return ui.Panel.Config.Chat.CommandOnReceiveMessage
+		})
+		if commandTemplate == "" {
+			return
+		}
+		logger.Debugf(ctx, "CommandOnReceiveMessage: <%s>", commandTemplate)
+		defer logger.Debugf(ctx, "/CommandOnReceiveMessage")
+
+		ui.Panel.execCommand(ctx, commandTemplate, msg)
+	})
 }
 
 func (ui *chatUI) listLength() int {

pkg/streampanel/config/config.go

@@ -27,12 +27,27 @@ type OAuthConfig struct {
 	} `yaml:"listen_ports"`
 }
 
+type ChatConfig struct {
+	CommandOnReceiveMessage        string `yaml:"command_on_receive_message,omitempty"`
+	EnableNotifications            *bool  `yaml:"enable_notifications,omitempty"`
+	EnableReceiveMessageSoundAlarm *bool  `yaml:"enable_receive_message_sound_alarm,omitempty"`
+}
+
+func (cfg ChatConfig) NotificationsEnabled() bool {
+	return cfg.EnableNotifications == nil || *cfg.EnableNotifications
+}
+
+func (cfg ChatConfig) ReceiveMessageSoundAlarmEnabled() bool {
+	return cfg.EnableReceiveMessageSoundAlarm == nil || *cfg.EnableReceiveMessageSoundAlarm
+}
+
 type Config struct {
 	RemoteStreamDAddr string           `yaml:"streamd_remote"`
 	BuiltinStreamD    streamd.Config   `yaml:"streamd_builtin"`
 	Screenshot        ScreenshotConfig `yaml:"screenshot"`
 	Browser           BrowserConfig    `yaml:"browser"`
 	OAuth             OAuthConfig      `yaml:"oauth"`
+	Chat              ChatConfig       `yaml:"chat"`
 }
 
 func DefaultConfig() Config {

pkg/streampanel/panel.go

@@ -23,6 +23,7 @@ import (
 	"fyne.io/fyne/v2/layout"
 	"fyne.io/fyne/v2/theme"
 	"fyne.io/fyne/v2/widget"
+	child_process_manager "github.com/AgustinSRG/go-child-process-manager"
 	"github.com/facebookincubator/go-belt"
 	"github.com/facebookincubator/go-belt/tool/experimental/errmon"
 	"github.com/facebookincubator/go-belt/tool/logger"
@@ -78,6 +79,7 @@ type Panel struct {
 
 	app                   fyne.App
 	Config                Config
+	configLocker          xsync.RWMutex
 	streamMutex           xsync.Mutex
 	updateTimerHandler    *updateTimerHandler
 	profilesOrder         []streamcontrol.ProfileName
@@ -1479,9 +1481,16 @@ func (p *Panel) openSettingsWindow(ctx context.Context) error {
 		return fmt.Errorf("unable to get config: %w", err)
 	}
 
+	return xsync.DoA2R1(ctx, &p.configLocker, p.openSettingsWindowNoLock, ctx, cfg)
+}
+
+func (p *Panel) openSettingsWindowNoLock(
+	ctx context.Context,
+	streamDCfg *streamdconfig.Config,
+) error {
 	{
 		var buf bytes.Buffer
-		_, err := cfg.WriteTo(&buf)
+		_, err := streamDCfg.WriteTo(&buf)
 		if err != nil {
 			logger.Warnf(ctx, "unable to serialize the config: %v", err)
 		} else {
@@ -1506,20 +1515,20 @@ func (p *Panel) openSettingsWindow(ctx context.Context) error {
 	w := p.app.NewWindow(AppName + ": Settings")
 	resizeWindow(w, fyne.NewSize(400, 900))
 
-	if obsCfg, ok := cfg.Backends[obs.ID]; ok {
+	if obsCfg, ok := streamDCfg.Backends[obs.ID]; ok {
 		logger.Debugf(ctx, "current OBS config: %#+v", obsCfg)
 	}
 
-	cmdBeforeStartStream, _ := cfg.Backends[obs.ID].GetCustomString(
+	cmdBeforeStartStream, _ := streamDCfg.Backends[obs.ID].GetCustomString(
 		config.CustomConfigKeyBeforeStreamStart,
 	)
-	cmdBeforeStopStream, _ := cfg.Backends[obs.ID].GetCustomString(
+	cmdBeforeStopStream, _ := streamDCfg.Backends[obs.ID].GetCustomString(
 		config.CustomConfigKeyBeforeStreamStop,
 	)
-	cmdAfterStartStream, _ := cfg.Backends[obs.ID].GetCustomString(
+	cmdAfterStartStream, _ := streamDCfg.Backends[obs.ID].GetCustomString(
 		config.CustomConfigKeyAfterStreamStart,
 	)
-	cmdAfterStopStream, _ := cfg.Backends[obs.ID].GetCustomString(
+	cmdAfterStopStream, _ := streamDCfg.Backends[obs.ID].GetCustomString(
 		config.CustomConfigKeyAfterStreamStop,
 	)
 
@@ -1532,18 +1541,36 @@ func (p *Panel) openSettingsWindow(ctx context.Context) error {
 	afterStopStreamCommandEntry := widget.NewEntry()
 	afterStopStreamCommandEntry.SetText(cmdAfterStopStream)
 
+	afterReceivedChatMessage := widget.NewEntry()
+	afterReceivedChatMessage.SetText(p.Config.Chat.CommandOnReceiveMessage)
+
+	enableChatNotifications := widget.NewCheck(
+		"Enable on-screen notifications for chat messages",
+		func(b bool) {},
+	)
+	enableChatNotifications.SetChecked(p.Config.Chat.NotificationsEnabled())
+	enableChatMessageSoundsAlerts := widget.NewCheck(
+		"Enable sound alerts for chat messages",
+		func(b bool) {},
+	)
+	enableChatMessageSoundsAlerts.SetChecked(p.Config.Chat.ReceiveMessageSoundAlarmEnabled())
+
 	oldScreenshoterEnabled := p.Config.Screenshot.Enabled != nil && *p.Config.Screenshot.Enabled
 
 	mpvPathEntry := widget.NewEntry()
-	mpvPathEntry.SetText(cfg.StreamServer.VideoPlayer.MPV.Path)
+	mpvPathEntry.SetText(streamDCfg.StreamServer.VideoPlayer.MPV.Path)
 	mpvPathEntry.OnChanged = func(s string) {
-		cfg.StreamServer.VideoPlayer.MPV.Path = s
+		streamDCfg.StreamServer.VideoPlayer.MPV.Path = s
 	}
 
 	cancelButton := widget.NewButtonWithIcon("Cancel", theme.CancelIcon(), func() {
 		w.Close()
 	})
 	saveButton := widget.NewButtonWithIcon("Save", theme.DocumentSaveIcon(), func() {
+		p.Config.Chat.CommandOnReceiveMessage = afterReceivedChatMessage.Text
+		p.Config.Chat.EnableNotifications = ptr(enableChatNotifications.Checked)
+		p.Config.Chat.EnableReceiveMessageSoundAlarm = ptr(enableChatMessageSoundsAlerts.Checked)
+
 		if err := p.SaveConfig(ctx); err != nil {
 			p.DisplayError(fmt.Errorf("unable to save the local config: %w", err))
 		} else {
@@ -1553,7 +1580,7 @@ func (p *Panel) openSettingsWindow(ctx context.Context) error {
 			}
 		}
 
-		obsCfg := cfg.Backends[obs.ID]
+		obsCfg := streamDCfg.Backends[obs.ID]
 		obsCfg.SetCustomString(
 			config.CustomConfigKeyBeforeStreamStart, beforeStartStreamCommandEntry.Text)
 		obsCfg.SetCustomString(
@@ -1562,9 +1589,9 @@ func (p *Panel) openSettingsWindow(ctx context.Context) error {
 			config.CustomConfigKeyAfterStreamStart, afterStartStreamCommandEntry.Text)
 		obsCfg.SetCustomString(
 			config.CustomConfigKeyAfterStreamStop, afterStopStreamCommandEntry.Text)
-		cfg.Backends[obs.ID] = obsCfg
+		streamDCfg.Backends[obs.ID] = obsCfg
 
-		if err := p.SetStreamDConfig(ctx, cfg); err != nil {
+		if err := p.SetStreamDConfig(ctx, streamDCfg); err != nil {
 			p.DisplayError(fmt.Errorf("unable to update the remote config: %w", err))
 		} else {
 			if err := p.StreamD.SaveConfig(ctx); err != nil {
@@ -1667,14 +1694,14 @@ func (p *Panel) openSettingsWindow(ctx context.Context) error {
 			widget.NewRichTextFromMarkdown(`# Streaming platforms`),
 			container.NewHBox(
 				widget.NewButtonWithIcon("(Re-)login in OBS", theme.LoginIcon(), func() {
-					if cfg.Backends[obs.ID] == nil {
+					if streamDCfg.Backends[obs.ID] == nil {
-						obs.InitConfig(cfg.Backends)
+						obs.InitConfig(streamDCfg.Backends)
 					}
 
-					cfg.Backends[obs.ID].Enable = nil
+					streamDCfg.Backends[obs.ID].Enable = nil
-					cfg.Backends[obs.ID].Config = obs.PlatformSpecificConfig{}
+					streamDCfg.Backends[obs.ID].Config = obs.PlatformSpecificConfig{}
 
-					if err := p.SetStreamDConfig(ctx, cfg); err != nil {
+					if err := p.SetStreamDConfig(ctx, streamDCfg); err != nil {
 						p.DisplayError(err)
 						return
 					}
@@ -1688,14 +1715,14 @@ func (p *Panel) openSettingsWindow(ctx context.Context) error {
 			),
 			container.NewHBox(
 				widget.NewButtonWithIcon("(Re-)login in Twitch", theme.LoginIcon(), func() {
-					if cfg.Backends[twitch.ID] == nil {
+					if streamDCfg.Backends[twitch.ID] == nil {
-						twitch.InitConfig(cfg.Backends)
+						twitch.InitConfig(streamDCfg.Backends)
 					}
 
-					cfg.Backends[twitch.ID].Enable = nil
+					streamDCfg.Backends[twitch.ID].Enable = nil
-					cfg.Backends[twitch.ID].Config = twitch.PlatformSpecificConfig{}
+					streamDCfg.Backends[twitch.ID].Config = twitch.PlatformSpecificConfig{}
 
-					if err := p.SetStreamDConfig(ctx, cfg); err != nil {
+					if err := p.SetStreamDConfig(ctx, streamDCfg); err != nil {
 						p.DisplayError(err)
 						return
 					}
@@ -1709,14 +1736,14 @@ func (p *Panel) openSettingsWindow(ctx context.Context) error {
 			),
 			container.NewHBox(
 				widget.NewButtonWithIcon("(Re-)login in Kick", theme.LoginIcon(), func() {
-					if cfg.Backends[kick.ID] == nil {
+					if streamDCfg.Backends[kick.ID] == nil {
-						kick.InitConfig(cfg.Backends)
+						kick.InitConfig(streamDCfg.Backends)
 					}
 
-					cfg.Backends[kick.ID].Enable = nil
+					streamDCfg.Backends[kick.ID].Enable = nil
-					cfg.Backends[kick.ID].Config = kick.PlatformSpecificConfig{}
+					streamDCfg.Backends[kick.ID].Config = kick.PlatformSpecificConfig{}
 
-					if err := p.SetStreamDConfig(ctx, cfg); err != nil {
+					if err := p.SetStreamDConfig(ctx, streamDCfg); err != nil {
 						p.DisplayError(err)
 						return
 					}
@@ -1730,13 +1757,13 @@ func (p *Panel) openSettingsWindow(ctx context.Context) error {
 			),
 			container.NewHBox(
 				widget.NewButtonWithIcon("(Re-)login in YouTube", theme.LoginIcon(), func() {
-					if cfg.Backends[youtube.ID] == nil {
+					if streamDCfg.Backends[youtube.ID] == nil {
-						youtube.InitConfig(cfg.Backends)
+						youtube.InitConfig(streamDCfg.Backends)
 					}
 
-					cfg.Backends[youtube.ID].Enable = nil
+					streamDCfg.Backends[youtube.ID].Enable = nil
-					cfg.Backends[youtube.ID].Config = youtube.PlatformSpecificConfig{}
+					streamDCfg.Backends[youtube.ID].Config = youtube.PlatformSpecificConfig{}
-					if err := p.SetStreamDConfig(ctx, cfg); err != nil {
+					if err := p.SetStreamDConfig(ctx, streamDCfg); err != nil {
 						p.DisplayError(err)
 						return
 					}
@@ -1750,6 +1777,11 @@ func (p *Panel) openSettingsWindow(ctx context.Context) error {
 			),
 			widget.NewSeparator(),
 			widget.NewSeparator(),
+			widget.NewRichTextFromMarkdown(`# Chat`),
+			enableChatNotifications,
+			enableChatMessageSoundsAlerts,
+			widget.NewSeparator(),
+			widget.NewSeparator(),
 			widget.NewRichTextFromMarkdown(`# Dashboard`),
 			enableScreenshotSendingCheckbox,
 			widget.NewLabel("The screen/display to screenshot:"),
@@ -1780,6 +1812,9 @@ func (p *Panel) openSettingsWindow(ctx context.Context) error {
 			beforeStopStreamCommandEntry,
 			widget.NewLabel("Run command on stream stop (after):"),
 			afterStopStreamCommandEntry,
+			widget.NewSeparator(),
+			widget.NewLabel("Run command on receiving a chat message (after):"),
+			afterReceivedChatMessage,
 		),
 		container.NewHBox(
 			cancelButton,
@@ -2457,8 +2492,12 @@ func (p *Panel) getSelectedProfile() Profile {
 	return xsync.DoA2R1(ctx, &p.configCacheLocker, getProfile, p.configCache, *p.selectedProfileName)
 }
 
-func (p *Panel) execCommand(ctx context.Context, cmdString string) {
+func (p *Panel) execCommand(
-	cmdExpanded, err := expandCommand(cmdString)
+	ctx context.Context,
+	cmdString string,
+	execContext any,
+) {
+	cmdExpanded, err := expandCommand(ctx, cmdString, execContext)
 	if err != nil {
 		p.DisplayError(err)
 	}
@@ -2469,12 +2508,21 @@ func (p *Panel) execCommand(ctx context.Context, cmdString string) {
 
 	logger.Infof(ctx, "executing %s with arguments %v", cmdExpanded[0], cmdExpanded[1:])
 	cmd := exec.Command(cmdExpanded[0], cmdExpanded[1:]...)
+	err = child_process_manager.ConfigureCommand(cmd)
+	if err != nil {
+		logger.Errorf(ctx, "unable to configure the command so that the process will die automatically: %v", err)
+	}
 	var stdout, stderr bytes.Buffer
 	cmd.Stdout = &stdout
 	cmd.Stderr = &stderr
 	observability.Go(ctx, func() {
 		err := cmd.Run()
-		if err != nil {
+		if err == nil {
+			err = child_process_manager.AddChildProcess(cmd.Process)
+			if err != nil {
+				logger.Errorf(ctx, "unable to add the process to the clean up list: %v", err)
+			}
+		} else {
 			p.DisplayError(err)
 		}
 
@@ -2669,7 +2717,7 @@ func (p *Panel) startStream(ctx context.Context) {
 		platCfg = p.configCache.Backends[obs.ID]
 	})
 	if onStreamStart, ok := platCfg.GetCustomString(config.CustomConfigKeyAfterStreamStart); ok {
-		p.execCommand(ctx, onStreamStart)
+		p.execCommand(ctx, onStreamStart, nil)
 	}
 
 	p.startStopButton.Refresh()
@@ -2736,7 +2784,7 @@ func (p *Panel) stopStreamNoLock(ctx context.Context) {
 		platCfg = p.configCache.Backends[obs.ID]
 	})
 	if onStreamStop, ok := platCfg.GetCustomString(config.CustomConfigKeyAfterStreamStop); ok {
-		p.execCommand(ctx, onStreamStop)
+		p.execCommand(ctx, onStreamStop, nil)
 	}
 
 	p.startStopButton.SetText(startStreamString())

pkg/streampanel/template.go

@@ -1,60 +1,43 @@
 package streampanel
 
 import (
-	"strings"
+	"context"
+	"fmt"
 
+	"github.com/facebookincubator/go-belt/tool/logger"
+	"github.com/go-andiamo/splitter"
 	"github.com/xaionaro-go/streamctl/pkg/expression"
 )
 
-func splitWithQuotes(s string) []string {
+var commandSplitter splitter.Splitter
-	var result []string
-	var current string
-	inQuotes := false
-	quoteChar := byte(0)
 
-	for i := 0; i < len(s); i++ {
+func init() {
-		c := s[i]
+	commandSplitter = splitter.MustCreateSplitter(
-
+		' ',
-		if inQuotes {
+		splitter.DoubleQuotesBackSlashEscaped,
-			if c == quoteChar {
+	).AddDefaultOptions(
-				inQuotes = false
+		splitter.Trim(" \t\r\n"),
-				quoteChar = 0
+		splitter.IgnoreEmpties,
-			} else {
+		splitter.UnescapeQuotes,
-				current += string(c)
+	)
-			}
-		} else {
-			switch c {
-			case ' ', '\t':
-				if len(current) > 0 {
-					result = append(result, current)
-					current = ""
-				}
-			case '\'', '"':
-				inQuotes = true
-				quoteChar = c
-			default:
-				current += string(c)
-			}
-		}
-	}
-
-	if len(current) > 0 {
-		result = append(result, current)
-	}
-
-	return result
 }
 
-func expandCommand(cmdString string) ([]string, error) {
+func expandCommand(
-	cmdStringExpanded, err := expression.Eval[string](expression.Expression(cmdString), nil)
+	ctx context.Context,
+	cmdString string,
+	context any,
+) ([]string, error) {
+	cmdStringExpanded, err := expression.Eval[string](expression.Expression(cmdString), context)
 	if err != nil {
 		return nil, err
 	}
 
-	cmdStringExpandedClean := strings.Trim(cmdStringExpanded, " \t\r\n")
+	logger.Debugf(ctx, "expanded command is: <%s>", cmdStringExpanded)
-	if len(cmdStringExpandedClean) == 0 {
+
-		return nil, nil
+	args, err := commandSplitter.Split(cmdStringExpanded)
+	if err != nil {
+		return nil, fmt.Errorf("unable to split '%s': %w", cmdStringExpanded, err)
 	}
 
-	return splitWithQuotes(cmdStringExpandedClean), nil
+	return args, nil
 }