srt, udp: support publishing and reading MPEG-1/2/4 video with SRT an…

…d UDP/MPEG-TS

README.md

@@ -20,23 +20,23 @@ Live streams can be published to the server with:
 
 |protocol|variants|video codecs|audio codecs|
 |--------|--------|------------|------------|
-|[SRT clients](#srt-clients)||H265, H264|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3)|
-|[SRT servers](#srt-servers)||H265, H264|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3)|
+|[SRT clients](#srt-clients)||H265, H264, MPEG-4 Video (H263, Xvid), MPEG-1/2 Video|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3)|
+|[SRT servers](#srt-servers)||H265, H264, MPEG-4 Video (H263, Xvid), MPEG-1/2 Video|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3)|
 |[WebRTC clients](#webrtc-clients)|Browser-based, WHIP|AV1, VP9, VP8, H264|Opus, G722, G711|
 |[WebRTC servers](#webrtc-servers)|WHEP|AV1, VP9, VP8, H264|Opus, G722, G711|
 |[RTSP clients](#rtsp-clients)|UDP, TCP, RTSPS|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), G726, G722, G711, LPCM and any RTP-compatible codec|
 |[RTSP cameras and servers](#rtsp-cameras-and-servers)|UDP, UDP-Multicast, TCP, RTSPS|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), G726, G722, G711, LPCM and any RTP-compatible codec|
 |[RTMP clients](#rtmp-clients)|RTMP, RTMPS, Enhanced RTMP|AV1, VP9, H265, H264|MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3)|
 |[RTMP cameras and servers](#rtmp-cameras-and-servers)|RTMP, RTMPS, Enhanced RTMP|H264|MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3)|
 |[HLS cameras and servers](#hls-cameras-and-servers)|Low-Latency HLS, MP4-based HLS, legacy HLS|AV1, VP9, H265, H264|Opus, MPEG-4 Audio (AAC)|
-|[UDP/MPEG-TS](#udpmpeg-ts)|Unicast, broadcast, multicast|H265, H264|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3)|
+|[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)|
 |[Raspberry Pi Cameras](#raspberry-pi-cameras)||H264||
 
 And can be read from the server with:
 
 |protocol|variants|video codecs|audio codecs|
 |--------|--------|------------|------------|
-|[SRT](#srt)||H265, H264|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3)|
+|[SRT](#srt)||H265, H264, MPEG-4 Video (H263, Xvid), MPEG-1/2 Video|Opus, MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3)|
 |[WebRTC](#webrtc)|Browser-based, WHEP|AV1, VP9, VP8, H264|Opus, G722, G711|
 |[RTSP](#rtsp)|UDP, UDP-Multicast, TCP, RTSPS|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), G726, G722, G711, LPCM and any RTP-compatible codec|
 |[RTMP](#rtmp)|RTMP, RTMPS, Enhanced RTMP|H264|MPEG-4 Audio (AAC), MPEG-1/2 Audio (MP3)|

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/abema/go-mp4 v0.13.0
 	github.com/alecthomas/kong v0.8.0
 	github.com/bluenviron/gohlslib v1.0.2
-	github.com/bluenviron/gortsplib/v4 v4.0.2-0.20230911215322-4ede58cda208
+	github.com/bluenviron/gortsplib/v4 v4.0.2-0.20230916090332-99773e19afc9
 	github.com/bluenviron/mediacommon v1.2.0
 	github.com/datarhei/gosrt v0.5.4
 	github.com/fsnotify/fsnotify v1.6.0

go.sum

@@ -14,8 +14,8 @@ github.com/benburkert/openpgp v0.0.0-20160410205803-c2471f86866c h1:8XZeJrs4+ZYh
 github.com/benburkert/openpgp v0.0.0-20160410205803-c2471f86866c/go.mod h1:x1vxHcL/9AVzuk5HOloOEPrtJY0MaalYr78afXZ+pWI=
 github.com/bluenviron/gohlslib v1.0.2 h1:LDA/CubL525e9rLWw+G/9GbFS6iXwozmOg8KJBT4qF4=
 github.com/bluenviron/gohlslib v1.0.2/go.mod h1:oam0wsI2XqcHLTG6NM8HRvxAQsa3hIA0MLRiTOE7CB8=
-github.com/bluenviron/gortsplib/v4 v4.0.2-0.20230911215322-4ede58cda208 h1:w1aishvJ4U2TuM1SsFmWoKuFNogqhla3eh2Qn6AyRtE=
-github.com/bluenviron/gortsplib/v4 v4.0.2-0.20230911215322-4ede58cda208/go.mod h1:l6LO4TlHC3YdLIbEn89GeeSgzHaJlpAFF7NLZ7h4A+A=
+github.com/bluenviron/gortsplib/v4 v4.0.2-0.20230916090332-99773e19afc9 h1:NIJRhT/AYhPNe1GdWqAhDsYOTo6/hvAz5pEe1Ss+NuE=
+github.com/bluenviron/gortsplib/v4 v4.0.2-0.20230916090332-99773e19afc9/go.mod h1:aAHfFXD3LE19h86jUGJK7PpM1uwp9VFVReuH89ZlpuE=
 github.com/bluenviron/mediacommon v1.2.0 h1:5tz92r2S4gPSiTlycepjXFZCgwGfVL2htCeVsoBac+U=
 github.com/bluenviron/mediacommon v1.2.0/go.mod h1:/vlOVSebDwzdRtQONOKLua0fOSJg1tUDHpP+h9a0uqM=
 github.com/bytedance/sonic v1.5.0/go.mod h1:ED5hyg4y6t3/9Ku1R6dU/4KyJ48DZ4jPhfY1O2AihPM=

internal/core/srt_conn.go

@@ -295,6 +295,42 @@ func (c *srtConn) runPublishReader(sconn srt.Conn, path *path) error {
 				return nil
 			})
 
+		case *mpegts.CodecMPEG4Video:
+			medi = &description.Media{
+				Type: description.MediaTypeVideo,
+				Formats: []format.Format{&format.MPEG4Video{
+					PayloadTyp: 96,
+				}},
+			}
+
+			r.OnDataMPEGxVideo(track, func(pts int64, frame []byte) error {
+				stream.WriteUnit(medi, medi.Formats[0], &unit.MPEG4Video{
+					Base: unit.Base{
+						NTP: time.Now(),
+						PTS: decodeTime(pts),
+					},
+					Frame: frame,
+				})
+				return nil
+			})
+
+		case *mpegts.CodecMPEG1Video:
+			medi = &description.Media{
+				Type:    description.MediaTypeVideo,
+				Formats: []format.Format{&format.MPEG1Video{}},
+			}
+
+			r.OnDataMPEGxVideo(track, func(pts int64, frame []byte) error {
+				stream.WriteUnit(medi, medi.Formats[0], &unit.MPEG1Video{
+					Base: unit.Base{
+						NTP: time.Now(),
+						PTS: decodeTime(pts),
+					},
+					Frame: frame,
+				})
+				return nil
+			})
+
 		case *mpegts.CodecOpus:
 			medi = &description.Media{
 				Type: description.MediaTypeAudio,
@@ -509,6 +545,40 @@ func (c *srtConn) runRead(req srtNewConnReq, pathName string, user string, pass
 					return bw.Flush()
 				})
 
+			case *format.MPEG4Video:
+				track := addTrack(medi, &mpegts.CodecMPEG4Video{})
+
+				res.stream.AddReader(writer, medi, forma, func(u unit.Unit) error {
+					tunit := u.(*unit.MPEG4Video)
+					if tunit.Frame == nil {
+						return nil
+					}
+
+					sconn.SetWriteDeadline(time.Now().Add(time.Duration(c.writeTimeout)))
+					err = w.WriteMPEGxVideo(track, durationGoToMPEGTS(tunit.PTS), tunit.Frame)
+					if err != nil {
+						return err
+					}
+					return bw.Flush()
+				})
+
+			case *format.MPEG1Video:
+				track := addTrack(medi, &mpegts.CodecMPEG1Video{})
+
+				res.stream.AddReader(writer, medi, forma, func(u unit.Unit) error {
+					tunit := u.(*unit.MPEG1Video)
+					if tunit.Frame == nil {
+						return nil
+					}
+
+					sconn.SetWriteDeadline(time.Now().Add(time.Duration(c.writeTimeout)))
+					err = w.WriteMPEGxVideo(track, durationGoToMPEGTS(tunit.PTS), tunit.Frame)
+					if err != nil {
+						return err
+					}
+					return bw.Flush()
+				})
+
 			case *format.MPEG4AudioGeneric:
 				track := addTrack(medi, &mpegts.CodecMPEG4Audio{
 					Config: *forma.Config,

internal/core/srt_source.go

@@ -112,6 +112,25 @@ func (s *srtSource) runReader(sconn srt.Conn) error {
 		var medi *description.Media
 
 		switch tcodec := track.Codec.(type) {
+		case *mpegts.CodecH265:
+			medi = &description.Media{
+				Type: description.MediaTypeVideo,
+				Formats: []format.Format{&format.H265{
+					PayloadTyp: 96,
+				}},
+			}
+
+			r.OnDataH26x(track, func(pts int64, _ int64, au [][]byte) error {
+				stream.WriteUnit(medi, medi.Formats[0], &unit.H265{
+					Base: unit.Base{
+						NTP: time.Now(),
+						PTS: decodeTime(pts),
+					},
+					AU: au,
+				})
+				return nil
+			})
+
 		case *mpegts.CodecH264:
 			medi = &description.Media{
 				Type: description.MediaTypeVideo,
@@ -132,21 +151,38 @@ func (s *srtSource) runReader(sconn srt.Conn) error {
 				return nil
 			})
 
-		case *mpegts.CodecH265:
+		case *mpegts.CodecMPEG4Video:
 			medi = &description.Media{
 				Type: description.MediaTypeVideo,
-				Formats: []format.Format{&format.H265{
+				Formats: []format.Format{&format.MPEG4Video{
 					PayloadTyp: 96,
 				}},
 			}
 
-			r.OnDataH26x(track, func(pts int64, _ int64, au [][]byte) error {
-				stream.WriteUnit(medi, medi.Formats[0], &unit.H265{
+			r.OnDataMPEGxVideo(track, func(pts int64, frame []byte) error {
+				stream.WriteUnit(medi, medi.Formats[0], &unit.MPEG4Video{
 					Base: unit.Base{
 						NTP: time.Now(),
 						PTS: decodeTime(pts),
 					},
-					AU: au,
+					Frame: frame,
+				})
+				return nil
+			})
+
+		case *mpegts.CodecMPEG1Video:
+			medi = &description.Media{
+				Type:    description.MediaTypeVideo,
+				Formats: []format.Format{&format.MPEG1Video{}},
+			}
+
+			r.OnDataMPEGxVideo(track, func(pts int64, frame []byte) error {
+				stream.WriteUnit(medi, medi.Formats[0], &unit.MPEG1Video{
+					Base: unit.Base{
+						NTP: time.Now(),
+						PTS: decodeTime(pts),
+					},
+					Frame: frame,
 				})
 				return nil
 			})

internal/core/udp_source.go

@@ -81,7 +81,7 @@ func (s *udpSource) run(ctx context.Context, cnf *conf.PathConf, _ chan *conf.Pa
 	var pc packetConn
 
 	if ip4 := addr.IP.To4(); ip4 != nil && addr.IP.IsMulticast() {
-		pc, err = multicast.NewMultiConn(hostPort, net.ListenPacket)
+		pc, err = multicast.NewMultiConn(hostPort, true, net.ListenPacket)
 		if err != nil {
 			return err
 		}
@@ -183,6 +183,42 @@ func (s *udpSource) runReader(pc net.PacketConn) error {
 				return nil
 			})
 
+		case *mpegts.CodecMPEG4Video:
+			medi = &description.Media{
+				Type: description.MediaTypeVideo,
+				Formats: []format.Format{&format.MPEG4Video{
+					PayloadTyp: 96,
+				}},
+			}
+
+			r.OnDataMPEGxVideo(track, func(pts int64, frame []byte) error {
+				stream.WriteUnit(medi, medi.Formats[0], &unit.MPEG4Video{
+					Base: unit.Base{
+						NTP: time.Now(),
+						PTS: decodeTime(pts),
+					},
+					Frame: frame,
+				})
+				return nil
+			})
+
+		case *mpegts.CodecMPEG1Video:
+			medi = &description.Media{
+				Type:    description.MediaTypeVideo,
+				Formats: []format.Format{&format.MPEG1Video{}},
+			}
+
+			r.OnDataMPEGxVideo(track, func(pts int64, frame []byte) error {
+				stream.WriteUnit(medi, medi.Formats[0], &unit.MPEG1Video{
+					Base: unit.Base{
+						NTP: time.Now(),
+						PTS: decodeTime(pts),
+					},
+					Frame: frame,
+				})
+				return nil
+			})
+
 		case *mpegts.CodecOpus:
 			medi = &description.Media{
 				Type: description.MediaTypeAudio,

internal/formatprocessor/h264.go

@@ -13,22 +13,22 @@ import (
 )
 
 // extract SPS and PPS without decoding RTP packets
-func rtpH264ExtractSPSPPS(pkt *rtp.Packet) ([]byte, []byte) {
-	if len(pkt.Payload) < 1 {
+func rtpH264ExtractSPSPPS(payload []byte) ([]byte, []byte) {
+	if len(payload) < 1 {
 		return nil, nil
 	}
 
-	typ := h264.NALUType(pkt.Payload[0] & 0x1F)
+	typ := h264.NALUType(payload[0] & 0x1F)
 
 	switch typ {
 	case h264.NALUTypeSPS:
-		return pkt.Payload, nil
+		return payload, nil
 
 	case h264.NALUTypePPS:
-		return nil, pkt.Payload
+		return nil, payload
 
 	case h264.NALUTypeSTAPA:
-		payload := pkt.Payload[1:]
+		payload := payload[1:]
 		var sps []byte
 		var pps []byte
 
@@ -111,19 +111,11 @@ func (t *formatProcessorH264) createEncoder(
 	return t.encoder.Init()
 }
 
-func (t *formatProcessorH264) updateTrackParametersFromRTPPacket(pkt *rtp.Packet) {
-	sps, pps := rtpH264ExtractSPSPPS(pkt)
-	update := false
-
-	if sps != nil && !bytes.Equal(sps, t.format.SPS) {
-		update = true
-	}
+func (t *formatProcessorH264) updateTrackParametersFromRTPPacket(payload []byte) {
+	sps, pps := rtpH264ExtractSPSPPS(payload)
 
-	if pps != nil && !bytes.Equal(pps, t.format.PPS) {
-		update = true
-	}
-
-	if update {
+	if (sps != nil && !bytes.Equal(sps, t.format.SPS)) ||
+		(pps != nil && !bytes.Equal(pps, t.format.PPS)) {
 		if sps == nil {
 			sps = t.format.SPS
 		}
@@ -257,7 +249,7 @@ func (t *formatProcessorH264) ProcessRTPPacket( //nolint:dupl
 		},
 	}
 
-	t.updateTrackParametersFromRTPPacket(pkt)
+	t.updateTrackParametersFromRTPPacket(pkt.Payload)
 
 	if t.encoder == nil {
 		// remove padding

internal/formatprocessor/h265.go

@@ -13,25 +13,25 @@ import (
 )
 
 // extract VPS, SPS and PPS without decoding RTP packets
-func rtpH265ExtractVPSSPSPPS(pkt *rtp.Packet) ([]byte, []byte, []byte) {
-	if len(pkt.Payload) < 2 {
+func rtpH265ExtractVPSSPSPPS(payload []byte) ([]byte, []byte, []byte) {
+	if len(payload) < 2 {
 		return nil, nil, nil
 	}
 
-	typ := h265.NALUType((pkt.Payload[0] >> 1) & 0b111111)
+	typ := h265.NALUType((payload[0] >> 1) & 0b111111)
 
 	switch typ {
 	case h265.NALUType_VPS_NUT:
-		return pkt.Payload, nil, nil
+		return payload, nil, nil
 
 	case h265.NALUType_SPS_NUT:
-		return nil, pkt.Payload, nil
+		return nil, payload, nil
 
 	case h265.NALUType_PPS_NUT:
-		return nil, nil, pkt.Payload
+		return nil, nil, payload
 
 	case h265.NALUType_AggregationUnit:
-		payload := pkt.Payload[2:]
+		payload := payload[2:]
 		var vps []byte
 		var sps []byte
 		var pps []byte
@@ -55,7 +55,7 @@ func rtpH265ExtractVPSSPSPPS(pkt *rtp.Packet) ([]byte, []byte, []byte) {
 			nalu := payload[:size]
 			payload = payload[size:]
 
-			typ = h265.NALUType((pkt.Payload[0] >> 1) & 0b111111)
+			typ = h265.NALUType((payload[0] >> 1) & 0b111111)
 
 			switch typ {
 			case h265.NALUType_VPS_NUT:
@@ -118,23 +118,12 @@ func (t *formatProcessorH265) createEncoder(
 	return t.encoder.Init()
 }
 
-func (t *formatProcessorH265) updateTrackParametersFromRTPPacket(pkt *rtp.Packet) {
-	vps, sps, pps := rtpH265ExtractVPSSPSPPS(pkt)
-	update := false
-
-	if vps != nil && !bytes.Equal(vps, t.format.VPS) {
-		update = true
-	}
-
-	if sps != nil && !bytes.Equal(sps, t.format.SPS) {
-		update = true
-	}
-
-	if pps != nil && !bytes.Equal(pps, t.format.PPS) {
-		update = true
-	}
+func (t *formatProcessorH265) updateTrackParametersFromRTPPacket(payload []byte) {
+	vps, sps, pps := rtpH265ExtractVPSSPSPPS(payload)
 
-	if update {
+	if (vps != nil && !bytes.Equal(vps, t.format.VPS)) ||
+		(sps != nil && !bytes.Equal(sps, t.format.SPS)) ||
+		(pps != nil && !bytes.Equal(pps, t.format.PPS)) {
 		if vps == nil {
 			vps = t.format.VPS
 		}
@@ -279,7 +268,7 @@ func (t *formatProcessorH265) ProcessRTPPacket( //nolint:dupl
 		},
 	}
 
-	t.updateTrackParametersFromRTPPacket(pkt)
+	t.updateTrackParametersFromRTPPacket(pkt.Payload)
 
 	if t.encoder == nil {
 		// remove padding