feat: add support for MPEG-TS over unix

README.md

@@ -30,6 +30,7 @@ Live streams can be published to the server with:
 |[RTMP cameras and servers](#rtmp-cameras-and-servers)|RTMP, RTMPS, Enhanced RTMP|AV1, VP9, H265, H264|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3), AC-3, G711 (PCMA, PCMU), LPCM|
 |[HLS cameras and servers](#hls-cameras-and-servers)|Low-Latency HLS, MP4-based HLS, legacy HLS|AV1, VP9, [H265](#supported-browsers-1), H264|Opus, MPEG-4 Audio (AAC)|
 |[UDP/MPEG-TS](#udpmpeg-ts)|Unicast, broadcast, multicast|H265, H264, MPEG-4 Video (H263, Xvid), MPEG-1/2 Video|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3), AC-3|
+|[Unix/MPEG-TS](#unixmpeg-ts)||H265, H264, MPEG-4 Video (H263, Xvid), MPEG-1/2 Video|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3), AC-3|
 |[Raspberry Pi Cameras](#raspberry-pi-cameras)||H264||
 
 Live streams can be read from the server with:
@@ -858,6 +859,26 @@ The resulting stream will be available in path `/mypath`.
 
 Known clients that can publish with WebRTC and WHIP are [FFmpeg](#ffmpeg) and [GStreamer](#gstreamer).
 
+#### Unix/MPEG-TS
+
+The server supports ingesting Unix/MPEG-TS packets (i.e. MPEG-TS packets sent over unix sockets). For instance, you can generate a multicast Unix/MPEG-TS stream with FFmpeg:
+
+```sh
+ffmpeg -re -f lavfi -i testsrc=size=1280x720:rate=30 \
+-c:v libx264 -pix_fmt yuv420p -preset ultrafast -b:v 600k \
+-f mpegts unix:/tmp/ffmpeg.socket
+```
+
+Edit `mediamtx.yml` and replace everything inside section `paths` with the following content:
+
+```yml
+paths:
+  mypath:
+    source: unix:/tmp/ffmpeg.socket
+```
+
+The resulting stream will be available in path `/mypath`.
+
 ## Read from the server
 
 ### By software

internal/conf/path.go

@@ -396,6 +396,12 @@
 			return fmt.Errorf("'%s' is not a valid UDP URL", pconf.Source)
 		}
 
+	case strings.HasPrefix(pconf.Source, "unix:"):
+		_, _, err := net.SplitHostPort(pconf.Source)
+		if err != nil {
+			return fmt.Errorf("'%s' is not a valid unix socket", pconf.Source)
+		}
+
 	case strings.HasPrefix(pconf.Source, "srt://"):
 		_, err := gourl.Parse(pconf.Source)
 		if err != nil {

internal/core/static_source_handler.go

@@ -16,6 +16,7 @@
 	rtspsource "github.com/bluenviron/mediamtx/internal/staticsources/rtsp"
 	srtsource "github.com/bluenviron/mediamtx/internal/staticsources/srt"
 	udpsource "github.com/bluenviron/mediamtx/internal/staticsources/udp"
+	unixsource "github.com/bluenviron/mediamtx/internal/staticsources/unix"
 	webrtcsource "github.com/bluenviron/mediamtx/internal/staticsources/webrtc"
 )
 
@@ -102,6 +103,12 @@
 			Parent:      s,
 		}
 
+	case strings.HasPrefix(s.conf.Source, "unix:"):
+		s.instance = &unixsource.Source{
+			ReadTimeout: s.readTimeout,
+			Parent:      s,
+		}
+
 	case strings.HasPrefix(s.conf.Source, "srt://"):
 		s.instance = &srtsource.Source{
 			ReadTimeout: s.readTimeout,

internal/staticsources/unix/source.go

@@ -0,0 +1,140 @@
+// Package unix contains the UNIX static source.
+package unix
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"time"
+
+	"github.com/bluenviron/gortsplib/v4/pkg/description"
+	mcmpegts "github.com/bluenviron/mediacommon/v2/pkg/formats/mpegts"
+
+	"github.com/bluenviron/mediamtx/internal/conf"
+	"github.com/bluenviron/mediamtx/internal/counterdumper"
+	"github.com/bluenviron/mediamtx/internal/defs"
+	"github.com/bluenviron/mediamtx/internal/logger"
+	"github.com/bluenviron/mediamtx/internal/protocols/mpegts"
+	"github.com/bluenviron/mediamtx/internal/stream"
+)
+
+// Source is a unix static source.
+type Source struct {
+	ReadTimeout conf.Duration
+	Parent      defs.StaticSourceParent
+}
+
+// Log implements logger.Writer.
+func (s *Source) Log(level logger.Level, format string, args ...interface{}) {
+	s.Parent.Log(level, "[unix source] "+format, args...)
+}
+
+// Run implements StaticSource.
+func (s *Source) Run(params defs.StaticSourceRunParams) error {
+	s.Log(logger.Debug, "connecting")
+
+	network, address, err := net.SplitHostPort(params.ResolvedSource)
+	if err != nil {
+		return err
+	}
+
+	err = os.Remove(address)
+	if err != nil {
+		// not really important if it fails
+		s.Log(logger.Debug, "Failed to remove previous unix socket", err)
+	}
+
+	var socket net.Listener
+	socket, err = net.Listen(network, address)
+	if err != nil {
+		return err
+	}
+	defer socket.Close()
+
+	conn, err := socket.Accept()
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	readerErr := make(chan error)
+	go func() {
+		readerErr <- s.runReader(conn)
+	}()
+
+	select {
+	case err := <-readerErr:
+		return err
+
+	case <-params.Context.Done():
+		socket.Close()
+		<-readerErr
+		return fmt.Errorf("terminated")
+	}
+}
+
+func (s *Source) runReader(conn net.Conn) error {
+	conn.SetReadDeadline(time.Now().Add(time.Duration(s.ReadTimeout)))
+	r := &mcmpegts.Reader{
+		R: conn,
+	}
+	err := r.Initialize()
+	if err != nil {
+		return err
+	}
+
+	decodeErrors := &counterdumper.CounterDumper{
+		OnReport: func(val uint64) {
+			s.Log(logger.Warn, "%d decode %s",
+				val,
+				func() string {
+					if val == 1 {
+						return "error"
+					}
+					return "errors"
+				}())
+		},
+	}
+
+	decodeErrors.Start()
+	defer decodeErrors.Stop()
+
+	r.OnDecodeError(func(_ error) {
+		decodeErrors.Increase()
+	})
+
+	var stream *stream.Stream
+
+	medias, err := mpegts.ToStream(r, &stream, s)
+	if err != nil {
+		return err
+	}
+
+	res := s.Parent.SetReady(defs.PathSourceStaticSetReadyReq{
+		Desc:               &description.Session{Medias: medias},
+		GenerateRTPPackets: true,
+	})
+	if res.Err != nil {
+		return res.Err
+	}
+
+	defer s.Parent.SetNotReady(defs.PathSourceStaticSetNotReadyReq{})
+
+	stream = res.Stream
+
+	for {
+		conn.SetReadDeadline(time.Now().Add(time.Duration(s.ReadTimeout)))
+		err := r.Read()
+		if err != nil {
+			return err
+		}
+	}
+}
+
+// APISourceDescribe implements StaticSource.
+func (*Source) APISourceDescribe() defs.APIPathSourceOrReader {
+	return defs.APIPathSourceOrReader{
+		Type: "unixSource",
+		ID:   "",
+	}
+}

internal/staticsources/unix/source_test.go

@@ -0,0 +1,62 @@
+package unix
+
+import (
+	"bufio"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/bluenviron/mediacommon/v2/pkg/formats/mpegts"
+	"github.com/stretchr/testify/require"
+
+	"github.com/bluenviron/mediamtx/internal/conf"
+	"github.com/bluenviron/mediamtx/internal/defs"
+	"github.com/bluenviron/mediamtx/internal/test"
+)
+
+func TestSource(t *testing.T) {
+	te := test.NewSourceTester(
+		func(p defs.StaticSourceParent) defs.StaticSource {
+			return &Source{
+				ReadTimeout: conf.Duration(10 * time.Second),
+				Parent:      p,
+			}
+		},
+		"unix:/tmp/mediamtx-test.sock",
+		&conf.Path{},
+	)
+	defer te.Close()
+
+	time.Sleep(50 * time.Millisecond)
+
+	conn, err := net.Dial("unix", "/tmp/mediamtx-test.sock")
+	require.NoError(t, err)
+	defer conn.Close()
+
+	track := &mpegts.Track{
+		Codec: &mpegts.CodecH264{},
+	}
+
+	bw := bufio.NewWriter(conn)
+	w := &mpegts.Writer{
+		W:      bw,
+		Tracks: []*mpegts.Track{track},
+	}
+	err = w.Initialize()
+	require.NoError(t, err)
+
+	err = w.WriteH264(track, 0, 0, [][]byte{{ // IDR
+		5, 1,
+	}})
+	require.NoError(t, err)
+
+	err = w.WriteH264(track, 0, 0, [][]byte{{ // non-IDR
+		5, 2,
+	}})
+	require.NoError(t, err)
+
+	err = bw.Flush()
+	require.NoError(t, err)
+
+	<-te.Unit
+}