diff --git a/README.md b/README.md index 5d3efd83..06c024ab 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,8 @@ Live streams can be published to the server with: |[RTMP clients](#rtmp-clients)|RTMP, RTMPS, Enhanced RTMP|AV1, VP9, H265, H264|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3), AC-3, G711 (PCMA, PCMU), LPCM| |[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)| -|[MPEG-TS](#mpeg-ts)|MPEG-TS over UDP, MPEG-TS over Unix socket|H265, H264, MPEG-4 Video (H263, Xvid), MPEG-1/2 Video|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3), AC-3| +|[MPEG-TS](#mpeg-ts)|MPEG-TS over UDP, MPEG-TS over Unix sockets|H265, H264, MPEG-4 Video (H263, Xvid), MPEG-1/2 Video|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3), AC-3| +|[RTP](#rtp)|RTP over UDP, RTP over Unix sockets|AV1, VP9, VP8, H265, H264, MPEG-4 Video (H263, Xvid), MPEG-1/2 Video, M-JPEG and any RTP-compatible codec|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3), AC-3, G726, G722, G711 (PCMA, PCMU), LPCM and any RTP-compatible codec| |[Raspberry Pi Cameras](#raspberry-pi-cameras)||H264|| Live streams can be read from the server with: @@ -102,6 +103,7 @@ _rtsp-simple-server_ has been rebranded as _MediaMTX_. The reason is pretty obvi * [RTMP cameras and servers](#rtmp-cameras-and-servers) * [HLS cameras and servers](#hls-cameras-and-servers) * [MPEG-TS](#mpeg-ts) + * [RTP](#rtp) * [Read from the server](#read-from-the-server) * [By software](#by-software-1) * [FFmpeg](#ffmpeg-1) @@ -275,7 +277,7 @@ Otherwise, [compile the server from source](#openwrt-1). #### FFmpeg -FFmpeg can publish a stream to the server in several ways (SRT client, SRT server, RTSP client, RTMP client, MPEG-TS over UDP, MPEG-TS over Unix sockets, WebRTC with WHIP). The recommended one consists in publishing as a [RTSP client](#rtsp-clients): +FFmpeg can publish a stream to the server in several ways (SRT client, SRT server, RTSP client, RTMP client, MPEG-TS over UDP, MPEG-TS over Unix sockets, WebRTC with WHIP, RTP over UDP, rtp over Unix sockets). The recommended one consists in publishing as a [RTSP client](#rtsp-clients): ``` ffmpeg -re -stream_loop -1 -i file.ts -c copy -f rtsp rtsp://localhost:8554/mystream @@ -291,7 +293,7 @@ The resulting stream is available in path `/mystream`. #### GStreamer -GStreamer can publish a stream to the server in several ways (SRT client, SRT server, RTSP client, RTMP client, MPEG-TS over UDP, WebRTC with WHIP). The recommended one consists in publishing as a [RTSP client](#rtsp-clients): +GStreamer can publish a stream to the server in several ways (SRT client, SRT server, RTSP client, RTMP client, MPEG-TS over UDP, WebRTC with WHIP, RTP over UDP). The recommended one consists in publishing as a [RTSP client](#rtsp-clients): ```sh gst-launch-1.0 rtspclientsink name=s location=rtsp://localhost:8554/mystream \ @@ -887,7 +889,7 @@ The resulting stream is available in path `/proxied`. #### MPEG-TS -The server supports ingesting MPEG-TS streams, shipped in several ways (UDP packets or Unix sockets). +The server supports ingesting MPEG-TS streams, shipped in two different ways (UDP packets or Unix sockets). In order to read a UDP MPEG-TS stream, edit `mediamtx.yml` and replace everything inside section `paths` with the following content: @@ -951,6 +953,66 @@ ffmpeg -re -f lavfi -i testsrc=size=1280x720:rate=30 \ -f mpegts unix:/tmp/socket.sock ``` +#### RTP + +The server supports ingesting RTP streams, shipped in two different ways (UDP packets or Unix sockets). + +In order to read a UDP RTP stream, edit `mediamtx.yml` and replace everything inside section `paths` with the following content: + +```yml +paths: + mypath: + source: udp+rtp://238.0.0.1:1234 + rtpSDP: | + v=0 + o=- 123456789 123456789 IN IP4 192.168.1.100 + s=H264 Video Stream + c=IN IP4 192.168.1.100 + t=0 0 + m=video 5004 RTP/AVP 96 + a=rtpmap:96 H264/90000 + a=fmtp:96 profile-level-id=42e01e;packetization-mode=1;sprop-parameter-sets=Z0LAHtkDxWhAAAADAEAAAAwDxYuS,aMuMsg== +``` + +`rtpSDP` must contain a valid SDP, that is a description of the RTP session. + +FFmpeg can generate a RTP over UDP stream: + +```sh +ffmpeg -re -f lavfi -i testsrc=size=1280x720:rate=30 \ +-c:v libx264 -pix_fmt yuv420p -preset ultrafast -b:v 600k \ +-f rtp udp://238.0.0.1:1234?pkt_size=1316 +``` + +The stream is available on path `/mypath`. + +Known clients that can publish with UDP and MPEG-TS are [FFmpeg](#ffmpeg) and [GStreamer](#gstreamer). + +Unix sockets are more efficient than UDP packets and can be used as transport by specifying the `unix+rtp` scheme: + +```yml +paths: + mypath: + source: unix+rtp:///tmp/socket.sock + rtpSDP: | + v=0 + o=- 123456789 123456789 IN IP4 192.168.1.100 + s=H264 Video Stream + c=IN IP4 192.168.1.100 + t=0 0 + m=video 5004 RTP/AVP 96 + a=rtpmap:96 H264/90000 + a=fmtp:96 profile-level-id=42e01e;packetization-mode=1;sprop-parameter-sets=Z0LAHtkDxWhAAAADAEAAAAwDxYuS,aMuMsg== +``` + +FFmpeg can generate such streams: + +```sh +ffmpeg -re -f lavfi -i testsrc=size=1280x720:rate=30 \ +-c:v libx264 -pix_fmt yuv420p -preset ultrafast -b:v 600k \ +-f rtp unix:/tmp/socket.sock +``` + ## Read from the server ### By software diff --git a/api/openapi.yaml b/api/openapi.yaml index 0f1bf8ef..8994ae64 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -383,6 +383,10 @@ components: rtspRangeStart: type: string + # RTP source + rtpSDP: + type: string + # Redirect source sourceRedirect: type: string @@ -595,6 +599,7 @@ components: - srtConn - srtSource - mpegtsSource + - rtpSource - webRTCSession - webRTCSource id: diff --git a/internal/conf/path.go b/internal/conf/path.go index e7d01f6a..a3c7a15c 100644 --- a/internal/conf/path.go +++ b/internal/conf/path.go @@ -153,6 +153,9 @@ type Path struct { RTSPRangeType RTSPRangeType `json:"rtspRangeType"` RTSPRangeStart string `json:"rtspRangeStart"` + // RTP source + RTPSDP string `json:"rtpSDP"` + // Redirect source SourceRedirect string `json:"sourceRedirect"` @@ -419,7 +422,7 @@ func (pconf *Path) validate( case strings.HasPrefix(pconf.Source, "udp://"): _, _, err := net.SplitHostPort(pconf.Source[len("udp://"):]) if err != nil { - return fmt.Errorf("'%s' is not a valid UDP URL", pconf.Source) + return fmt.Errorf("'%s' is not a valid UDP+MPEGTS URL", pconf.Source) } case strings.HasPrefix(pconf.Source, "udp+mpegts://"): @@ -430,6 +433,21 @@ func (pconf *Path) validate( case strings.HasPrefix(pconf.Source, "unix+mpegts://"): + case strings.HasPrefix(pconf.Source, "udp+rtp://"): + _, _, err := net.SplitHostPort(pconf.Source[len("udp+rtp://"):]) + if err != nil { + return fmt.Errorf("'%s' is not a valid UDP+RTP URL", pconf.Source) + } + + if pconf.RTPSDP == "" { + return fmt.Errorf("`rtpSDP` was not provided") + } + + case strings.HasPrefix(pconf.Source, "unix+rtp://"): + if pconf.RTPSDP == "" { + return fmt.Errorf("`rtpSDP` was not provided") + } + case strings.HasPrefix(pconf.Source, "srt://"): _, err := gourl.Parse(pconf.Source) if err != nil { diff --git a/internal/staticsources/handler.go b/internal/staticsources/handler.go index b6272281..e999111a 100644 --- a/internal/staticsources/handler.go +++ b/internal/staticsources/handler.go @@ -15,6 +15,7 @@ import ( ssmpegts "github.com/bluenviron/mediamtx/internal/staticsources/mpegts" ssrpicamera "github.com/bluenviron/mediamtx/internal/staticsources/rpicamera" ssrtmp "github.com/bluenviron/mediamtx/internal/staticsources/rtmp" + ssrtp "github.com/bluenviron/mediamtx/internal/staticsources/rtp" ssrtsp "github.com/bluenviron/mediamtx/internal/staticsources/rtsp" sssrt "github.com/bluenviron/mediamtx/internal/staticsources/srt" sswebrtc "github.com/bluenviron/mediamtx/internal/staticsources/webrtc" @@ -138,6 +139,13 @@ func (s *Handler) Initialize() { Parent: s, } + case strings.HasPrefix(s.Conf.Source, "udp+rtp://") || + strings.HasPrefix(s.Conf.Source, "unix+rtp://"): + s.instance = &ssrtp.Source{ + ReadTimeout: s.ReadTimeout, + Parent: s, + } + case s.Conf.Source == "rpiCamera": s.instance = &ssrpicamera.Source{ RTPMaxPayloadSize: s.RTPMaxPayloadSize, diff --git a/internal/staticsources/rtp/source.go b/internal/staticsources/rtp/source.go new file mode 100644 index 00000000..7dc37483 --- /dev/null +++ b/internal/staticsources/rtp/source.go @@ -0,0 +1,179 @@ +// Package rtp contains the RTP static source. +package rtp + +import ( + "fmt" + "net" + "net/url" + "time" + + "github.com/bluenviron/gortsplib/v4/pkg/description" + "github.com/bluenviron/gortsplib/v4/pkg/format" + "github.com/bluenviron/gortsplib/v4/pkg/rtptime" + "github.com/bluenviron/gortsplib/v4/pkg/sdp" + "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/stream" + "github.com/pion/rtp" +) + +type parent interface { + logger.Writer + SetReady(req defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes + SetNotReady(req defs.PathSourceStaticSetNotReadyReq) +} + +// Source is a RTP static source. +type Source struct { + ReadTimeout conf.Duration + Parent parent +} + +// Log implements logger.Writer. +func (s *Source) Log(level logger.Level, format string, args ...interface{}) { + s.Parent.Log(level, "[RTP source] "+format, args...) +} + +// Run implements StaticSource. +func (s *Source) Run(params defs.StaticSourceRunParams) error { + var sd sdp.SessionDescription + err := sd.Unmarshal([]byte(params.Conf.RTPSDP)) + if err != nil { + return err + } + + var desc description.Session + err = desc.Unmarshal(&sd) + if err != nil { + return err + } + + s.Log(logger.Debug, "connecting") + + u, err := url.Parse(params.ResolvedSource) + if err != nil { + return err + } + q := u.Query() + + var nc net.Conn + + switch u.Scheme { + case "unix+rtp": + nc, err = createUnix(u) + if err != nil { + return err + } + + default: + nc, err = createUDP(u.Host, q) + if err != nil { + return err + } + } + + readerErr := make(chan error) + go func() { + readerErr <- s.runReader(&desc, nc) + }() + + select { + case err = <-readerErr: + nc.Close() + return err + + case <-params.Context.Done(): + nc.Close() + <-readerErr + return fmt.Errorf("terminated") + } +} + +func (s *Source) runReader(desc *description.Session, nc net.Conn) error { + 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() + + var stream *stream.Stream + + timeDecoder := &rtptime.GlobalDecoder2{} + timeDecoder.Initialize() + + mediasByPayloadType := make(map[uint8]*description.Media) + formatsByPayloadType := make(map[uint8]format.Format) + + for _, media := range desc.Medias { + for _, forma := range media.Formats { + mediasByPayloadType[forma.PayloadType()] = media + formatsByPayloadType[forma.PayloadType()] = forma + } + } + + for { + buf := make([]byte, 1500) + nc.SetReadDeadline(time.Now().Add(time.Duration(s.ReadTimeout))) + n, err := nc.Read(buf) + if err != nil { + return err + } + + var pkt rtp.Packet + err = pkt.Unmarshal(buf[:n]) + if err != nil { + if stream != nil { + decodeErrors.Increase() + continue + } + return err + } + + if stream == nil { + res := s.Parent.SetReady(defs.PathSourceStaticSetReadyReq{ + Desc: desc, + GenerateRTPPackets: false, + }) + if res.Err != nil { + return res.Err + } + + defer s.Parent.SetNotReady(defs.PathSourceStaticSetNotReadyReq{}) + + stream = res.Stream + } + + media, ok := mediasByPayloadType[pkt.PayloadType] + if !ok { + continue + } + + forma := formatsByPayloadType[pkt.PayloadType] + + pts, ok := timeDecoder.Decode(forma, &pkt) + if !ok { + continue + } + + stream.WriteRTPPacket(media, forma, &pkt, time.Now(), pts) + } +} + +// APISourceDescribe implements StaticSource. +func (*Source) APISourceDescribe() defs.APIPathSourceOrReader { + return defs.APIPathSourceOrReader{ + Type: "rtpSource", + ID: "", + } +} diff --git a/internal/staticsources/rtp/source_test.go b/internal/staticsources/rtp/source_test.go new file mode 100644 index 00000000..ea1f46e2 --- /dev/null +++ b/internal/staticsources/rtp/source_test.go @@ -0,0 +1,229 @@ +package rtp + +import ( + "context" + "net" + "os" + "path/filepath" + "testing" + "time" + + "github.com/bluenviron/gortsplib/v4/pkg/format/rtph264" + "github.com/stretchr/testify/require" + + "github.com/bluenviron/mediamtx/internal/conf" + "github.com/bluenviron/mediamtx/internal/defs" + "github.com/bluenviron/mediamtx/internal/test" +) + +func multicastCapableInterface(t *testing.T) string { + intfs, err := net.Interfaces() + require.NoError(t, err) + + for _, intf := range intfs { + if (intf.Flags & net.FlagMulticast) != 0 { + return intf.Name + } + } + + t.Errorf("unable to find a multicast IP") + return "" +} + +func TestSourceUDP(t *testing.T) { + for _, ca := range []string{ + "unicast", + "multicast", + "multicast with interface", + "unicast with source", + } { + t.Run(ca, func(t *testing.T) { + var src string + + switch ca { + case "unicast": + src = "udp+rtp://127.0.0.1:9001" + + case "multicast": + src = "udp+rtp://238.0.0.1:9001" + + case "multicast with interface": + src = "udp+rtp://238.0.0.1:9001?interface=" + multicastCapableInterface(t) + + case "unicast with source": + src = "udp+rtp://127.0.0.1:9001?source=127.0.1.1" + } + + p := &test.StaticSourceParent{} + p.Initialize() + defer p.Close() + + so := &Source{ + ReadTimeout: conf.Duration(10 * time.Second), + Parent: p, + } + + done := make(chan struct{}) + defer func() { <-done }() + + ctx, ctxCancel := context.WithCancel(context.Background()) + defer ctxCancel() + + go func() { + so.Run(defs.StaticSourceRunParams{ //nolint:errcheck + Context: ctx, + ResolvedSource: src, + Conf: &conf.Path{ + RTPSDP: "v=0\n" + + "o=- 123456789 123456789 IN IP4 192.168.1.100\n" + + "s=H264 Video Stream\n" + + "c=IN IP4 192.168.1.100\n" + + "t=0 0\n" + + "m=video 5004 RTP/AVP 96\n" + + "a=rtpmap:96 H264/90000\n" + + "a=fmtp:96 profile-level-id=42e01e;packetization-mode=1\n", + }, + }) + close(done) + }() + + time.Sleep(50 * time.Millisecond) + + var dest string + + switch ca { + case "unicast": + dest = "127.0.0.1:9001" + + case "multicast": + dest = "238.0.0.1:9001" + + case "multicast with interface": + dest = "238.0.0.1:9001" + + case "unicast with source": + dest = "127.0.0.1:9001" + } + + udest, err := net.ResolveUDPAddr("udp", dest) + require.NoError(t, err) + + var usrc *net.UDPAddr + if ca == "unicast with source" { + usrc, err = net.ResolveUDPAddr("udp", "127.0.1.1:9020") + require.NoError(t, err) + } + + conn, err := net.DialUDP("udp", usrc, udest) + require.NoError(t, err) + defer conn.Close() //nolint:errcheck + + enc := &rtph264.Encoder{ + PayloadType: 96, + } + err = enc.Init() + require.NoError(t, err) + + pkts, err := enc.Encode([][]byte{ + {5, 1}, + }) + require.NoError(t, err) + + for _, pkt := range pkts { + var buf []byte + buf, err = pkt.Marshal() + require.NoError(t, err) + + _, err = conn.Write(buf) + require.NoError(t, err) + } + + <-p.Unit + }) + } +} + +func TestSourceUnixSocket(t *testing.T) { + for _, ca := range []string{ + "relative", + "absolute", + } { + t.Run(ca, func(t *testing.T) { + var pa string + if ca == "relative" { + pa = "test.sock" + } else { + pa = filepath.Join(os.TempDir(), "test.sock") + } + + func() { + p := &test.StaticSourceParent{} + p.Initialize() + defer p.Close() + + so := &Source{ + ReadTimeout: conf.Duration(10 * time.Second), + Parent: p, + } + + done := make(chan struct{}) + defer func() { <-done }() + + ctx, ctxCancel := context.WithCancel(context.Background()) + defer ctxCancel() + + go func() { + so.Run(defs.StaticSourceRunParams{ //nolint:errcheck + Context: ctx, + ResolvedSource: "unix+rtp://" + pa, + Conf: &conf.Path{ + RTPSDP: "v=0\n" + + "o=- 123456789 123456789 IN IP4 192.168.1.100\n" + + "s=H264 Video Stream\n" + + "c=IN IP4 192.168.1.100\n" + + "t=0 0\n" + + "m=video 5004 RTP/AVP 96\n" + + "a=rtpmap:96 H264/90000\n" + + "a=fmtp:96 profile-level-id=42e01e;packetization-mode=1\n", + }, + }) + close(done) + }() + + time.Sleep(50 * time.Millisecond) + + _, err := os.Stat(pa) + require.NoError(t, err) + + conn, err := net.Dial("unix", pa) + require.NoError(t, err) + defer conn.Close() + + enc := &rtph264.Encoder{ + PayloadType: 96, + } + err = enc.Init() + require.NoError(t, err) + + pkts, err := enc.Encode([][]byte{ + {5, 1}, + }) + require.NoError(t, err) + + for _, pkt := range pkts { + var buf []byte + buf, err = pkt.Marshal() + require.NoError(t, err) + + _, err = conn.Write(buf) + require.NoError(t, err) + } + + <-p.Unit + }() + + _, err := os.Stat(pa) + require.Error(t, err) + }) + } +} diff --git a/internal/staticsources/rtp/udp.go b/internal/staticsources/rtp/udp.go new file mode 100644 index 00000000..9634a39d --- /dev/null +++ b/internal/staticsources/rtp/udp.go @@ -0,0 +1,160 @@ +package rtp + +import ( + "fmt" + "net" + "net/url" + "time" + + "github.com/bluenviron/gortsplib/v4/pkg/multicast" + "github.com/bluenviron/mediamtx/internal/restrictnetwork" +) + +const ( + // same size as GStreamer's rtspsrc + udpKernelReadBufferSize = 0x80000 +) + +type udpConn struct { + pc net.PacketConn + sourceIP net.IP +} + +func (r *udpConn) Close() error { + return r.pc.Close() +} + +func (r *udpConn) Read(p []byte) (int, error) { + for { + n, addr, err := r.pc.ReadFrom(p) + + if r.sourceIP != nil && addr != nil && !addr.(*net.UDPAddr).IP.Equal(r.sourceIP) { + continue + } + + return n, err + } +} + +func (r *udpConn) Write(_ []byte) (int, error) { + panic("unimplemented") +} + +func (r *udpConn) LocalAddr() net.Addr { + panic("unimplemented") +} + +func (r *udpConn) RemoteAddr() net.Addr { + panic("unimplemented") +} + +func (r *udpConn) SetDeadline(_ time.Time) error { + panic("unimplemented") +} + +func (r *udpConn) SetReadDeadline(t time.Time) error { + return r.pc.SetReadDeadline(t) +} + +func (r *udpConn) SetWriteDeadline(_ time.Time) error { + panic("unimplemented") +} + +func defaultInterfaceForMulticast(multicastAddr *net.UDPAddr) (*net.Interface, error) { + conn, err := net.Dial("udp4", multicastAddr.String()) + if err != nil { + return nil, err + } + localAddr := conn.LocalAddr().(*net.UDPAddr) + conn.Close() + + interfaces, err := net.Interfaces() + if err != nil { + return nil, err + } + + for _, iface := range interfaces { + var addrs []net.Addr + addrs, err = iface.Addrs() + if err != nil { + continue + } + + for _, addr := range addrs { + var ip net.IP + switch v := addr.(type) { + case *net.IPNet: + ip = v.IP + case *net.IPAddr: + ip = v.IP + } + + if ip != nil && ip.Equal(localAddr.IP) { + return &iface, nil + } + } + } + + return nil, fmt.Errorf("could not find any interface for using multicast address %s", multicastAddr) +} + +type packetConn interface { + net.PacketConn + SetReadBuffer(int) error +} + +func createUDP(host string, q url.Values) (net.Conn, error) { + var sourceIP net.IP + + if src := q.Get("source"); src != "" { + sourceIP = net.ParseIP(src) + if sourceIP == nil { + return nil, fmt.Errorf("invalid source IP") + } + } + + addr, err := net.ResolveUDPAddr("udp", host) + if err != nil { + return nil, err + } + + var pc packetConn + + if ip4 := addr.IP.To4(); ip4 != nil && addr.IP.IsMulticast() { + var intf *net.Interface + + if intfName := q.Get("interface"); intfName != "" { + intf, err = net.InterfaceByName(intfName) + if err != nil { + return nil, err + } + } else { + intf, err = defaultInterfaceForMulticast(addr) + if err != nil { + return nil, err + } + } + + pc, err = multicast.NewSingleConn(intf, addr.String(), net.ListenPacket) + if err != nil { + return nil, err + } + } else { + var tmp net.PacketConn + tmp, err = net.ListenPacket(restrictnetwork.Restrict("udp", addr.String())) + if err != nil { + return nil, err + } + pc = tmp.(*net.UDPConn) + } + + // defer pc.Close() + + err = pc.SetReadBuffer(udpKernelReadBufferSize) + if err != nil { + pc.Close() + return nil, err + } + + return &udpConn{pc: pc, sourceIP: sourceIP}, nil +} diff --git a/internal/staticsources/rtp/unix.go b/internal/staticsources/rtp/unix.go new file mode 100644 index 00000000..418b916f --- /dev/null +++ b/internal/staticsources/rtp/unix.go @@ -0,0 +1,136 @@ +package rtp + +import ( + "fmt" + "net" + "net/url" + "os" + "sync" + "time" +) + +type unixConn struct { + l net.Listener + c net.Conn + mutex sync.Mutex + closed bool + deadline time.Time +} + +func (r *unixConn) Close() error { + r.mutex.Lock() + defer r.mutex.Unlock() + + r.closed = true + + r.l.Close() + + if r.c != nil { + r.c.Close() + } + + return nil +} + +func (r *unixConn) acceptWithDeadline() (net.Conn, error) { + done := make(chan struct{}) + defer func() { <-done }() + + terminate := make(chan struct{}) + defer close(terminate) + + go func() { + defer close(done) + select { + case <-time.After(time.Until(r.deadline)): + r.l.Close() + case <-terminate: + return + } + }() + + c, err := r.l.Accept() + if err != nil { + if time.Now().After(r.deadline) { + return nil, fmt.Errorf("deadline exceeded") + } + return nil, err + } + return c, nil +} + +func (r *unixConn) setConn(c net.Conn) error { + r.mutex.Lock() + defer r.mutex.Unlock() + + if r.closed { + return fmt.Errorf("closed") + } + + r.c = c + return nil +} + +func (r *unixConn) Read(p []byte) (int, error) { + if r.c == nil { + c, err := r.acceptWithDeadline() + if err != nil { + return 0, err + } + + err = r.setConn(c) + if err != nil { + return 0, err + } + } + + r.c.SetReadDeadline(r.deadline) + return r.c.Read(p) +} + +func (r *unixConn) Write(_ []byte) (int, error) { + panic("unimplemented") +} + +func (r *unixConn) LocalAddr() net.Addr { + panic("unimplemented") +} + +func (r *unixConn) RemoteAddr() net.Addr { + panic("unimplemented") +} + +func (r *unixConn) SetDeadline(_ time.Time) error { + panic("unimplemented") +} + +func (r *unixConn) SetReadDeadline(t time.Time) error { + r.deadline = t + return nil +} + +func (r *unixConn) SetWriteDeadline(_ time.Time) error { + panic("unimplemented") +} + +func createUnix(u *url.URL) (net.Conn, error) { + var pa string + if u.Path != "" { + pa = u.Path + } else { + pa = u.Host + } + + if pa == "" { + return nil, fmt.Errorf("invalid unix path") + } + + os.Remove(pa) + + socket, err := net.Listen("unix", pa) + if err != nil { + return nil, err + } + + return &unixConn{l: socket}, nil +} diff --git a/mediamtx.yml b/mediamtx.yml index d674a1e9..206765a9 100644 --- a/mediamtx.yml +++ b/mediamtx.yml @@ -440,6 +440,8 @@ pathDefaults: # * https://existing-url/stream.m3u8 -> the stream is pulled from another HLS server / camera with HTTPS # * udp+mpegts://ip:port -> the stream is pulled from MPEG-TS over UDP, by listening on the specified address # * unix+mpegts://socket -> the stream is pulled from MPEG-TS over Unix socket, by using the socket + # * udp+rtp://ip:port -> the stream is pulled from RTP over UDP, by listening on the specified address + # * unix+rtp://socket -> the stream is pulled from RTP over Unix socket, by using the socket # * srt://existing-url -> the stream is pulled from another SRT server / camera # * whep://existing-url -> the stream is pulled from another WebRTC server / camera # * wheps://existing-url -> the stream is pulled from another WebRTC server / camera with HTTPS @@ -529,6 +531,12 @@ pathDefaults: # * smpte: duration such as "300ms", "1.5m" or "2h45m", valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h" rtspRangeStart: + ############################################### + # Default path settings -> RTP source (when source is RTP) + + # session description protocol (SDP) of the RTP stream. + rtpSDP: + ############################################### # Default path settings -> Redirect source (when source is "redirect")