project structure refactoring

external/process/extprocess.go

@@ -0,0 +1,132 @@
+package process
+
+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{}
+
+	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.Debugf("Starting external process: %s %v", u.path, u.args)
+	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)
+	}
+
+	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()
+			}*/
+		}
+	}
+}

external/process/pipedprocess.go

@@ -0,0 +1,136 @@
+package process
+
+import (
+	"os"
+	"os/exec"
+	"strings"
+	"sync"
+	"syscall"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// PipedCmd struct manages a sequence of piped commands.
+type PipedCmd struct {
+	cmds    []*exec.Cmd
+	mu      sync.Mutex
+	running bool
+}
+
+// InitPipedCmd initializes a PipedCmd instance with a sequence of commands.
+func InitPipedCmd(cmd string) *PipedCmd {
+	pipedCmd := &PipedCmd{}
+	pipedCmd.ChangeCmd(cmd)
+	return pipedCmd
+}
+
+func (p *PipedCmd) ChangeCmd(cmd string) {
+	cmds := strings.Split(cmd, "|")
+	for i, c := range cmds {
+		cmds[i] = strings.TrimSpace(c)
+	}
+
+	log.Debugf("Cmds: %+v", cmds)
+
+	pipedCmds := make([]*exec.Cmd, len(cmds))
+
+	// Initialize each command in the sequence
+	for i, cmd := range cmds {
+		if len(cmd) == 0 {
+			continue
+		}
+		cmdArgs := strings.Split(cmd, " ")
+		pipedCmds[i] = exec.Command(cmdArgs[0], cmdArgs[1:]...)
+	}
+
+	p.cmds = pipedCmds
+}
+
+// Start begins execution of all commands in the piped sequence.
+func (p *PipedCmd) Start() error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if p.running {
+		log.Debugf("Process is already running.")
+		return nil
+	}
+
+	// Pipe each command's output to the next command's input
+	for i := 0; i < len(p.cmds)-1; i++ {
+		stdout, err := p.cmds[i].StdoutPipe()
+		if err != nil {
+			return err
+		}
+		p.cmds[i+1].Stdin = stdout
+	}
+	// pipe stdout and stderr of the last command into os
+	p.cmds[len(p.cmds)-1].Stdout = os.Stdout
+	p.cmds[len(p.cmds)-1].Stderr = os.Stderr
+
+	// Start each command in the sequence
+	for _, cmd := range p.cmds {
+		if err := cmd.Start(); err != nil {
+			p.terminateAll()
+			return err
+		} else {
+			log.Debugf("Started process: (%d) %v", cmd.Process.Pid, cmd.Args)
+		}
+	}
+
+	p.running = true
+
+	// Monitor commands in a goroutine to handle failures
+	go p.monitorCommands()
+
+	return nil
+}
+
+// monitorCommands waits for each command to finish and checks for errors.
+func (p *PipedCmd) monitorCommands() {
+	var wg sync.WaitGroup
+	wg.Add(len(p.cmds))
+
+	for _, cmd := range p.cmds {
+		go func(cmd *exec.Cmd) {
+			defer wg.Done()
+			err := cmd.Wait()
+			if err != nil && err.Error() != "signal: terminated" {
+				log.Debugf("Command failed: %v", err)
+				p.terminateAll() // Terminate all if any command fails
+			}
+		}(cmd)
+	}
+
+	// Wait for all commands to complete or terminate
+	wg.Wait()
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.running = false
+}
+
+// terminateAll sends a termination signal to all running commands.
+func (p *PipedCmd) terminateAll() {
+	for _, cmd := range p.cmds {
+		if cmd.Process != nil {
+			log.Debugf("Sending SIGTERM to: (%d) %v", cmd.Process.Pid, cmd.Args)
+			_ = cmd.Process.Signal(syscall.SIGTERM) // Send SIGTERM to allow graceful termination
+		}
+	}
+
+	p.running = false
+}
+
+// Stop manually stops all commands in the sequence.
+func (p *PipedCmd) Stop() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if !p.running {
+		log.Debug("Process is not running.")
+		return
+	}
+
+	log.Debug("Stopping process...")
+	p.terminateAll()
+}