add audio support, trying to manage MSD device

http/hw/stream/audio.go

@@ -1,33 +1,47 @@
 package stream
 
 import (
-	"github.com/gin-gonic/gin"
 	"io"
 	"net/http"
 	"os/exec"
+
+	"github.com/gin-gonic/gin"
+	log "github.com/sirupsen/logrus"
 )
 
+type Audio struct {
+}
+
 func AudioHandler(c *gin.Context) {
 	cmd := exec.Command("ffmpeg",
 		"-f", "alsa",
 		"-i", "hw:0,0",
 		"-acodec", "aac",
 		"-f", "mp4", "-")
+
+	//c := "arecord -D hw:0,0 -f cd -r 44100 -c 2 | /app/ffmpeg -re -f wav -i pipe:0 -c:a aac -b:a 128k -ar 44100 -ac 2 -f rtp rtp://127.0.0.1:5006?pkt_size=1200"
+	//cmd := exec.Command("sh", "-c", c)
 	// ffmpeg -f alsa -i hw:0,0 -acodec aac -f mp4 -
 	stdout, err := cmd.StdoutPipe()
 	if err != nil {
 		c.String(http.StatusInternalServerError, "Failed to capture audio: %v", err)
+		log.Errorf("Failed to capture audio: %v", err)
 		return
 	}
 	if err := cmd.Start(); err != nil {
 		c.String(http.StatusInternalServerError, "Failed to start ffmpeg: %v", err)
+		log.Errorf("Failed to start ffmpeg: %v", err)
 		return
 	}
 	defer cmd.Wait()
 
-	c.Header("Content-Type", "audio/mp4")
+	//c.Header("Content-Type", "audio/aac")
+	c.Header("Content-Type", "audio/wav")
+	c.Header("Transfer-Encoding", "chunked")
+
 	if _, err := io.Copy(c.Writer, stdout); err != nil {
 		c.String(http.StatusInternalServerError, "Failed to stream audio: %v", err)
+		log.Errorf("Failed to stream audio: %v", err)
 		return
 	}
 }

http/hw/stream/extprocess.go

