From 52832223f819c267ef4c46a31393bbacba29f238 Mon Sep 17 00:00:00 2001 From: Alex X Date: Tue, 30 Apr 2024 06:42:17 +0300 Subject: [PATCH] Code refactoring after #859 --- internal/exec/exec.go | 5 +++ internal/execbc/init.go | 21 ------------ main.go | 2 -- pkg/execbc/client.go | 53 ------------------------------- pkg/stdin/client.go | 41 ++++++++++++++++++++++++ pkg/{execbc => stdin}/consumer.go | 20 +++--------- pkg/{execbc => stdin}/pipe.go | 9 +++--- 7 files changed, 55 insertions(+), 96 deletions(-) delete mode 100644 internal/execbc/init.go delete mode 100644 pkg/execbc/client.go create mode 100644 pkg/stdin/client.go rename pkg/{execbc => stdin}/consumer.go (80%) rename pkg/{execbc => stdin}/pipe.go (71%) diff --git a/internal/exec/exec.go b/internal/exec/exec.go index aefab201..e3ffe5a2 100644 --- a/internal/exec/exec.go +++ b/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 diff --git a/internal/execbc/init.go b/internal/execbc/init.go deleted file mode 100644 index 7d346331..00000000 --- a/internal/execbc/init.go +++ /dev/null @@ -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 -} diff --git a/main.go b/main.go index aa218fd3..91bc9938 100644 --- a/main.go +++ b/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 diff --git a/pkg/execbc/client.go b/pkg/execbc/client.go deleted file mode 100644 index c5b3490d..00000000 --- a/pkg/execbc/client.go +++ /dev/null @@ -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() -} diff --git a/pkg/stdin/client.go b/pkg/stdin/client.go new file mode 100644 index 00000000..00337f34 --- /dev/null +++ b/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 +} diff --git a/pkg/execbc/consumer.go b/pkg/stdin/consumer.go similarity index 80% rename from pkg/execbc/consumer.go rename to pkg/stdin/consumer.go index 9595a47c..827ea735 100644 --- a/pkg/execbc/consumer.go +++ b/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 - } - return + return c.cmd.Run() } func (c *Client) Stop() (err error) { if c.sender != nil { c.sender.Close() } - - if c.conn != nil { - _ = c.Close() - return c.conn.Close() - } - - return nil + return c.pipe.Close() } func (c *Client) MarshalJSON() ([]byte, error) { info := &core.Info{ - Type: "Command Backchannel PCMA", + Type: "Exec active consumer", Medias: c.medias, Send: c.send, } diff --git a/pkg/execbc/pipe.go b/pkg/stdin/pipe.go similarity index 71% rename from pkg/execbc/pipe.go rename to pkg/stdin/pipe.go index 5ad73398..c58a1076 100644 --- a/pkg/execbc/pipe.go +++ b/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()) }