add MPEG-1/2 video decoder and encoder (#415)

README.md

@@ -109,7 +109,7 @@ In RTSP, media streams are routed between server and clients by using RTP packet
 |H265||[link](https://pkg.go.dev/github.com/bluenviron/gortsplib/v4/pkg/format#H265)|:heavy_check_mark:|
 |H264||[link](https://pkg.go.dev/github.com/bluenviron/gortsplib/v4/pkg/format#H264)|:heavy_check_mark:|
 |MPEG-4 Video (H263, Xvid)||[link](https://pkg.go.dev/github.com/bluenviron/gortsplib/v4/pkg/format#MPEG4VideoES)|:heavy_check_mark:|
-|MPEG-1/2 Video||[link](https://pkg.go.dev/github.com/bluenviron/gortsplib/v4/pkg/format#MPEG1Video)||
+|MPEG-1/2 Video||[link](https://pkg.go.dev/github.com/bluenviron/gortsplib/v4/pkg/format#MPEG1Video)|:heavy_check_mark:|
 |M-JPEG||[link](https://pkg.go.dev/github.com/bluenviron/gortsplib/v4/pkg/format#MJPEG)|:heavy_check_mark:|
 
 ### Audio
@@ -127,7 +127,7 @@ In RTSP, media streams are routed between server and clients by using RTP packet
 |G711 (PCMA, PCMU)||[link](https://pkg.go.dev/github.com/bluenviron/gortsplib/v4/pkg/format#G711)|:heavy_check_mark:|
 |LPCM||[link](https://pkg.go.dev/github.com/bluenviron/gortsplib/v4/pkg/format#LPCM)|:heavy_check_mark:|
 
-### Mixed
+### Other
 
 |format / codec|variant|documentation|encoder and decoder available|
 |--------------|-------|-------------|-----------------------------|

pkg/format/mpeg1_video.go

@@ -2,6 +2,8 @@ package format //nolint:dupl
 
 import (
 	"github.com/pion/rtp"
+
+	"github.com/bluenviron/gortsplib/v4/pkg/format/rtpmpeg1video"
 )
 
 // MPEG1Video is a RTP format for a MPEG-1/2 Video codec.
@@ -41,3 +43,27 @@ func (f *MPEG1Video) FMTP() map[string]string {
 func (f *MPEG1Video) PTSEqualsDTS(*rtp.Packet) bool {
 	return true
 }
+
+// CreateDecoder creates a decoder able to decode the content of the format.
+func (f *MPEG1Video) CreateDecoder() (*rtpmpeg1video.Decoder, error) {
+	d := &rtpmpeg1video.Decoder{}
+
+	err := d.Init()
+	if err != nil {
+		return nil, err
+	}
+
+	return d, nil
+}
+
+// CreateEncoder creates an encoder able to encode the content of the format.
+func (f *MPEG1Video) CreateEncoder() (*rtpmpeg1video.Encoder, error) {
+	e := &rtpmpeg1video.Encoder{}
+
+	err := e.Init()
+	if err != nil {
+		return nil, err
+	}
+
+	return e, nil
+}

pkg/format/mpeg1_video_test.go

@@ -13,3 +13,21 @@ func TestMPEG1VideoAttributes(t *testing.T) {
 	require.Equal(t, 90000, format.ClockRate())
 	require.Equal(t, true, format.PTSEqualsDTS(&rtp.Packet{}))
 }
+
+func TestMPEG1VideoDecEncoder(t *testing.T) {
+	format := &MPEG1Video{}
+
+	enc, err := format.CreateEncoder()
+	require.NoError(t, err)
+
+	pkts, err := enc.Encode([]byte{1, 2, 3, 4})
+	require.NoError(t, err)
+	require.Equal(t, format.PayloadType(), pkts[0].PayloadType)
+
+	dec, err := format.CreateDecoder()
+	require.NoError(t, err)
+
+	byts, err := dec.Decode(pkts[0])
+	require.NoError(t, err)
+	require.Equal(t, []byte{1, 2, 3, 4}, byts)
+}

pkg/format/rtpmpeg1video/decoder.go

@@ -0,0 +1,150 @@
+package rtpmpeg1video
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/pion/rtp"
+)
+
+const (
+	maxFrameSize = 1 * 1024 * 1024
+)
+
+// ErrMorePacketsNeeded is returned when more packets are needed.
+var ErrMorePacketsNeeded = errors.New("need more packets")
+
+// ErrNonStartingPacketAndNoPrevious is returned when we received a non-starting
+// packet of a fragmented frame and we didn't received anything before.
+// It's normal to receive this when decoding a stream that has been already
+// running for some time.
+var ErrNonStartingPacketAndNoPrevious = errors.New(
+	"received a non-starting fragment without any previous starting fragment")
+
+func joinFragments(fragments [][]byte, size int) []byte {
+	ret := make([]byte, size)
+	n := 0
+	for _, p := range fragments {
+		n += copy(ret[n:], p)
+	}
+	return ret
+}
+
+// Decoder is a RTP/MPEG-1/2 Video decoder.
+// Specification: https://datatracker.ietf.org/doc/html/rfc2250
+type Decoder struct {
+	fragments     [][]byte
+	fragmentsSize int
+
+	sliceBuffer     [][]byte
+	sliceBufferSize int
+}
+
+// Init initializes the decoder.
+func (d *Decoder) Init() error {
+	return nil
+}
+
+func (d *Decoder) decodeSlice(pkt *rtp.Packet) ([]byte, error) {
+	if len(pkt.Payload) < 4 {
+		d.fragments = d.fragments[:0] // discard pending fragments
+		d.fragmentsSize = 0
+		return nil, fmt.Errorf("payload is too short")
+	}
+
+	mbz := pkt.Payload[0] >> 3
+	if mbz != 0 {
+		d.fragments = d.fragments[:0] // discard pending fragments
+		d.fragmentsSize = 0
+		return nil, fmt.Errorf("invalid MBZ: %v", mbz)
+	}
+
+	t := (pkt.Payload[0] >> 2) & 0x01
+	if t != 0 {
+		d.fragments = d.fragments[:0] // discard pending fragments
+		d.fragmentsSize = 0
+		return nil, fmt.Errorf("MPEG-2 video-specific header extension is not supported yet")
+	}
+
+	an := pkt.Payload[2] >> 7
+	if an != 0 {
+		d.fragments = d.fragments[:0] // discard pending fragments
+		d.fragmentsSize = 0
+		return nil, fmt.Errorf("AN not supported yet")
+	}
+
+	n := (pkt.Payload[2] >> 6) & 0x01
+	if n != 0 {
+		d.fragments = d.fragments[:0] // discard pending fragments
+		d.fragmentsSize = 0
+		return nil, fmt.Errorf("N not supported yet")
+	}
+
+	b := (pkt.Payload[2] >> 4) & 0x01
+	e := (pkt.Payload[2] >> 3) & 0x01
+
+	switch {
+	case b == 1 && e == 1:
+		return pkt.Payload[4:], nil
+
+	case b == 1:
+		d.fragments = d.fragments[:0] // discard pending fragments
+		d.fragments = append(d.fragments, pkt.Payload[4:])
+		d.fragmentsSize = len(pkt.Payload[4:])
+		return nil, ErrMorePacketsNeeded
+
+	case e == 1:
+		if d.fragmentsSize == 0 {
+			return nil, ErrNonStartingPacketAndNoPrevious
+		}
+
+		d.fragments = append(d.fragments, pkt.Payload[4:])
+		d.fragmentsSize += len(pkt.Payload[4:])
+
+		slice := joinFragments(d.fragments, d.fragmentsSize)
+		d.fragments = d.fragments[:0]
+		d.fragmentsSize = 0
+		return slice, nil
+
+	default:
+		if d.fragmentsSize == 0 {
+			return nil, ErrNonStartingPacketAndNoPrevious
+		}
+
+		d.fragments = append(d.fragments, pkt.Payload[4:])
+		d.fragmentsSize += len(pkt.Payload[4:])
+		return nil, ErrMorePacketsNeeded
+	}
+}
+
+// Decode decodes frames from a RTP packet.
+func (d *Decoder) Decode(pkt *rtp.Packet) ([]byte, error) {
+	slice, err := d.decodeSlice(pkt)
+	if err != nil {
+		return nil, err
+	}
+
+	addSize := len(slice)
+
+	if (d.sliceBufferSize + addSize) > maxFrameSize {
+		d.sliceBuffer = nil
+		d.sliceBufferSize = 0
+		return nil, fmt.Errorf("frame size (%d) is too big, maximum is %d",
+			d.sliceBufferSize+addSize, maxFrameSize)
+	}
+
+	d.sliceBuffer = append(d.sliceBuffer, slice)
+	d.sliceBufferSize += addSize
+
+	if !pkt.Marker {
+		return nil, ErrMorePacketsNeeded
+	}
+
+	ret := joinFragments(d.sliceBuffer, d.sliceBufferSize)
+
+	// do not reuse sliceBuffer to avoid race conditions
+	d.sliceBuffer = nil
+	d.sliceBufferSize = 0
+
+	return ret, nil
+}

pkg/format/rtpmpeg1video/decoder_test.go

@@ -0,0 +1,54 @@
+package rtpmpeg1video
+
+import (
+	"testing"
+
+	"github.com/pion/rtp"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDecode(t *testing.T) {
+	for _, ca := range cases {
+		t.Run(ca.name, func(t *testing.T) {
+			d := &Decoder{}
+			err := d.Init()
+			require.NoError(t, err)
+
+			var frame []byte
+
+			for _, pkt := range ca.pkts {
+				frame, err = d.Decode(pkt)
+			}
+
+			require.NoError(t, err)
+			require.Equal(t, ca.frame, frame)
+		})
+	}
+}
+
+func FuzzDecoder(f *testing.F) {
+	f.Fuzz(func(t *testing.T, a []byte, b []byte) {
+		d := &Decoder{}
+		d.Init() //nolint:errcheck
+
+		d.Decode(&rtp.Packet{ //nolint:errcheck
+			Header: rtp.Header{
+				Version:        2,
+				SequenceNumber: 17645,
+				Timestamp:      2289527317,
+				SSRC:           0x9dbb7812,
+			},
+			Payload: a,
+		})
+
+		d.Decode(&rtp.Packet{ //nolint:errcheck
+			Header: rtp.Header{
+				Version:        2,
+				SequenceNumber: 17646,
+				Timestamp:      2289527317,
+				SSRC:           0x9dbb7812,
+			},
+			Payload: b,
+		})
+	})
+}

pkg/format/rtpmpeg1video/encoder.go

@@ -0,0 +1,239 @@
+package rtpmpeg1video
+
+import (
+	"bytes"
+	"crypto/rand"
+
+	"github.com/pion/rtp"
+)
+
+const (
+	rtpVersion            = 2
+	defaultPayloadMaxSize = 1460 // 1500 (UDP MTU) - 20 (IP header) - 8 (UDP header) - 12 (RTP header)
+)
+
+func randUint32() (uint32, error) {
+	var b [4]byte
+	_, err := rand.Read(b[:])
+	if err != nil {
+		return 0, err
+	}
+	return uint32(b[0])<<24 | uint32(b[1])<<16 | uint32(b[2])<<8 | uint32(b[3]), nil
+}
+
+func lenAggregated(slices [][]byte, slice []byte) int {
+	l := 4 + len(slice)
+	for _, fr := range slices {
+		l += len(fr)
+	}
+	return l
+}
+
+func packetCount(avail, le int) int {
+	n := le / avail
+	if (le % avail) != 0 {
+		n++
+	}
+	return n
+}
+
+// Encoder is a RTP/MPEG-1/2 Video encoder.
+// Specification: https://datatracker.ietf.org/doc/html/rfc2250
+type Encoder struct {
+	// SSRC of packets (optional).
+	// It defaults to a random value.
+	SSRC *uint32
+
+	// initial sequence number of packets (optional).
+	// It defaults to a random value.
+	InitialSequenceNumber *uint16
+
+	// maximum size of packet payloads (optional).
+	// It defaults to 1460.
+	PayloadMaxSize int
+
+	sequenceNumber uint16
+}
+
+// Init initializes the encoder.
+func (e *Encoder) Init() error {
+	if e.SSRC == nil {
+		v, err := randUint32()
+		if err != nil {
+			return err
+		}
+		e.SSRC = &v
+	}
+	if e.InitialSequenceNumber == nil {
+		v, err := randUint32()
+		if err != nil {
+			return err
+		}
+		v2 := uint16(v)
+		e.InitialSequenceNumber = &v2
+	}
+	if e.PayloadMaxSize == 0 {
+		e.PayloadMaxSize = defaultPayloadMaxSize
+	}
+
+	e.sequenceNumber = *e.InitialSequenceNumber
+	return nil
+}
+
+// Encode encodes frames into RTP packets.
+func (e *Encoder) Encode(frame []byte) ([]*rtp.Packet, error) {
+	var rets []*rtp.Packet
+	var batch [][]byte
+
+	var temporalReference uint16
+	beginOfSequence := uint8(0)
+	var frameType uint8
+
+	for {
+		var slice []byte
+		end := bytes.Index(frame[4:], []byte{0, 0, 1})
+		if end >= 0 {
+			slice, frame = frame[:end+4], frame[end+4:]
+		} else {
+			slice, frame = frame, nil
+		}
+
+		if lenAggregated(batch, slice) <= e.PayloadMaxSize {
+			batch = append(batch, slice)
+		} else {
+			// write current batch
+			if batch != nil {
+				pkts, err := e.writeBatch(batch,
+					temporalReference,
+					beginOfSequence,
+					frameType)
+				if err != nil {
+					return nil, err
+				}
+				rets = append(rets, pkts...)
+				beginOfSequence = 0
+			}
+
+			// initialize new batch
+			batch = [][]byte{slice}
+		}
+
+		switch slice[3] {
+		case 0:
+			temporalReference = uint16(slice[4])<<2 | uint16(slice[5])>>6
+			frameType = (slice[5] >> 3) & 0b111
+
+		case 0xB8:
+			beginOfSequence = 1
+		}
+
+		if frame == nil {
+			break
+		}
+	}
+
+	// write last batch
+	pkts, err := e.writeBatch(batch,
+		temporalReference,
+		beginOfSequence,
+		frameType)
+	if err != nil {
+		return nil, err
+	}
+	rets = append(rets, pkts...)
+
+	rets[len(rets)-1].Marker = true
+
+	return rets, nil
+}
+
+func (e *Encoder) writeBatch(
+	slices [][]byte,
+	temporalReference uint16,
+	beginOfSequence uint8,
+	frameType uint8,
+) ([]*rtp.Packet, error) {
+	if len(slices) != 1 || lenAggregated(slices, nil) < e.PayloadMaxSize {
+		return e.writeAggregated(slices, temporalReference, beginOfSequence, frameType)
+	}
+
+	return e.writeFragmented(slices[0], temporalReference, beginOfSequence, frameType)
+}
+
+func (e *Encoder) writeFragmented(
+	slice []byte,
+	temporalReference uint16,
+	beginOfSequence uint8,
+	frameType uint8,
+) ([]*rtp.Packet, error) {
+	avail := e.PayloadMaxSize - 4
+	le := len(slice)
+	packetCount := packetCount(avail, le)
+
+	ret := make([]*rtp.Packet, packetCount)
+	le = avail
+	start := uint8(1)
+	end := uint8(0)
+
+	for i := range ret {
+		if i == (packetCount - 1) {
+			le = len(slice)
+			end = 1
+		}
+
+		payload := make([]byte, 4+le)
+		payload[0] = byte(temporalReference >> 8)
+		payload[1] = byte(temporalReference)
+		payload[2] = beginOfSequence<<5 | start<<4 | end<<3 | frameType
+		copy(payload[4:], slice)
+		slice = slice[le:]
+
+		ret[i] = &rtp.Packet{
+			Header: rtp.Header{
+				Version:        rtpVersion,
+				PayloadType:    32,
+				SequenceNumber: e.sequenceNumber,
+				SSRC:           *e.SSRC,
+			},
+			Payload: payload,
+		}
+
+		e.sequenceNumber++
+		start = 0
+		beginOfSequence = 0
+	}
+
+	return ret, nil
+}
+
+func (e *Encoder) writeAggregated(
+	slices [][]byte,
+	temporalReference uint16,
+	beginOfSequence uint8,
+	frameType uint8,
+) ([]*rtp.Packet, error) {
+	payload := make([]byte, lenAggregated(slices, nil))
+
+	payload[0] = byte(temporalReference >> 8)
+	payload[1] = byte(temporalReference)
+	payload[2] = beginOfSequence<<5 | 1<<4 | 1<<3 | frameType
+
+	n := 4
+	for _, slice := range slices {
+		n += copy(payload[n:], slice)
+	}
+
+	pkt := &rtp.Packet{
+		Header: rtp.Header{
+			Version:        rtpVersion,
+			PayloadType:    32,
+			SequenceNumber: e.sequenceNumber,
+			SSRC:           *e.SSRC,
+		},
+		Payload: payload,
+	}
+
+	e.sequenceNumber++
+
+	return []*rtp.Packet{pkt}, nil
+}

pkg/format/rtpmpeg1video/encoder_test.go

@@ -0,0 +1,179 @@
+package rtpmpeg1video
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/pion/rtp"
+	"github.com/stretchr/testify/require"
+)
+
+func uint16Ptr(v uint16) *uint16 {
+	return &v
+}
+
+func uint32Ptr(v uint32) *uint32 {
+	return &v
+}
+
+func mergeBytes(vals ...[]byte) []byte {
+	size := 0
+	for _, v := range vals {
+		size += len(v)
+	}
+	res := make([]byte, size)
+
+	pos := 0
+	for _, v := range vals {
+		n := copy(res[pos:], v)
+		pos += n
+	}
+
+	return res
+}
+
+var cases = []struct {
+	name  string
+	frame []byte
+	pkts  []*rtp.Packet
+}{
+	{
+		"single",
+		bytes.Repeat([]byte{1, 2, 3, 4}, 240/4),
+		[]*rtp.Packet{{
+			Header: rtp.Header{
+				Version:        2,
+				Marker:         true,
+				PayloadType:    32,
+				SequenceNumber: 17645,
+				SSRC:           0x9dbb7812,
+			},
+			Payload: mergeBytes(
+				[]byte{0, 0, 0x18, 0},
+				bytes.Repeat([]byte{1, 2, 3, 4}, 240/4),
+			),
+		}},
+	},
+	{
+		"aggregated",
+		mergeBytes(
+			[]byte{0, 0, 1},
+			bytes.Repeat([]byte{1, 2, 3, 4}, 128/4),
+			[]byte{0, 0, 1},
+			bytes.Repeat([]byte{5, 6, 7, 8}, 128/4),
+		),
+		[]*rtp.Packet{{
+			Header: rtp.Header{
+				Version:        2,
+				Marker:         true,
+				PayloadType:    32,
+				SequenceNumber: 17645,
+				SSRC:           0x9dbb7812,
+			},
+			Payload: mergeBytes(
+				[]byte{0, 0, 0x18, 0},
+				[]byte{0, 0, 1},
+				bytes.Repeat([]byte{1, 2, 3, 4}, 128/4),
+				[]byte{0, 0, 1},
+				bytes.Repeat([]byte{5, 6, 7, 8}, 128/4),
+			),
+		}},
+	},
+	{
+		"fragmented",
+		mergeBytes(
+			[]byte{0, 0, 1},
+			bytes.Repeat([]byte{1}, 2000),
+		),
+		[]*rtp.Packet{
+			{
+				Header: rtp.Header{
+					Version:        2,
+					Marker:         false,
+					PayloadType:    32,
+					SequenceNumber: 17645,
+					SSRC:           0x9dbb7812,
+				},
+				Payload: mergeBytes(
+					[]byte{0, 0, 0x10, 0},
+					[]byte{0, 0, 1},
+					bytes.Repeat([]byte{1}, 1453),
+				),
+			},
+			{
+				Header: rtp.Header{
+					Version:        2,
+					Marker:         true,
+					PayloadType:    32,
+					SequenceNumber: 17646,
+					SSRC:           0x9dbb7812,
+				},
+				Payload: mergeBytes(
+					[]byte{0, 0, 0x08, 0},
+					bytes.Repeat([]byte{1}, 547),
+				),
+			},
+		},
+	},
+	{
+		"fragmented to the limit",
+		mergeBytes(
+			[]byte{0, 0, 1},
+			bytes.Repeat([]byte{1}, 2909),
+		),
+		[]*rtp.Packet{
+			{
+				Header: rtp.Header{
+					Version:        2,
+					Marker:         false,
+					PayloadType:    32,
+					SequenceNumber: 17645,
+					SSRC:           2646308882,
+				},
+				Payload: mergeBytes(
+					[]byte{0, 0, 0x10, 0},
+					[]byte{0, 0, 1},
+					bytes.Repeat([]byte{1}, 1453),
+				),
+			},
+			{
+				Header: rtp.Header{
+					Version:        2,
+					Marker:         true,
+					PayloadType:    32,
+					SequenceNumber: 17646,
+					SSRC:           2646308882,
+				},
+				Payload: mergeBytes(
+					[]byte{0, 0, 0x08, 0},
+					bytes.Repeat([]byte{1}, 1456),
+				),
+			},
+		},
+	},
+}
+
+func TestEncode(t *testing.T) {
+	for _, ca := range cases {
+		t.Run(ca.name, func(t *testing.T) {
+			e := &Encoder{
+				SSRC:                  uint32Ptr(0x9dbb7812),
+				InitialSequenceNumber: uint16Ptr(0x44ed),
+			}
+			err := e.Init()
+			require.NoError(t, err)
+
+			pkts, err := e.Encode(ca.frame)
+			require.NoError(t, err)
+			require.Equal(t, ca.pkts, pkts)
+		})
+	}
+}
+
+func TestEncodeRandomInitialState(t *testing.T) {
+	e := &Encoder{}
+	err := e.Init()
+	require.NoError(t, err)
+	require.NotEqual(t, nil, e.SSRC)
+	require.NotEqual(t, nil, e.InitialSequenceNumber)
+}

pkg/format/rtpmpeg1video/rtpmpeg1video.go

@@ -0,0 +1,2 @@
+// Package rtpmpeg1video contains a RTP/MPEG-1/2 Video decoder and encoder.
+package rtpmpeg1video

pkg/format/rtpmpeg1video/testdata/fuzz/FuzzDecoder/4160bdda4af61380

@@ -0,0 +1,3 @@
+go test fuzz v1
+[]byte("\x000A0")
+[]byte("0")

pkg/format/rtpmpeg1video/testdata/fuzz/FuzzDecoder/42233dc5f72ca2b6

@@ -0,0 +1,3 @@
+go test fuzz v1
+[]byte("0")
+[]byte("\x000(0")

pkg/format/rtpmpeg1video/testdata/fuzz/FuzzDecoder/5ee1068973dc548f

@@ -0,0 +1,3 @@
+go test fuzz v1
+[]byte("0")
+[]byte("\x000 0")

pkg/format/rtpmpeg1video/testdata/fuzz/FuzzDecoder/63724012a19ffb49

@@ -0,0 +1,3 @@
+go test fuzz v1
+[]byte("0")
+[]byte("\x06000")

pkg/format/rtpmpeg1video/testdata/fuzz/FuzzDecoder/7c611a05bf27455e

@@ -0,0 +1,3 @@
+go test fuzz v1
+[]byte("\x000\x800")
+[]byte("0")

pkg/format/rtpmpeg1video/testdata/fuzz/FuzzDecoder/84ed65595ad05a58

@@ -0,0 +1,3 @@
+go test fuzz v1
+[]byte("0")
+[]byte("")

pkg/format/rtpmpeg1video/testdata/fuzz/FuzzDecoder/9e01191a999ecfc0

@@ -0,0 +1,3 @@
+go test fuzz v1
+[]byte("0")
+[]byte("0000")

pkg/format/rtpmpeg1video/testdata/fuzz/FuzzDecoder/d2d959045ed14be7

@@ -0,0 +1,3 @@
+go test fuzz v1
+[]byte("\x000000")
+[]byte("\x000 0")