@@ -4,6 +4,7 @@ import (
 	"io"
 	"os"
 	"os/exec"
+	"rkkvm/config"
 	"sync"
 
 	log "github.com/sirupsen/logrus"
@@ -25,7 +26,7 @@ type ExtProcess struct {
 func Init(path string, args []string) *ExtProcess {
 	return &ExtProcess{
 		args: args,
-		path: path,
+		path: config.RootFS + path,
 	}
 }
 
@@ -41,7 +42,7 @@ func (u *ExtProcess) Start() {
 	u.stopChan = make(chan struct{})
 	u.finished = make(chan struct{})
 
-	log.Debug("Starting external process...")
+	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
@@ -99,8 +100,6 @@ func (u *ExtProcess) Stop() {
 		log.Errorf("Failed to close stdin: %v", err)
 	}
 
-	log.Info("waiting for finish")
-
 	select {
 	case <-u.finished:
 		log.Info("stopped as expected")

http/hw/stream/ffmpeg.go

@@ -13,6 +13,18 @@ import (
 // vp8
 //ffmpeg -re -i /dev/video0 -c:v libvpx -crf 10 -b:v 3M -deadline 1 -g 10 -error-resilient 1 -auto-alt-ref 1 -an -f rtp rtp://127.0.0.1:5004?pkt_size=1200
 
+/*
+arecord -D hw:0,0 -f cd -r 44100 -c 2 | /app/ffmpeg -re -init_hw_device rkmpp=hw -filter_hw_device hw \
+-f wav -thread_queue_size 4096 -i pipe:0 -f v4l2 -i /dev/video0 \
+-vf hwupload,scale_rkrga=h=720:force_original_aspect_ratio=1 \
+-c:v h264_rkmpp -flags +low_delay -b:v 6000000 -framerate 60 -g 10 \
+-map 0:a -c:a aac -b:a 128k -ar 44100 -ac 2 -f rtp rtp://127.0.0.1:5006?pkt_size=1200 \
+-map 1:v -f rtp rtp://127.0.0.1:5004?pkt_size=1200
+
+
+arecord -D hw:0,0 -f cd -r 44100 -c 2 | /app/ffmpeg -re -f wav -i pipe:0 -c:a aac -b:a 128k -ar 44100 -ac 2 -f rtp rtp://127.0.0.1:5006?pkt_size=1200
+*/
+
 // https://jsfiddle.net/z7ms3u5r/
 
 var ffmpeg *FFmpeg
@@ -40,10 +52,28 @@ func (f *FFmpeg) SetFPS(fps int) {
 	f.ChangeArgs(f.FormatArgs())
 }
 
+func (f *FFmpeg) SetResolution(height int) {
+	f.Height = height
+	if f.Height <= 0 {
+		f.Height = 1080
+	}
+
+	f.ChangeArgs(f.FormatArgs())
+}
+
+func (f *FFmpeg) SetGOP(gop int) {
+	f.GOP = gop
+	if f.GOP < 0 {
+		f.GOP = 0
+	}
+
+	f.ChangeArgs(f.FormatArgs())
+}
+
 func InitFFmpeg(path string, args []string) *FFmpeg {
 	ffmpeg = &FFmpeg{
 		ExtProcess: Init(path, args),
-		FFmpeg:     config.Get().FFmpeg,
+		FFmpeg:     config.Get().Video,
 	}
 	return ffmpeg
 }

http/hw/stream/pipedprocess.go

@@ -0,0 +1,270 @@
+package stream
+
+import (
+	"log"
+	"os/exec"
+	"rkkvm/config"
+	"strings"
+	"sync"
+	"syscall"
+)
+
+var pipedCmd *PipedCmd
+
+// PipedCmd struct manages a sequence of piped commands.
+type PipedCmd struct {
+	cmds     []*exec.Cmd
+	mu       sync.Mutex
+	running  bool
+	finished chan struct{}
+}
+
+// InitPipedCmd initializes a PipedCmd instance with a sequence of commands.
+func InitPipedCmd(cmds []string) *PipedCmd {
+	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(config.RootFS+"/"+cmdArgs[0], cmdArgs[1:]...)
+	}
+
+	pipedCmd = &PipedCmd{
+		cmds:     pipedCmds,
+		finished: make(chan struct{}),
+	}
+	return pipedCmd
+}
+
+// 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.Println("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
+	}
+
+	// Start each command in the sequence
+	for _, cmd := range p.cmds {
+		if err := cmd.Start(); err != nil {
+			p.terminateAll() // Clean up if any command fails to start
+			return err
+		}
+	}
+
+	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 {
+				log.Printf("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()
+	p.running = false
+	close(p.finished)
+	p.mu.Unlock()
+}
+
+// terminateAll sends a termination signal to all running commands.
+func (p *PipedCmd) terminateAll() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	for _, cmd := range p.cmds {
+		if cmd.Process != nil {
+			_ = cmd.Process.Signal(syscall.SIGTERM) // Send SIGTERM to allow graceful termination
+		}
+	}
+}
+
+// Stop manually stops all commands in the sequence.
+func (p *PipedCmd) Stop() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if !p.running {
+		log.Println("Process is not running.")
+		return
+	}
+
+	log.Println("Stopping process...")
+	p.terminateAll()
+	p.running = false
+	close(p.finished)
+}
+
+/*
+import (
+	"os"
+	"os/exec"
+	"strings"
+	"sync"
+
+	log "github.com/sirupsen/logrus"
+)
+
+type PipedCmd struct {
+	cmds     []string
+	cmdsExec []*exec.Cmd
+	mu       sync.Mutex
+	running  bool
+	stopChan chan struct{}
+	finished chan struct{}
+}
+
+// Init initializes the PipedCmd with a slice of command strings
+func InitPipedCmd(cmds []string) *PipedCmd {
+	return &PipedCmd{
+		cmds: cmds,
+	}
+}
+
+// Start initializes and starts the piped commands
+func (p *PipedCmd) Start() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if p.running {
+		log.Debug("process is already running.")
+		return
+	}
+
+	p.stopChan = make(chan struct{})
+	p.finished = make(chan struct{})
+
+	log.Debugf("Starting piped commands: <%s>", strings.Join(p.cmds, " | "))
+
+	// Create commands and set up pipes
+	for i, cmdStr := range p.cmds {
+		// Split command string into command and arguments
+		cmdParts := strings.Fields(cmdStr)
+		if len(cmdParts) == 0 {
+			log.Errorf("Empty command string at index %d", i)
+			continue
+		}
+
+		cmd := exec.Command(cmdParts[0], cmdParts[1:]...)
+
+		// Set up pipes for stdin/stdout
+		if i > 0 {
+			stdin, err := p.cmdsExec[i-1].StdoutPipe()
+			if err != nil {
+				log.Errorf("Couldn't set up stdout pipe for command %s: %v", p.cmdsExec[i-1].Path, err)
+				return
+			}
+			cmd.Stdin = stdin
+		}
+
+		cmd.Stderr = os.Stderr // Log stderr to standard error output
+
+		p.cmdsExec = append(p.cmdsExec, cmd)
+	}
+
+	// Start the first command
+	if err := p.cmdsExec[0].Start(); err != nil {
+		log.Errorf("Failed to start command: %v", err)
+		return
+	}
+	p.running = true
+
+	// Start remaining commands
+	for _, cmd := range p.cmdsExec[1:] {
+		if err := cmd.Start(); err != nil {
+			log.Errorf("Failed to start command: %v", err)
+			return
+		}
+	}
+
+	go func() {
+		// Wait for the last command to finish
+		err := p.cmdsExec[len(p.cmdsExec)-1].Wait()
+		p.running = false
+		log.Errorf("process exited with error: %v", err)
+
+		// Signal that the process has finished
+		close(p.finished)
+		close(p.stopChan)
+	}()
+}
+
+// Stop terminates the piped commands gracefully
+func (p *PipedCmd) Stop() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if !p.running {
+		log.Debug("process is not running.")
+		return
+	}
+
+	log.Warn("Stopping process...")
+
+	for _, cmd := range p.cmdsExec {
+		if err := cmd.Process.Kill(); err != nil {
+			log.Errorf("Failed to kill process: %v", err)
+		}
+	}
+
+	select {
+	case <-p.finished:
+		log.Info("stopped as expected")
+	case <-p.stopChan:
+		log.Info("was killed")
+	}
+
+	p.running = false
+}
+
+// ChangeArgs updates the command arguments
+func (p *PipedCmd) ChangeArgs(newCmds []string) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	p.cmds = newCmds
+	log.Printf("Updated process commands: %v", p.cmds)
+}
+
+// Watch monitors the process and restarts if it stops unexpectedly
+func (p *PipedCmd) Watch() {
+	for {
+		select {
+		case <-p.stopChan:
+			log.Errorf("process stopped unexpectedly. Restarting...")
+			p.Start()
+		}
+	}
+}
+*/

http/hw/stream/video.go

@@ -7,9 +7,11 @@ import (
 	"net/http"
 	"rkkvm/config"
 	"rkkvm/http/hw/rtc"
+	"rkkvm/http/reqrsp"
 	"strconv"
 
 	"github.com/gin-gonic/gin"
+	"github.com/pion/webrtc/v4"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -136,23 +138,29 @@ func MjpegHandler(c *gin.Context) {
 	}
 }
 
-func WebRTCHandshake(c *gin.Context) {
-	str, err := c.GetRawData()
-	if err != nil {
-		c.String(http.StatusBadRequest, err.Error())
+func WebRTCPeerConnect(c *gin.Context) {
+	var offer webrtc.SessionDescription
+	if err := c.ShouldBindJSON(&offer); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid offer"})
 		return
 	}
 
-	log.Debugf("Client session description: %s", string(str))
-
-	r := rtc.Get()
-	localSession, err := r.Handshake(string(str))
+	peer, err := rtc.NewPeer()
 	if err != nil {
-		c.String(http.StatusBadRequest, err.Error())
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create peer connection"})
 		return
 	}
 
-	c.String(http.StatusOK, localSession)
+	answer, err := rtc.Get().AddPeer(peer, offer)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to add peer"})
+		return
+	}
+
+	c.JSON(http.StatusOK, reqrsp.NanoKVMRsp{
+		Msg:  reqrsp.MsgSuccess,
+		Data: answer,
+	})
 }
 
 func WebRTCSettings(c *gin.Context) {