initial commit

http/hw/stream/extprocess.go

@@ -0,0 +1,135 @@
+package stream
+
+import (
+	"io"
+	"os"
+	"os/exec"
+	"sync"
+
+	log "github.com/sirupsen/logrus"
+)
+
+type ExtProcess struct {
+	path     string
+	cmd      *exec.Cmd
+	mu       sync.Mutex
+	args     []string
+	running  bool
+	stopChan chan struct{}
+	finished chan struct{}
+	state    State
+
+	stdin io.WriteCloser
+}
+
+func Init(path string, args []string) *ExtProcess {
+	return &ExtProcess{
+		args: args,
+		path: path,
+	}
+}
+
+func (u *ExtProcess) Start() {
+	u.mu.Lock()
+	defer u.mu.Unlock()
+
+	if u.running {
+		log.Debug("process is already running.")
+		return
+	}
+
+	u.stopChan = make(chan struct{})
+	u.finished = make(chan struct{})
+
+	log.Debug("Starting external process...")
+	u.cmd = exec.Command(u.path, u.args...)
+	u.cmd.Stdout = os.Stdout
+	u.cmd.Stderr = os.Stderr
+	stdin, err := u.cmd.StdinPipe()
+	if err != nil {
+		log.Errorf("Couldn't handle stdin pipe: %v", err)
+		return
+	}
+	u.stdin = stdin
+
+	err = u.cmd.Start()
+	if err != nil {
+		log.Errorf("Failed to start process: %v", err)
+	}
+	u.running = true
+
+	go func() {
+		err = u.cmd.Wait()
+		u.running = false
+		log.Errorf("process exited with error: %v", err)
+
+		// planned stop
+		if err == nil {
+			u.finished <- struct{}{}
+			close(u.stopChan)
+			return
+		}
+
+		u.stopChan <- struct{}{}
+		close(u.finished)
+	}()
+}
+
+// Stop the ustreamer application
+func (u *ExtProcess) Stop() {
+	u.mu.Lock()
+	defer u.mu.Unlock()
+
+	if !u.running {
+		log.Debug("process is not running.")
+		return
+	}
+
+	log.Warn("Stopping process...")
+	_, err := u.stdin.Write([]byte("q")) // TODO: works for ffmpeg, should be general or move into FFmpeg struct
+	if err != nil {
+		log.Errorf("Failed to stop process: %v", err)
+
+		if err := u.cmd.Process.Kill(); err != nil {
+			log.Errorf("Failed to kill process: %v", err)
+		}
+	}
+
+	if err := u.stdin.Close(); err != nil {
+		log.Errorf("Failed to close stdin: %v", err)
+	}
+
+	log.Info("waiting for finish")
+
+	select {
+	case <-u.finished:
+		log.Info("stopped as expected")
+	case <-u.stopChan:
+		log.Info("was killed")
+	}
+
+	u.running = false
+}
+
+func (u *ExtProcess) ChangeArgs(newArgs []string) {
+	u.mu.Lock()
+	defer u.mu.Unlock()
+
+	u.args = newArgs
+	log.Printf("Updated process arguments: %v", u.args)
+}
+
+func (u *ExtProcess) Watch() {
+	for {
+		select {
+		case <-u.stopChan:
+			log.Errorf("process stopped unexpectedly. Restarting...")
+			u.Start()
+			/*case <-time.After(1 * time.Second): // Adjust the monitoring interval as needed
+			if !u.running {
+				log.Errorf("process is not running. Restarting...")
+				u.Start()
+			}*/
+		}
+	}
+}