Update RTSP server filters

cmd/api/ws.go

@@ -64,22 +64,14 @@ func (ctx *Context) Close() {
 
 func (ctx *Context) Write(msg interface{}) {
 	ctx.mu.Lock()
-	defer ctx.mu.Unlock()
 
-	var err error
-
-	switch msg := msg.(type) {
-	case *streamer.Message:
-		err = ctx.Conn.WriteJSON(msg)
-	case []byte:
-		err = ctx.Conn.WriteMessage(websocket.BinaryMessage, msg)
-	default:
-		return
+	if data, ok := msg.([]byte); ok {
+		_ = ctx.Conn.WriteMessage(websocket.BinaryMessage, data)
+	} else {
+		_ = ctx.Conn.WriteJSON(msg)
 	}
 
-	if err != nil {
-		//panic(err) // TODO: fix panic
-	}
+	ctx.mu.Unlock()
 }
 
 func (ctx *Context) Error(err error) {

cmd/mp4/mp4.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
 )
 
 func Init() {
@@ -35,35 +36,29 @@ func handlerKeyframe(w http.ResponseWriter, r *http.Request) {
 
 	exit := make(chan []byte)
 
-	cons := &mp4.Consumer{}
+	cons := &mp4.Keyframe{}
 	cons.Listen(func(msg interface{}) {
-		switch msg := msg.(type) {
-		case []byte:
-			exit <- msg
+		if data, ok := msg.([]byte); ok && exit != nil {
+			exit <- data
+			exit = nil
 		}
 	})
 
 	if err := stream.AddConsumer(cons); err != nil {
-		log.Error().Err(err).Msg("[api.keyframe] add consumer")
+		log.Error().Err(err).Caller().Send()
 		return
 	}
 
-	defer stream.RemoveConsumer(cons)
+	data := <-exit
 
-	w.Header().Set("Content-Type", cons.MimeType())
-
-	data, err := cons.Init()
-	if err != nil {
-		log.Error().Err(err).Msg("[api.keyframe] init")
-		return
-	}
-	data = append(data, <-exit...)
+	stream.RemoveConsumer(cons)
 
 	// Apple Safari won't show frame without length
 	w.Header().Set("Content-Length", strconv.Itoa(len(data)))
+	w.Header().Set("Content-Type", cons.MimeType)
 
 	if _, err := w.Write(data); err != nil {
-		log.Error().Err(err).Msg("[api.keyframe] add consumer")
+		log.Error().Err(err).Caller().Send()
 	}
 }
 
@@ -80,20 +75,20 @@ func handlerMP4(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	exit := make(chan struct{})
+	exit := make(chan error)
 
 	cons := &mp4.Consumer{}
 	cons.Listen(func(msg interface{}) {
-		switch msg := msg.(type) {
-		case []byte:
-			if _, err := w.Write(msg); err != nil {
-				exit <- struct{}{}
+		if data, ok := msg.([]byte); ok {
+			if _, err := w.Write(data); err != nil && exit != nil {
+				exit <- err
+				exit = nil
 			}
 		}
 	})
 
 	if err := stream.AddConsumer(cons); err != nil {
-		log.Error().Err(err).Msg("[api.mp4] add consumer")
+		log.Error().Err(err).Caller().Send()
 		return
 	}
 
@@ -103,18 +98,36 @@ func handlerMP4(w http.ResponseWriter, r *http.Request) {
 
 	data, err := cons.Init()
 	if err != nil {
-		log.Error().Err(err).Msg("[api.mp4] init")
+		log.Error().Err(err).Caller().Send()
 		return
 	}
 
 	if _, err = w.Write(data); err != nil {
-		log.Error().Err(err).Msg("[api.mp4] write")
+		log.Error().Err(err).Caller().Send()
 		return
 	}
 
-	<-exit
+	cons.Start()
 
-	log.Trace().Msg("[api.mp4] close")
+	var duration *time.Timer
+	if s := r.URL.Query().Get("duration"); s != "" {
+		if i, _ := strconv.Atoi(s); i > 0 {
+			duration = time.AfterFunc(time.Second*time.Duration(i), func() {
+				if exit != nil {
+					exit <- nil
+					exit = nil
+				}
+			})
+		}
+	}
+
+	err = <-exit
+
+	log.Trace().Err(err).Caller().Send()
+
+	if duration != nil {
+		duration.Stop()
+	}
 }
 
 func isChromeFirst(w http.ResponseWriter, r *http.Request) bool {

cmd/mp4/mse.go

@@ -9,6 +9,8 @@ import (
 
 const MsgTypeMSE = "mse" // fMP4
 
+const packetSize = 8192
+
 func handlerWS(ctx *api.Context, msg *streamer.Message) {
 	src := ctx.Request.URL.Query().Get("src")
 	stream := streams.GetOrNew(src)
@@ -21,14 +23,17 @@ func handlerWS(ctx *api.Context, msg *streamer.Message) {
 	cons.RemoteAddr = ctx.Request.RemoteAddr
 
 	cons.Listen(func(msg interface{}) {
-		switch msg.(type) {
-		case *streamer.Message, []byte:
-			ctx.Write(msg)
+		if data, ok := msg.([]byte); ok {
+			for len(data) > packetSize {
+				ctx.Write(data[:packetSize])
+				data = data[packetSize:]
+			}
+			ctx.Write(data)
 		}
 	})
 
 	if err := stream.AddConsumer(cons); err != nil {
-		log.Warn().Err(err).Msg("[api.mse] add consumer")
+		log.Warn().Err(err).Caller().Send()
 		ctx.Error(err)
 		return
 	}
@@ -37,16 +42,16 @@ func handlerWS(ctx *api.Context, msg *streamer.Message) {
 		stream.RemoveConsumer(cons)
 	})
 
-	ctx.Write(&streamer.Message{
-		Type: MsgTypeMSE, Value: cons.MimeType(),
-	})
+	ctx.Write(&streamer.Message{Type: MsgTypeMSE, Value: cons.MimeType()})
 
 	data, err := cons.Init()
 	if err != nil {
-		log.Warn().Err(err).Msg("[api.mse] init")
+		log.Warn().Err(err).Caller().Send()
 		ctx.Error(err)
 		return
 	}
 
 	ctx.Write(data)
+
+	cons.Start()
 }

cmd/rtsp/rtsp.go

@@ -209,16 +209,27 @@ func initMedias(conn *rtsp.Conn) {
 	for key, value := range conn.URL.Query() {
 		switch key {
 		case streamer.KindVideo, streamer.KindAudio:
-			for _, value := range value {
+			for _, name := range value {
+				name = strings.ToUpper(name)
+
+				// check aliases
+				switch name {
+				case "COPY":
+					name = "" // pass empty codecs list
+				case "MJPEG":
+					name = streamer.CodecJPEG
+				case "AAC":
+					name = streamer.CodecAAC
+				}
+
 				media := &streamer.Media{
 					Kind: key, Direction: streamer.DirectionRecvonly,
 				}
 
-				switch value {
-				case "", "copy": // pass empty codecs list
-				default:
-					codec := streamer.NewCodec(value)
-					media.Codecs = append(media.Codecs, codec)
+				// empty codecs match all codecs
+				if name != "" {
+					// empty clock rate and channels match any values
+					media.Codecs = []*streamer.Codec{{Name: name}}
 				}
 
 				conn.Medias = append(conn.Medias, media)

cmd/streams/producer.go

@@ -65,23 +65,28 @@ func (p *Producer) GetTrack(media *streamer.Media, codec *streamer.Codec) *strea
 		return nil
 	}
 
+	for _, track := range p.tracks {
+		if track.Codec == codec {
+			return track
+		}
+	}
+
+	// can't get new tracks after start
+	if p.state == stateStart {
+		return nil
+	}
+
 	track := p.element.GetTrack(media, codec)
 	if track == nil {
 		return nil
 	}
 
-	for _, t := range p.tracks {
-		if track == t {
-			return track
-		}
-	}
+	p.tracks = append(p.tracks, track)
 
 	if p.state == stateMedias {
 		p.state = stateTracks
 	}
 
-	p.tracks = append(p.tracks, track)
-
 	return track
 }
 

pkg/aac/rtp.go

@@ -0,0 +1,57 @@
+package aac
+
+import (
+	"encoding/binary"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/pion/rtp"
+)
+
+const RTPPacketVersionAAC = 0
+
+func RTPDepay(track *streamer.Track) streamer.WrapperFunc {
+	return func(push streamer.WriterFunc) streamer.WriterFunc {
+		return func(packet *rtp.Packet) error {
+			// support ONLY 2 bytes header size!
+			// streamtype=5;profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3;config=1408
+			headersSize := binary.BigEndian.Uint16(packet.Payload) >> 3
+
+			//log.Printf("[RTP/AAC] units: %d, size: %4d, ts: %10d, %t", headersSize/2, len(packet.Payload), packet.Timestamp, packet.Marker)
+
+			clone := *packet
+			clone.Version = RTPPacketVersionAAC
+			clone.Payload = packet.Payload[2+headersSize:]
+			return push(&clone)
+		}
+	}
+}
+
+func RTPPay(mtu uint16) streamer.WrapperFunc {
+	sequencer := rtp.NewRandomSequencer()
+
+	return func(push streamer.WriterFunc) streamer.WriterFunc {
+		return func(packet *rtp.Packet) error {
+			if packet.Version != RTPPacketVersionAAC {
+				return push(packet)
+			}
+
+			// support ONLY one unit in payload
+			size := uint16(len(packet.Payload))
+			// 2 bytes header size + 2 bytes first payload size
+			payload := make([]byte, 2+2+size)
+			payload[1] = 16 // header size in bits
+			binary.BigEndian.PutUint16(payload[2:], size<<3)
+			copy(payload[4:], packet.Payload)
+
+			clone := rtp.Packet{
+				Header: rtp.Header{
+					Version:        2,
+					Marker:         true,
+					SequenceNumber: sequencer.NextSequenceNumber(),
+					Timestamp:      packet.Timestamp,
+				},
+				Payload: payload,
+			}
+			return push(&clone)
+		}
+	}
+}

pkg/h264/avc.go

@@ -6,12 +6,6 @@ import (
 	"github.com/pion/rtp"
 )
 
-const PayloadTypeAVC = 255
-
-func IsAVC(codec *streamer.Codec) bool {
-	return codec.PayloadType == PayloadTypeAVC
-}
-
 func EncodeAVC(nals ...[]byte) (avc []byte) {
 	var i, n int
 

pkg/h264/rtp.go

@@ -87,29 +87,28 @@ func RTPPay(mtu uint16) streamer.WrapperFunc {
 
 	return func(push streamer.WriterFunc) streamer.WriterFunc {
 		return func(packet *rtp.Packet) error {
-			if packet.Version == RTPPacketVersionAVC {
-				payloads := payloader.Payload(mtu, packet.Payload)
-				last := len(payloads) - 1
-				for i, payload := range payloads {
-					clone := rtp.Packet{
-						Header: rtp.Header{
-							Version: 2,
-							Marker:  i == last,
-							//PayloadType:    packet.PayloadType,
-							SequenceNumber: sequencer.NextSequenceNumber(),
-							Timestamp:      packet.Timestamp,
-							//SSRC:           packet.SSRC,
-						},
-						Payload: payload,
-					}
-					if err := push(&clone); err != nil {
-						return err
-					}
-				}
-				return nil
+			if packet.Version != RTPPacketVersionAVC {
+				return push(packet)
 			}
 
-			return push(packet)
+			payloads := payloader.Payload(mtu, packet.Payload)
+			last := len(payloads) - 1
+			for i, payload := range payloads {
+				clone := rtp.Packet{
+					Header: rtp.Header{
+						Version:        2,
+						Marker:         i == last,
+						SequenceNumber: sequencer.NextSequenceNumber(),
+						Timestamp:      packet.Timestamp,
+					},
+					Payload: payload,
+				}
+				if err := push(&clone); err != nil {
+					return err
+				}
+			}
+
+			return nil
 		}
 	}
 }

pkg/ivideon/client.go

@@ -6,7 +6,6 @@ import (
 	"encoding/binary"
 	"encoding/json"
 	"fmt"
-	"github.com/AlexxIT/go2rtc/pkg/h264"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/deepch/vdk/codec/h264parser"
 	"github.com/deepch/vdk/format/fmp4/fmp4io"
@@ -162,9 +161,12 @@ func (c *Client) getTracks() error {
 					continue
 				}
 
-				codec := streamer.NewCodec(streamer.CodecH264)
-				codec.FmtpLine = "profile-level-id=" + msg.CodecString[i+1:]
-				codec.PayloadType = h264.PayloadTypeAVC
+				codec := &streamer.Codec{
+					Name:        streamer.CodecH264,
+					ClockRate:   90000,
+					FmtpLine:    "profile-level-id=" + msg.CodecString[i+1:],
+					PayloadType: streamer.PayloadTypeMP4,
+				}
 
 				i = bytes.Index(msg.Data, []byte("avcC")) - 4
 				if i < 0 {

pkg/mp4/const.go

@@ -3,6 +3,8 @@ package mp4
 import (
 	"encoding/binary"
 	"github.com/deepch/vdk/format/mp4/mp4io"
+	"github.com/deepch/vdk/format/mp4f"
+	"github.com/deepch/vdk/format/mp4f/mp4fio"
 	"time"
 )
 
@@ -37,25 +39,17 @@ func MOOV() *mp4io.Movie {
 			SelectionDuration: time0,
 			CurrentTime:       time0,
 		},
-		MovieExtend: &mp4io.MovieExtend{
-			Tracks: []*mp4io.TrackExtend{
-				{
-					TrackId:               1,
-					DefaultSampleDescIdx:  1,
-					DefaultSampleDuration: 40,
-				},
-			},
-		},
+		MovieExtend: &mp4io.MovieExtend{},
 	}
 }
 
-func TRAK() *mp4io.Track {
+func TRAK(id int) *mp4io.Track {
 	return &mp4io.Track{
 		// trak > tkhd
 		Header: &mp4io.TrackHeader{
-			TrackId:    int32(1), // change me
-			Flags:      0x0007,   // 7 ENABLED IN-MOVIE IN-PREVIEW
-			Duration:   0,        // OK
+			TrackId:    int32(id),
+			Flags:      0x0007, // 7 ENABLED IN-MOVIE IN-PREVIEW
+			Duration:   0,      // OK
 			Matrix:     matrix,
 			CreateTime: time0,
 			ModifyTime: time0,
@@ -92,3 +86,15 @@ func TRAK() *mp4io.Track {
 		},
 	}
 }
+
+func ESDS(conf []byte) *mp4f.FDummy {
+	esds := &mp4fio.ElemStreamDesc{DecConfig: conf}
+
+	b := make([]byte, esds.Len())
+	esds.Marshal(b)
+
+	return &mp4f.FDummy{
+		Data: b,
+		Tag_: mp4io.Tag(uint32(mp4io.ESDS)),
+	}
+}

pkg/mp4/consumer.go

@@ -2,7 +2,7 @@ package mp4
 
 import (
 	"encoding/json"
-	"fmt"
+	"github.com/AlexxIT/go2rtc/pkg/aac"
 	"github.com/AlexxIT/go2rtc/pkg/h264"
 	"github.com/AlexxIT/go2rtc/pkg/h265"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
@@ -28,44 +28,37 @@ func (c *Consumer) GetMedias() []*streamer.Media {
 			Kind:      streamer.KindVideo,
 			Direction: streamer.DirectionRecvonly,
 			Codecs: []*streamer.Codec{
-				{Name: streamer.CodecH264, ClockRate: 90000},
-				{Name: streamer.CodecH265, ClockRate: 90000},
+				{Name: streamer.CodecH264},
+				{Name: streamer.CodecH265},
+			},
+		},
+		{
+			Kind:      streamer.KindAudio,
+			Direction: streamer.DirectionRecvonly,
+			Codecs: []*streamer.Codec{
+				{Name: streamer.CodecAAC},
 			},
 		},
-		//{
-		//	Kind:      streamer.KindAudio,
-		//	Direction: streamer.DirectionRecvonly,
-		//	Codecs: []*streamer.Codec{
-		//		{Name: streamer.CodecAAC, ClockRate: 16000},
-		//	},
-		//},
 	}
 }
 
 func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *streamer.Track {
+	trackID := byte(len(c.codecs))
+	c.codecs = append(c.codecs, track.Codec)
+
 	codec := track.Codec
 	switch codec.Name {
 	case streamer.CodecH264:
-		c.codecs = append(c.codecs, track.Codec)
-
 		push := func(packet *rtp.Packet) error {
 			if packet.Version != h264.RTPPacketVersionAVC {
 				return nil
 			}
 
-			if c.muxer == nil {
+			if !c.start {
 				return nil
 			}
 
-			if !c.start {
-				if h264.IsKeyframe(packet.Payload) {
-					c.start = true
-				} else {
-					return nil
-				}
-			}
-
-			buf := c.muxer.Marshal(packet)
+			buf := c.muxer.Marshal(trackID, packet)
 			c.send += len(buf)
 			c.Fire(buf)
 
@@ -73,7 +66,7 @@ func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *strea
 		}
 
 		var wrapper streamer.WrapperFunc
-		if h264.IsAVC(codec) {
+		if codec.IsMP4() {
 			wrapper = h264.RepairAVC(track)
 		} else {
 			wrapper = h264.RTPDepay(track)
@@ -83,39 +76,51 @@ func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *strea
 		return track.Bind(push)
 
 	case streamer.CodecH265:
-		c.codecs = append(c.codecs, track.Codec)
-
 		push := func(packet *rtp.Packet) error {
 			if packet.Version != h264.RTPPacketVersionAVC {
 				return nil
 			}
 
 			if !c.start {
-				if h265.IsKeyframe(packet.Payload) {
-					c.start = true
-				} else {
-					return nil
-				}
+				return nil
 			}
 
-			buf := c.muxer.Marshal(packet)
+			buf := c.muxer.Marshal(trackID, packet)
 			c.send += len(buf)
 			c.Fire(buf)
 
 			return nil
 		}
 
-		if !h264.IsAVC(codec) {
+		if !codec.IsMP4() {
 			wrapper := h265.RTPDepay(track)
 			push = wrapper(push)
 		}
 
+		return track.Bind(push)
+
+	case streamer.CodecAAC:
+		push := func(packet *rtp.Packet) error {
+			if !c.start {
+				return nil
+			}
+
+			buf := c.muxer.Marshal(trackID, packet)
+			c.send += len(buf)
+			c.Fire(buf)
+
+			return nil
+		}
+
+		if !codec.IsMP4() {
+			wrapper := aac.RTPDepay(track)
+			push = wrapper(push)
+		}
+
 		return track.Bind(push)
 	}
 
-	fmt.Printf("[rtmp] unsupported codec: %+v\n", track.Codec)
-
-	return nil
+	panic("unsupported codec")
 }
 
 func (c *Consumer) MimeType() string {
@@ -123,12 +128,14 @@ func (c *Consumer) MimeType() string {
 }
 
 func (c *Consumer) Init() ([]byte, error) {
-	if c.muxer == nil {
-		c.muxer = &Muxer{}
-	}
+	c.muxer = &Muxer{}
 	return c.muxer.GetInit(c.codecs)
 }
 
+func (c *Consumer) Start() {
+	c.start = true
+}
+
 //
 
 func (c *Consumer) MarshalJSON() ([]byte, error) {

pkg/mp4/keyframe.go

@@ -0,0 +1,85 @@
+package mp4
+
+import (
+	"github.com/AlexxIT/go2rtc/pkg/h264"
+	"github.com/AlexxIT/go2rtc/pkg/h265"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/pion/rtp"
+)
+
+type Keyframe struct {
+	streamer.Element
+
+	MimeType string
+}
+
+func (c *Keyframe) GetMedias() []*streamer.Media {
+	return []*streamer.Media{
+		{
+			Kind:      streamer.KindVideo,
+			Direction: streamer.DirectionRecvonly,
+			Codecs: []*streamer.Codec{
+				{Name: streamer.CodecH264},
+				{Name: streamer.CodecH265},
+			},
+		},
+	}
+}
+
+func (c *Keyframe) AddTrack(media *streamer.Media, track *streamer.Track) *streamer.Track {
+	muxer := &Muxer{}
+
+	codecs := []*streamer.Codec{track.Codec}
+
+	init, err := muxer.GetInit(codecs)
+	if err != nil {
+		return nil
+	}
+
+	c.MimeType = muxer.MimeType(codecs)
+
+	switch track.Codec.Name {
+	case streamer.CodecH264:
+		push := func(packet *rtp.Packet) error {
+			if !h264.IsKeyframe(packet.Payload) {
+				return nil
+			}
+
+			buf := muxer.Marshal(0, packet)
+			c.Fire(append(init, buf...))
+
+			return nil
+		}
+
+		var wrapper streamer.WrapperFunc
+		if track.Codec.IsMP4() {
+			wrapper = h264.RepairAVC(track)
+		} else {
+			wrapper = h264.RTPDepay(track)
+		}
+		push = wrapper(push)
+
+		return track.Bind(push)
+
+	case streamer.CodecH265:
+		push := func(packet *rtp.Packet) error {
+			if !h265.IsKeyframe(packet.Payload) {
+				return nil
+			}
+
+			buf := muxer.Marshal(0, packet)
+			c.Fire(append(init, buf...))
+
+			return nil
+		}
+
+		if !track.Codec.IsMP4() {
+			wrapper := h265.RTPDepay(track)
+			push = wrapper(push)
+		}
+
+		return track.Bind(push)
+	}
+
+	panic("unsupported codec")
+}

pkg/mp4/muxer.go

@@ -2,10 +2,13 @@ package mp4
 
 import (
 	"encoding/binary"
+	"encoding/hex"
 	"fmt"
 	"github.com/AlexxIT/go2rtc/pkg/h264"
 	"github.com/AlexxIT/go2rtc/pkg/h265"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/deepch/vdk/av"
+	"github.com/deepch/vdk/codec/aacparser"
 	"github.com/deepch/vdk/codec/h264parser"
 	"github.com/deepch/vdk/codec/h265parser"
 	"github.com/deepch/vdk/format/fmp4/fmp4io"
@@ -16,22 +19,28 @@ import (
 
 type Muxer struct {
 	fragIndex uint32
-	dts       uint64
-	pts       uint32
-	data      []byte
-	total     int
+	dts       []uint64
+	pts       []uint32
+	//data      []byte
+	//total     int
 }
 
 func (m *Muxer) MimeType(codecs []*streamer.Codec) string {
 	s := `video/mp4; codecs="`
 
-	for _, codec := range codecs {
+	for i, codec := range codecs {
+		if i > 0 {
+			s += ","
+		}
+
 		switch codec.Name {
 		case streamer.CodecH264:
 			s += "avc1." + h264.GetProfileLevelID(codec.FmtpLine)
 		case streamer.CodecH265:
 			// +Safari +Chrome +Edge -iOS15 -Android13
 			s += "hvc1.1.6.L93.B0" // hev1.1.6.L93.B0
+		case streamer.CodecAAC:
+			s += "mp4a.40.2"
 		}
 	}
 
@@ -41,7 +50,7 @@ func (m *Muxer) MimeType(codecs []*streamer.Codec) string {
 func (m *Muxer) GetInit(codecs []*streamer.Codec) ([]byte, error) {
 	moov := MOOV()
 
-	for _, codec := range codecs {
+	for i, codec := range codecs {
 		switch codec.Name {
 		case streamer.CodecH264:
 			sps, pps := h264.GetParameterSet(codec.FmtpLine)
@@ -59,11 +68,14 @@ func (m *Muxer) GetInit(codecs []*streamer.Codec) ([]byte, error) {
 			width := codecData.Width()
 			height := codecData.Height()
 
-			trak := TRAK()
-			trak.Media.Header.TimeScale = int32(codec.ClockRate)
+			trak := TRAK(i + 1)
 			trak.Header.TrackWidth = float64(width)
 			trak.Header.TrackHeight = float64(height)
-
+			trak.Media.Header.TimeScale = int32(codec.ClockRate)
+			trak.Media.Handler = &mp4io.HandlerRefer{
+				SubType: [4]byte{'v', 'i', 'd', 'e'},
+				Name:    []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 'm', 'a', 'i', 'n', 0},
+			}
 			trak.Media.Info.Video = &mp4io.VideoMediaInfo{
 				Flags: 0x000001,
 			}
@@ -81,11 +93,6 @@ func (m *Muxer) GetInit(codecs []*streamer.Codec) ([]byte, error) {
 				},
 			}
 
-			trak.Media.Handler = &mp4io.HandlerRefer{
-				SubType: [4]byte{'v', 'i', 'd', 'e'},
-				Name:    []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 'm', 'a', 'i', 'n', 0},
-			}
-
 			moov.Tracks = append(moov.Tracks, trak)
 
 		case streamer.CodecH265:
@@ -102,11 +109,14 @@ func (m *Muxer) GetInit(codecs []*streamer.Codec) ([]byte, error) {
 			width := codecData.Width()
 			height := codecData.Height()
 
-			trak := TRAK()
-			trak.Media.Header.TimeScale = int32(codec.ClockRate)
+			trak := TRAK(i + 1)
 			trak.Header.TrackWidth = float64(width)
 			trak.Header.TrackHeight = float64(height)
-
+			trak.Media.Header.TimeScale = int32(codec.ClockRate)
+			trak.Media.Handler = &mp4io.HandlerRefer{
+				SubType: [4]byte{'v', 'i', 'd', 'e'},
+				Name:    []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 'm', 'a', 'i', 'n', 0},
+			}
 			trak.Media.Info.Video = &mp4io.VideoMediaInfo{
 				Flags: 0x000001,
 			}
@@ -124,13 +134,52 @@ func (m *Muxer) GetInit(codecs []*streamer.Codec) ([]byte, error) {
 				},
 			}
 
+			moov.Tracks = append(moov.Tracks, trak)
+
+		case streamer.CodecAAC:
+			s := streamer.Between(codec.FmtpLine, "config=", ";")
+			b, err := hex.DecodeString(s)
+			if err != nil {
+				return nil, err
+			}
+
+			codecData, err := aacparser.ParseMPEG4AudioConfigBytes(b)
+			if err != nil {
+				return nil, err
+			}
+
+			trak := TRAK(i + 1)
+			trak.Header.AlternateGroup = 1
+			trak.Header.Duration = 0
+			trak.Header.Volume = 1
+			trak.Media.Header.TimeScale = int32(codec.ClockRate)
+
 			trak.Media.Handler = &mp4io.HandlerRefer{
-				SubType: [4]byte{'v', 'i', 'd', 'e'},
+				SubType: [4]byte{'s', 'o', 'u', 'n'},
 				Name:    []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 'm', 'a', 'i', 'n', 0},
 			}
+			trak.Media.Info.Sound = &mp4io.SoundMediaInfo{}
+
+			trak.Media.Info.Sample.SampleDesc.MP4ADesc = &mp4io.MP4ADesc{
+				DataRefIdx:       1,
+				NumberOfChannels: int16(codecData.ChannelLayout.Count()),
+				SampleSize:       int16(av.FLTP.BytesPerSample() * 4),
+				SampleRate:       float64(codecData.SampleRate),
+				Unknowns:         []mp4io.Atom{ESDS(b)},
+			}
 
 			moov.Tracks = append(moov.Tracks, trak)
 		}
+
+		trex := &mp4io.TrackExtend{
+			TrackId:               uint32(i + 1),
+			DefaultSampleDescIdx:  1,
+			DefaultSampleDuration: 0,
+		}
+		moov.MovieExtend.Tracks = append(moov.MovieExtend.Tracks, trex)
+
+		m.pts = append(m.pts, 0)
+		m.dts = append(m.dts, 0)
 	}
 
 	data := make([]byte, moov.Len())
@@ -139,14 +188,12 @@ func (m *Muxer) GetInit(codecs []*streamer.Codec) ([]byte, error) {
 	return append(FTYP(), data...), nil
 }
 
-func (m *Muxer) Rewind() {
-	m.dts = 0
-	m.pts = 0
-}
-
-func (m *Muxer) Marshal(packet *rtp.Packet) []byte {
-	trackID := uint8(1)
+//func (m *Muxer) Rewind() {
+//	m.dts = 0
+//	m.pts = 0
+//}
 
+func (m *Muxer) Marshal(trackID byte, packet *rtp.Packet) []byte {
 	run := &mp4fio.TrackFragRun{
 		Flags:            0x000b05,
 		FirstSampleFlags: uint32(fmp4io.SampleNoDependencies),
@@ -161,12 +208,12 @@ func (m *Muxer) Marshal(packet *rtp.Packet) []byte {
 		Tracks: []*mp4fio.TrackFrag{
 			{
 				Header: &mp4fio.TrackFragHeader{
-					Data: []byte{0x00, 0x02, 0x00, 0x20, 0x00, 0x00, 0x00, trackID, 0x01, 0x01, 0x00, 0x00},
+					Data: []byte{0x00, 0x02, 0x00, 0x20, 0x00, 0x00, 0x00, trackID + 1, 0x01, 0x01, 0x00, 0x00},
 				},
 				DecodeTime: &mp4fio.TrackFragDecodeTime{
 					Version: 1,
 					Flags:   0,
-					Time:    m.dts,
+					Time:    m.dts[trackID],
 				},
 				Run: run,
 			},
@@ -179,12 +226,12 @@ func (m *Muxer) Marshal(packet *rtp.Packet) []byte {
 	}
 
 	newTime := packet.Timestamp
-	if m.pts > 0 {
+	if m.pts[trackID] > 0 {
 		//m.dts += uint64(newTime - m.pts)
-		entry.Duration = newTime - m.pts
-		m.dts += uint64(entry.Duration)
+		entry.Duration = newTime - m.pts[trackID]
+		m.dts[trackID] += uint64(entry.Duration)
 	}
-	m.pts = newTime
+	m.pts[trackID] = newTime
 
 	// important before moof.Len()
 	run.Entries = append(run.Entries, entry)
@@ -204,7 +251,7 @@ func (m *Muxer) Marshal(packet *rtp.Packet) []byte {
 
 	m.fragIndex++
 
-	m.total += moofLen + mdatLen
+	//m.total += moofLen + mdatLen
 
 	return buf
 }

pkg/mp4f/consumer.go

@@ -0,0 +1,164 @@
+package mp4f
+
+import (
+	"encoding/json"
+	"github.com/AlexxIT/go2rtc/pkg/h264"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/deepch/vdk/av"
+	"github.com/deepch/vdk/codec/aacparser"
+	"github.com/deepch/vdk/codec/h264parser"
+	"github.com/deepch/vdk/format/mp4f"
+	"github.com/pion/rtp"
+	"time"
+)
+
+type Consumer struct {
+	streamer.Element
+
+	UserAgent  string
+	RemoteAddr string
+
+	muxer    *mp4f.Muxer
+	streams  []av.CodecData
+	mimeType string
+	start    bool
+
+	send int
+}
+
+func (c *Consumer) GetMedias() []*streamer.Media {
+	return []*streamer.Media{
+		{
+			Kind:      streamer.KindVideo,
+			Direction: streamer.DirectionRecvonly,
+			Codecs: []*streamer.Codec{
+				{Name: streamer.CodecH264, ClockRate: 90000},
+			},
+		},
+		{
+			Kind:      streamer.KindAudio,
+			Direction: streamer.DirectionRecvonly,
+			Codecs: []*streamer.Codec{
+				{Name: streamer.CodecAAC, ClockRate: 16000},
+			},
+		},
+	}
+}
+
+func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *streamer.Track {
+	codec := track.Codec
+	trackID := int8(len(c.streams))
+
+	switch codec.Name {
+	case streamer.CodecH264:
+		sps, pps := h264.GetParameterSet(codec.FmtpLine)
+		stream, err := h264parser.NewCodecDataFromSPSAndPPS(sps, pps)
+		if err != nil {
+			return nil
+		}
+
+		c.mimeType += "avc1." + h264.GetProfileLevelID(codec.FmtpLine)
+		c.streams = append(c.streams, stream)
+
+		pkt := av.Packet{Idx: trackID, CompositionTime: time.Millisecond}
+
+		ts2time := time.Second / time.Duration(codec.ClockRate)
+
+		push := func(packet *rtp.Packet) error {
+			if packet.Version != h264.RTPPacketVersionAVC {
+				return nil
+			}
+
+			if !c.start {
+				return nil
+			}
+
+			pkt.Data = packet.Payload
+			newTime := time.Duration(packet.Timestamp) * ts2time
+			if pkt.Time > 0 {
+				pkt.Duration = newTime - pkt.Time
+			}
+			pkt.Time = newTime
+
+			ready, buf, _ := c.muxer.WritePacket(pkt, false)
+			if ready {
+				c.send += len(buf)
+				c.Fire(buf)
+			}
+
+			return nil
+		}
+
+		if !h264.IsAVC(codec) {
+			wrapper := h264.RTPDepay(track)
+			push = wrapper(push)
+		}
+
+		return track.Bind(push)
+
+	case streamer.CodecAAC:
+		stream, _ := aacparser.NewCodecDataFromMPEG4AudioConfigBytes([]byte{20, 8})
+
+		c.mimeType += ",mp4a.40.2"
+		c.streams = append(c.streams, stream)
+
+		pkt := av.Packet{Idx: trackID, CompositionTime: time.Millisecond}
+
+		ts2time := time.Second / time.Duration(codec.ClockRate)
+
+		push := func(packet *rtp.Packet) error {
+			if !c.start {
+				return nil
+			}
+
+			pkt.Data = packet.Payload
+			newTime := time.Duration(packet.Timestamp) * ts2time
+			if pkt.Time > 0 {
+				pkt.Duration = newTime - pkt.Time
+			}
+			pkt.Time = newTime
+
+			ready, buf, _ := c.muxer.WritePacket(pkt, false)
+			if ready {
+				c.send += len(buf)
+				c.Fire(buf)
+			}
+
+			return nil
+		}
+
+		return track.Bind(push)
+	}
+
+	panic("unsupported codec")
+}
+
+func (c *Consumer) MimeType() string {
+	return `video/mp4; codecs="` + c.mimeType + `"`
+}
+
+func (c *Consumer) Init() ([]byte, error) {
+	c.muxer = mp4f.NewMuxer(nil)
+	if err := c.muxer.WriteHeader(c.streams); err != nil {
+		return nil, err
+	}
+	_, data := c.muxer.GetInit(c.streams)
+	return data, nil
+}
+
+func (c *Consumer) Start()  {
+	c.start = true
+}
+
+//
+
+func (c *Consumer) MarshalJSON() ([]byte, error) {
+	v := map[string]interface{}{
+		"type":        "MSE server consumer",
+		"send":        c.send,
+		"remote_addr": c.RemoteAddr,
+		"user_agent":  c.UserAgent,
+	}
+
+	return json.Marshal(v)
+}

pkg/rtmp/client.go

@@ -4,7 +4,6 @@ import (
 	"encoding/base64"
 	"encoding/hex"
 	"fmt"
-	"github.com/AlexxIT/go2rtc/pkg/h264"
 	"github.com/AlexxIT/go2rtc/pkg/httpflv"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/deepch/vdk/av"
@@ -74,7 +73,7 @@ func (c *Client) Dial() (err error) {
 				Name:        streamer.CodecH264,
 				ClockRate:   90000,
 				FmtpLine:    fmtp,
-				PayloadType: h264.PayloadTypeAVC,
+				PayloadType: streamer.PayloadTypeMP4,
 			}
 
 			media := &streamer.Media{
@@ -93,17 +92,13 @@ func (c *Client) Dial() (err error) {
 			// TODO: fix support
 			cd := stream.(aacparser.CodecData)
 
-			// a=fmtp:97 streamtype=5;profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3;config=1588
-			fmtp := fmt.Sprintf(
-				"config=%s",
-				hex.EncodeToString(cd.ConfigBytes),
-			)
-
 			codec := &streamer.Codec{
 				Name:      streamer.CodecAAC,
 				ClockRate: uint32(cd.Config.SampleRate),
 				Channels:  uint16(cd.Config.ChannelConfig),
-				FmtpLine:  fmtp,
+				//  a=fmtp:97 streamtype=5;profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3;config=1588
+				FmtpLine:    "streamtype=5;profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3;config=" + hex.EncodeToString(cd.ConfigBytes),
+				PayloadType: streamer.PayloadTypeMP4,
 			}
 
 			media := &streamer.Media{

pkg/rtsp/conn.go

@@ -7,6 +7,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"github.com/AlexxIT/go2rtc/pkg/aac"
 	"github.com/AlexxIT/go2rtc/pkg/h264"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/AlexxIT/go2rtc/pkg/tcp"
@@ -764,9 +765,15 @@ func (c *Conn) bindTrack(
 		return nil
 	}
 
-	if h264.IsAVC(track.Codec) {
-		wrapper := h264.RTPPay(1500)
-		push = wrapper(push)
+	if track.Codec.IsMP4() {
+		switch track.Codec.Name {
+		case streamer.CodecH264:
+			wrapper := h264.RTPPay(1500)
+			push = wrapper(push)
+		case streamer.CodecAAC:
+			wrapper := aac.RTPPay(1500)
+			push = wrapper(push)
+		}
 	}
 
 	return track.Bind(push)

pkg/streamer/media.go

@@ -35,6 +35,8 @@ const (
 	CodecMPA  = "MPA" // payload: 14
 )
 
+const PayloadTypeMP4 byte = 255
+
 func GetKind(name string) string {
 	switch name {
 	case CodecH264, CodecH265, CodecVP8, CodecVP9, CodecAV1, CodecJPEG:
@@ -127,22 +129,6 @@ type Codec struct {
 	PayloadType uint8
 }
 
-func NewCodec(name string) *Codec {
-	name = strings.ToUpper(name)
-	switch name {
-	case CodecH264, CodecH265, CodecVP8, CodecVP9, CodecAV1, CodecJPEG:
-		return &Codec{Name: name, ClockRate: 90000}
-	case CodecPCMU, CodecPCMA:
-		return &Codec{Name: name, ClockRate: 8000}
-	case CodecOpus:
-		return &Codec{Name: name, ClockRate: 48000, Channels: 2}
-	case "MJPEG":
-		return &Codec{Name: CodecJPEG, ClockRate: 90000}
-	}
-
-	panic(fmt.Sprintf("unsupported codec: %s", name))
-}
-
 func (c *Codec) String() string {
 	s := fmt.Sprintf("%d %s/%d", c.PayloadType, c.Name, c.ClockRate)
 	if c.Channels > 0 {
@@ -151,6 +137,10 @@ func (c *Codec) String() string {
 	return s
 }
 
+func (c *Codec) IsMP4() bool {
+	return c.PayloadType == PayloadTypeMP4
+}
+
 func (c *Codec) Clone() *Codec {
 	clone := *c
 	return &clone

pkg/streamer/track.go

@@ -13,7 +13,7 @@ type Track struct {
 	Codec     *Codec
 	Direction string
 	sink      map[*Track]WriterFunc
-	sinkMu    sync.Mutex
+	sinkMu    sync.RWMutex
 }
 
 func (t *Track) String() string {
@@ -23,11 +23,11 @@ func (t *Track) String() string {
 }
 
 func (t *Track) WriteRTP(p *rtp.Packet) error {
-	t.sinkMu.Lock()
+	t.sinkMu.RLock()
 	for _, f := range t.sink {
 		_ = f(p)
 	}
-	t.sinkMu.Unlock()
+	t.sinkMu.RUnlock()
 	return nil
 }
 
@@ -59,7 +59,7 @@ func (t *Track) GetSink(from *Track) {
 }
 
 func (t *Track) HasSink() bool {
-	t.sinkMu.Lock()
-	defer t.sinkMu.Unlock()
+	t.sinkMu.RLock()
+	defer t.sinkMu.RUnlock()
 	return len(t.sink) > 0
 }

pkg/webrtc/consumer.go

@@ -57,7 +57,7 @@ func (c *Conn) AddTrack(media *streamer.Media, track *streamer.Track) *streamer.
 			wrapper := h264.RTPPay(1200)
 			push = wrapper(push)
 
-			if h264.IsAVC(codec) {
+			if codec.IsMP4() {
 				wrapper = h264.RepairAVC(track)
 			} else {
 				wrapper = h264.RTPDepay(track)

www/README.md

@@ -53,3 +53,5 @@ pc.ontrack = ev => {
 - https://www.webrtc-experiment.com/DetectRTC/
 - https://divtable.com/table-styler/
 - https://www.chromium.org/audio-video/
+- https://web.dev/i18n/en/fast-playback-with-preload/#manual_buffering
+- https://developer.mozilla.org/en-US/docs/Web/API/Media_Source_Extensions_API

www/mse.html

@@ -25,93 +25,75 @@
 <!-- muted is important for autoplay -->
 <video id="video" autoplay controls playsinline muted></video>
 <script>
-    const video = document.querySelector('#video');
-
     // support api_path
     const baseUrl = location.origin + location.pathname.substr(
         0, location.pathname.lastIndexOf("/")
     );
-    const ws = new WebSocket(`ws${baseUrl.substr(4)}/api/ws${location.search}`);
-    ws.binaryType = "arraybuffer";
+    const video = document.querySelector('#video');
 
-    let mediaSource;
+    function init() {
+        let mediaSource, sourceBuffer, queueBuffer = [];
 
-    ws.onopen = () => {
-        console.log("Start WS");
+        const ws = new WebSocket(`ws${baseUrl.substr(4)}/api/ws${location.search}`);
+        ws.binaryType = "arraybuffer";
 
-        // https://web.dev/i18n/en/fast-playback-with-preload/#manual_buffering
-        // https://developer.mozilla.org/en-US/docs/Web/API/Media_Source_Extensions_API
-        mediaSource = new MediaSource();
-        video.src = URL.createObjectURL(mediaSource);
-        mediaSource.onsourceopen = () => {
-            console.debug("mediaSource.onsourceopen");
-
-            mediaSource.onsourceopen = null;
-            URL.revokeObjectURL(video.src);
-            ws.send(JSON.stringify({"type": "mse"}));
+        ws.onopen = () => {
+            mediaSource = new MediaSource();
+            video.src = URL.createObjectURL(mediaSource);
+            mediaSource.onsourceopen = () => {
+                mediaSource.onsourceopen = null;
+                URL.revokeObjectURL(video.src);
+                ws.send(JSON.stringify({"type": "mse"}));
+            };
         };
-    };
 
-    let sourceBuffer, queueBuffer = [];
+        ws.onmessage = ev => {
+            if (typeof ev.data === 'string') {
+                const data = JSON.parse(ev.data);
+                console.debug("ws.onmessage", data);
 
-    ws.onmessage = ev => {
-        if (typeof ev.data === 'string') {
-            const data = JSON.parse(ev.data);
-            console.debug("ws.onmessage", data);
-
-            if (data.type === "mse") {
-                sourceBuffer = mediaSource.addSourceBuffer(data.value);
-                // important: segments supports TrackFragDecodeTime
-                // sequence supports only TrackFragRunEntry Duration
-                sourceBuffer.mode = "segments";
-                sourceBuffer.onupdateend = () => {
-                    if (!sourceBuffer.updating && queueBuffer.length > 0) {
-                        sourceBuffer.appendBuffer(queueBuffer.shift());
+                if (data.type === "mse") {
+                    sourceBuffer = mediaSource.addSourceBuffer(data.value);
+                    sourceBuffer.mode = "segments"; // segments or sequence
+                    sourceBuffer.onupdateend = () => {
+                        if (!sourceBuffer.updating && queueBuffer.length > 0) {
+                            try {
+                                sourceBuffer.appendBuffer(queueBuffer.shift());
+                            } catch (e) {
+                                // console.warn(e);
+                            }
+                        }
                     }
                 }
-            }
-        } else {
-            if (sourceBuffer.updating) {
-                queueBuffer.push(ev.data)
+            } else if (sourceBuffer.updating || queueBuffer.length > 0) {
+                queueBuffer.push(ev.data);
             } else {
-                sourceBuffer.appendBuffer(ev.data);
+                try {
+                    sourceBuffer.appendBuffer(ev.data);
+                } catch (e) {
+                    // console.warn(e);
+                }
+            }
+
+            if (video.seekable.length > 0) {
+                const delay = video.seekable.end(video.seekable.length - 1) - video.currentTime;
+                if (delay < 1) {
+                    video.playbackRate = 1;
+                } else if (delay > 10) {
+                    video.playbackRate = 10;
+                } else if (delay > 2) {
+                    video.playbackRate = Math.floor(delay);
+                }
             }
         }
+
+        video.onpause = () => {
+            ws.close();
+            setTimeout(init, 0);
+        }
     }
 
-    let offsetTime = 1, noWaiting = 0;
-
-    setInterval(() => {
-        if (video.paused || video.seekable.length === 0) return;
-
-        if (noWaiting < 0) {
-            offsetTime = Math.min(offsetTime * 1.1, 5);
-            console.debug("offset time up:", offsetTime);
-        } else if (noWaiting >= 30) {
-            noWaiting = 0;
-            offsetTime = Math.max(offsetTime * 0.9, 0.5);
-            console.debug("offset time down:", offsetTime);
-        }
-        noWaiting += 1;
-
-        const endTime = video.seekable.end(video.seekable.length - 1);
-        let playbackRate = (endTime - video.currentTime) / offsetTime;
-        if (playbackRate < 0.1) {
-            // video.currentTime = endTime - offsetTime;
-            playbackRate = 0.1;
-        } else if (playbackRate > 10) {
-            // video.currentTime = endTime - offsetTime;
-            playbackRate = 10;
-        }
-        // https://github.com/GoogleChrome/developer.chrome.com/issues/135
-        video.playbackRate = playbackRate;
-    }, 1000);
-
-    video.onwaiting = () => {
-        const endTime = video.seekable.end(video.seekable.length - 1);
-        video.currentTime = endTime - offsetTime;
-        noWaiting = -1;
-    }
+    init();
 </script>
 </body>
 </html>

www/webrtc.html

@@ -25,12 +25,12 @@
 <body>
 <video id="video" autoplay controls playsinline muted></video>
 <script>
+    const baseUrl = location.origin + location.pathname.substr(
+        0, location.pathname.lastIndexOf("/")
+    );
+
     function init(stream) {
         // support api_path
-        const baseUrl = location.origin + location.pathname.substr(
-            0, location.pathname.lastIndexOf("/")
-        );
-
         const ws = new WebSocket(`ws${baseUrl.substr(4)}/api/ws${location.search}`);
         ws.onopen = () => {
             console.debug('ws.onopen');