Fix streaming to Kick

pkg/proxy/tcp_proxy.go

@@ -0,0 +1,185 @@
+package proxy
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net"
+	"sync"
+
+	"github.com/facebookincubator/go-belt/tool/logger"
+
+	"github.com/xaionaro-go/streamctl/pkg/observability"
+)
+
+type TCPProxy struct {
+	config       TCPConfig
+	wg           sync.WaitGroup
+	dstAddr      string
+	stopListener context.CancelFunc
+}
+
+type TCPConfig struct {
+	DestinationIsTLS     bool
+	DestinationTLSConfig *tls.Config
+}
+
+func NewTCP(dstAddr string, config *TCPConfig) *TCPProxy {
+	if config == nil {
+		config = &TCPConfig{}
+	}
+	return &TCPProxy{
+		dstAddr: dstAddr,
+		config:  *config,
+	}
+}
+
+func (p *TCPProxy) ListenRandomPort(
+	ctx context.Context,
+) (*net.TCPAddr, error) {
+	if p.stopListener != nil {
+		return nil, fmt.Errorf("the proxy is already listening another port")
+	}
+
+	const localhost = "127.0.0.1"
+	listener, err := net.ListenTCP("tcp", &net.TCPAddr{
+		IP:   net.ParseIP(localhost),
+		Port: 0,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("unable to start listening a random port at '%s': %w", localhost, err)
+	}
+	resultingAddr := listener.Addr().(*net.TCPAddr)
+	ctx, cancelFunc := context.WithCancel(ctx)
+	p.stopListener = cancelFunc
+	p.wg.Add(1)
+	observability.Go(ctx, func() {
+		defer p.wg.Done()
+		<-ctx.Done()
+		listener.Close()
+	})
+	p.wg.Add(1)
+	observability.Go(ctx, func() {
+		defer p.wg.Done()
+		err := p.serve(ctx, listener)
+		if err != nil {
+			logger.Errorf(ctx, "unable to serve at '%s': %v", resultingAddr, err)
+		}
+	})
+	return resultingAddr, nil
+}
+
+// Serve accepts and handles the connections from the listener, and
+// it takes ownership of the listener (it will be closed when TLSProxyTLS
+// will be closed)
+func (p *TCPProxy) serve(
+	ctx context.Context,
+	listener net.Listener,
+) error {
+	for {
+		conn, err := listener.Accept()
+		if err != nil {
+			return fmt.Errorf("unable to accept a connection: %w", err)
+		}
+
+		p.wg.Add(1)
+		observability.Go(ctx, func() {
+			defer p.wg.Done()
+			err := p.handleConnection(ctx, conn)
+			if err != nil {
+				logger.Errorf(ctx, "unable to handle connection %s->%s: %v", conn.LocalAddr(), conn.RemoteAddr(), err)
+				_ = p.Close()
+			}
+		})
+	}
+}
+
+func (p *TCPProxy) Close() error {
+	if p.stopListener == nil {
+		return fmt.Errorf("the proxy is not opened, and thus cannot be closed")
+	}
+	p.stopListener()
+	p.stopListener = nil
+	p.wg.Wait()
+	return nil
+}
+
+func (p *TCPProxy) handleConnection(
+	ctx context.Context,
+	conn net.Conn,
+) error {
+	var (
+		dstConn net.Conn
+		err     error
+	)
+	if p.config.DestinationIsTLS {
+		dstConn, err = tls.Dial("tcp", p.dstAddr, p.config.DestinationTLSConfig)
+	} else {
+		dstConn, err = net.Dial("tcp", p.dstAddr)
+	}
+	if err != nil {
+		return fmt.Errorf("unable to dial destination '%s' (isTLS: %v): %w", p.dstAddr, p.config.DestinationIsTLS, err)
+	}
+
+	err = proxyConns(ctx, conn, dstConn)
+	if err != nil {
+		return fmt.Errorf("unable to proxy connections %s->%s <-> %s->%s: %w", conn.LocalAddr(), conn.RemoteAddr(), dstConn.LocalAddr(), dstConn.RemoteAddr(), err)
+	}
+
+	return nil
+}
+
+func proxyConns(
+	ctx context.Context,
+	srcConn net.Conn,
+	dstConn net.Conn,
+) error {
+	ctx, cancelFunc := context.WithCancel(ctx)
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	observability.Go(ctx, func() {
+		defer wg.Done()
+		err := pipeConn(srcConn, dstConn)
+		if err != nil {
+			cancelFunc()
+			logger.Error(ctx, "unable to pipe traffic from %s->%s to %s->%s: %v", srcConn.LocalAddr(), srcConn.RemoteAddr(), srcConn.LocalAddr(), dstConn.RemoteAddr(), err)
+		}
+	})
+	wg.Add(1)
+	observability.Go(ctx, func() {
+		defer wg.Done()
+		err := pipeConn(dstConn, srcConn)
+		if err != nil {
+			cancelFunc()
+			logger.Error(ctx, "unable to pipe traffic from %s->%s to %s->%s: %v", dstConn.RemoteAddr(), dstConn.LocalAddr(), srcConn.RemoteAddr(), srcConn.LocalAddr(), err)
+		}
+	})
+
+	<-ctx.Done()
+	_ = srcConn.Close()
+	_ = dstConn.Close()
+	wg.Wait()
+	return nil
+}
+
+func pipeConn(
+	from, to net.Conn,
+) error {
+	buf := make([]byte, 1<<10)
+	for {
+		r, err := from.Read(buf)
+		if err != nil {
+			return fmt.Errorf("unable to read from the source: %w", err)
+		}
+		packet := buf[:r]
+
+		w, err := to.Write(packet)
+		if err != nil {
+			return fmt.Errorf("unable to write to the output: %w", err)
+		}
+		if w < r {
+			return fmt.Errorf("only %d bytes of %d were written", w, r)
+		}
+	}
+}

pkg/recoder/libav/recoder/output.go

@@ -3,16 +3,18 @@ package recoder
 import (
 	"context"
 	"fmt"
-	"log"
 	"net/url"
 	"strings"
 
 	"github.com/asticode/go-astiav"
 	"github.com/asticode/go-astikit"
 	"github.com/facebookincubator/go-belt/tool/logger"
+	"github.com/xaionaro-go/streamctl/pkg/proxy"
 	"github.com/xaionaro-go/streamctl/pkg/recoder"
 )
 
+const unwrapTLSViaProxy = false
+
 type OutputConfig = recoder.OutputConfig
 
 type Output struct {
@@ -45,23 +47,59 @@ func NewOutputFromURL(
 	}
 
 	if streamKey != "" {
-		if !strings.HasSuffix(url.Path, "/") {
+		switch {
+		case url.Path == "" || url.Path == "/":
+			url.Path = "//"
+		case !strings.HasSuffix(url.Path, "/"):
 			url.Path += "/"
 		}
 		url.Path += streamKey
 	}
 
+	if url.Port() == "" {
+		switch url.Scheme {
+		case "rtmp":
+			url.Host += ":1935"
+		case "rtmps":
+			url.Host += ":443"
+		}
+	}
+
+	needUnwrapTLSFor := ""
+	switch url.Scheme {
+	case "rtmps":
+		needUnwrapTLSFor = "rtmp"
+	}
+
 	output := &Output{
 		Closer: astikit.NewCloser(),
 	}
 
+	if needUnwrapTLSFor != "" && unwrapTLSViaProxy {
+		proxy := proxy.NewTCP(url.Host, &proxy.TCPConfig{
+			DestinationIsTLS: true,
+		})
+		proxyAddr, err := proxy.ListenRandomPort(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("unable to make a TLS-proxy: %w", err)
+		}
+		output.Closer.Add(func() {
+			err := proxy.Close()
+			if err != nil {
+				logger.Errorf(ctx, "unable to close the TLS-proxy: %w", err)
+			}
+		})
+		url.Scheme = needUnwrapTLSFor
+		url.Host = proxyAddr.String()
+	}
+
 	formatContext, err := astiav.AllocOutputFormatContext(
 		nil,
 		formatFromScheme(url.Scheme),
 		url.String(),
 	)
 	if err != nil {
-		return nil, fmt.Errorf("allocating output format context failed: %w", err)
+		return nil, fmt.Errorf("allocating output format context failed using URL '%s': %w", url, err)
 	}
 	if formatContext == nil {
 		// TODO: is there a way to extract the actual error code or something?
@@ -70,20 +108,20 @@ func NewOutputFromURL(
 	output.FormatContext = formatContext
 	output.Closer.Add(output.FormatContext.Free)
 
-	// if output is a file:
 	if !output.FormatContext.OutputFormat().Flags().Has(astiav.IOFormatFlagNofile) {
-		logger.Tracef(ctx, "destination '%s' is a file", url.String())
+		// if output is a file:
+		logger.Tracef(ctx, "destination '%s' is a file", url)
 		ioContext, err := astiav.OpenIOContext(
 			url.String(),
 			astiav.NewIOContextFlags(astiav.IOContextFlagWrite),
 		)
 		if err != nil {
-			log.Fatal(fmt.Errorf("main: opening io context failed: %w", err))
+			return nil, fmt.Errorf("unable to open IO context (URL: '%s'): %w", url, err)
 		}
 		output.Closer.Add(func() {
 			err := ioContext.Close()
 			if err != nil {
-				logger.Errorf(ctx, "unable to close the IO context: %w", err)
+				logger.Errorf(ctx, "unable to close the IO context (URL: %s): %v", url, err)
 			}
 		})
 		output.FormatContext.SetPb(ioContext)

pkg/recoder/libav/saferecoder/process/server/server.go

@@ -168,9 +168,10 @@ func (srv *GRPCServer) newOutputByURL(
 	output, err := recoder.NewOutputFromURL(ctx, path.Url.Url, path.Url.AuthKey, config)
 	if err != nil {
 		return nil, fmt.Errorf(
-			"unable to initialize an output using URL '%s' and config %#+v",
+			"unable to initialize an output using URL '%s' and config %#+v: %w",
 			path.Url,
 			config,
+			err,
 		)
 	}
 

pkg/streamserver/streamforward/stream_forward.go

@@ -252,7 +252,7 @@ func (fwd *ActiveStreamForwarding) waitForPublisherAndStart(
 	defer func() {
 		err := output.Close()
 		if err != nil {
-			logger.Errorf(ctx, "unable to close the output: %w", err)
+			logger.Errorf(ctx, "unable to close the output: %v", err)
 		}
 	}()