Code refactoring after #859

internal/exec/exec.go

@@ -19,6 +19,7 @@ import (
 	"github.com/AlexxIT/go2rtc/pkg/magic"
 	pkg "github.com/AlexxIT/go2rtc/pkg/rtsp"
 	"github.com/AlexxIT/go2rtc/pkg/shell"
+	"github.com/AlexxIT/go2rtc/pkg/stdin"
 	"github.com/rs/zerolog"
 )
 
@@ -79,6 +80,10 @@ func execHandle(rawURL string) (core.Producer, error) {
 }
 
 func handlePipe(_ string, cmd *exec.Cmd, query url.Values) (core.Producer, error) {
+	if query.Get("backchannel") == "1" {
+		return stdin.NewClient(cmd)
+	}
+
 	r, err := PipeCloser(cmd, query)
 	if err != nil {
 		return nil, err

internal/execbc/init.go

@@ -1,21 +0,0 @@
-package execbc
-
-import (
-	"github.com/AlexxIT/go2rtc/internal/streams"
-	"github.com/AlexxIT/go2rtc/pkg/core"
-	"github.com/AlexxIT/go2rtc/pkg/execbc"
-	"github.com/AlexxIT/go2rtc/pkg/shell"
-)
-
-func Init() {
-	streams.HandleFunc("execbc", handle)
-}
-
-func handle(url string) (core.Producer, error) {
-	args := shell.QuoteSplit(url[7:])
-	con, err := execbc.NewClient(args)
-	if err != nil {
-		return nil, err
-	}
-	return con, nil
-}

main.go

@@ -9,7 +9,6 @@ import (
 	"github.com/AlexxIT/go2rtc/internal/dvrip"
 	"github.com/AlexxIT/go2rtc/internal/echo"
 	"github.com/AlexxIT/go2rtc/internal/exec"
-	"github.com/AlexxIT/go2rtc/internal/execbc"
 	"github.com/AlexxIT/go2rtc/internal/expr"
 	"github.com/AlexxIT/go2rtc/internal/ffmpeg"
 	"github.com/AlexxIT/go2rtc/internal/gopro"
@@ -81,7 +80,6 @@ func main() {
 	bubble.Init()   // bubble source
 	expr.Init()     // expr source
 	gopro.Init()    // gopro source
-	execbc.Init()   // Local Backchannel
 
 	// 6. Helper modules
 

pkg/execbc/client.go

@@ -1,53 +0,0 @@
-package execbc
-
-import (
-	"io"
-	"net"
-	"os/exec"
-
-	"github.com/AlexxIT/go2rtc/pkg/core"
-)
-
-type Client struct {
-	medias      []*core.Media
-	sender      *core.Sender
-	conn        net.Conn
-	send        int
-	pipeCloser  io.WriteCloser
-	commandArgs []string
-	cmd         *exec.Cmd
-}
-
-func NewClient(commandArgs []string) (*Client, error) {
-	c := &Client{commandArgs: commandArgs}
-	media := &core.Media{
-		Kind:      core.KindAudio,
-		Direction: core.DirectionSendonly,
-		Codecs: []*core.Codec{
-			{Name: core.CodecPCMA, ClockRate: 8000},
-		},
-	}
-
-	c.medias = append(c.medias, media)
-
-	cmdName := c.commandArgs[0]
-	args := c.commandArgs[1:]
-	cmd := *exec.Command(cmdName, args...)
-
-	pipeCloser, error := PipeCloser(&cmd)
-	if error != nil {
-		return nil, error
-	}
-	c.pipeCloser = pipeCloser
-	c.cmd = &cmd
-	return c, nil
-}
-
-func (c Client) Open() (err error) {
-	c.cmd.Run()
-	return
-}
-
-func (c Client) Close() (err error) {
-	return c.pipeCloser.Close()
-}

pkg/stdin/client.go

@@ -0,0 +1,41 @@
+package stdin
+
+import (
+	"io"
+	"os/exec"
+
+	"github.com/AlexxIT/go2rtc/pkg/core"
+)
+
+type Client struct {
+	cmd  *exec.Cmd
+	pipe io.WriteCloser
+
+	medias []*core.Media
+	sender *core.Sender
+	send   int
+}
+
+func NewClient(cmd *exec.Cmd) (*Client, error) {
+	pipe, err := PipeCloser(cmd)
+	if err != nil {
+		return nil, err
+	}
+
+	c := &Client{
+		pipe: pipe,
+		cmd:  cmd,
+		medias: []*core.Media{
+			{
+				Kind:      core.KindAudio,
+				Direction: core.DirectionSendonly,
+				Codecs: []*core.Codec{
+					{Name: core.CodecPCMA, ClockRate: 8000},
+					{Name: core.CodecPCM},
+				},
+			},
+		},
+	}
+
+	return c, nil
+}

pkg/stdin/consumer.go

@@ -1,4 +1,4 @@
-package execbc
+package stdin
 
 import (
 	"encoding/json"
@@ -19,8 +19,7 @@ func (c *Client) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiver
 	if c.sender == nil {
 		c.sender = core.NewSender(media, track.Codec)
 		c.sender.Handler = func(packet *rtp.Packet) {
-			c.pipeCloser.Write(packet.Payload)
+			_, _ = c.pipe.Write(packet.Payload)
-
 			c.send += len(packet.Payload)
 		}
 	}
@@ -30,28 +29,19 @@ func (c *Client) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiver
 }
 
 func (c *Client) Start() (err error) {
-	if err = c.Open(); err != nil {
+	return c.cmd.Run()
-		return
-	}
-	return
 }
 
 func (c *Client) Stop() (err error) {
 	if c.sender != nil {
 		c.sender.Close()
 	}
-
+	return c.pipe.Close()
-	if c.conn != nil {
-		_ = c.Close()
-		return c.conn.Close()
-	}
-
-	return nil
 }
 
 func (c *Client) MarshalJSON() ([]byte, error) {
 	info := &core.Info{
-		Type:   "Command Backchannel PCMA",
+		Type:   "Exec active consumer",
 		Medias: c.medias,
 		Send:   c.send,
 	}

pkg/stdin/pipe.go

@@ -1,10 +1,9 @@
-package execbc
+package stdin
 
 import (
+	"errors"
 	"io"
 	"os/exec"
-
-	"github.com/AlexxIT/go2rtc/pkg/core"
 )
 
 type pipeCloser struct {
@@ -15,13 +14,13 @@ type pipeCloser struct {
 
 func PipeCloser(cmd *exec.Cmd) (io.WriteCloser, error) {
 	stdin, err := cmd.StdinPipe()
-
 	if err != nil {
 		return nil, err
 	}
+
 	return pipeCloser{stdin, stdin, cmd}, nil
 }
 
 func (p pipeCloser) Close() (err error) {
-	return core.Any(p.Closer.Close(), p.cmd.Process.Kill(), p.cmd.Wait())
+	return errors.Join(p.Closer.Close(), p.cmd.Process.Kill(), p.cmd.Wait())
 }