Rewrite MP4, HLS, MPEG-TS consumers

internal/api/ws/ws.go

@@ -1,15 +1,17 @@
 package ws
 
 import (
-	"github.com/AlexxIT/go2rtc/internal/api"
+	"io"
-	"github.com/AlexxIT/go2rtc/internal/app"
-	"github.com/gorilla/websocket"
-	"github.com/rs/zerolog/log"
 	"net/http"
 	"net/url"
 	"strings"
 	"sync"
 	"time"
+
+	"github.com/AlexxIT/go2rtc/internal/api"
+	"github.com/AlexxIT/go2rtc/internal/app"
+	"github.com/gorilla/websocket"
+	"github.com/rs/zerolog/log"
 )
 
 func Init() {
@@ -101,13 +103,13 @@ func apiWS(w http.ResponseWriter, r *http.Request) {
 	}
 
 	tr := &Transport{Request: r}
-	tr.OnWrite(func(msg any) {
+	tr.OnWrite(func(msg any) error {
 		_ = ws.SetWriteDeadline(time.Now().Add(time.Second * 5))
 
 		if data, ok := msg.([]byte); ok {
-			_ = ws.WriteMessage(websocket.BinaryMessage, data)
+			return ws.WriteMessage(websocket.BinaryMessage, data)
 		} else {
-			_ = ws.WriteJSON(msg)
+			return ws.WriteJSON(msg)
 		}
 	})
 
@@ -147,11 +149,11 @@ type Transport struct {
 	wrmx   sync.Mutex
 
 	onChange func()
-	onWrite  func(msg any)
+	onWrite  func(msg any) error
 	onClose  []func()
 }
 
-func (t *Transport) OnWrite(f func(msg any)) {
+func (t *Transport) OnWrite(f func(msg any) error) {
 	t.mx.Lock()
 	if t.onChange != nil {
 		t.onChange()
@@ -162,7 +164,7 @@ func (t *Transport) OnWrite(f func(msg any)) {
 
 func (t *Transport) Write(msg any) {
 	t.wrmx.Lock()
-	t.onWrite(msg)
+	_ = t.onWrite(msg)
 	t.wrmx.Unlock()
 }
 
@@ -200,3 +202,20 @@ func (t *Transport) WithContext(f func(ctx map[any]any)) {
 	f(t.ctx)
 	t.mx.Unlock()
 }
+
+func (t *Transport) Writer() io.Writer {
+	return &writer{t: t}
+}
+
+type writer struct {
+	t *Transport
+}
+
+func (w *writer) Write(p []byte) (n int, err error) {
+	w.t.wrmx.Lock()
+	if err = w.t.onWrite(p); err == nil {
+		n = len(p)
+	}
+	w.t.wrmx.Unlock()
+	return
+}

internal/hls/hls.go

@@ -2,7 +2,6 @@ package hls
 
 import (
 	"net/http"
-	"strings"
 	"sync"
 	"time"
 
@@ -12,6 +11,7 @@ import (
 	"github.com/AlexxIT/go2rtc/internal/streams"
 	"github.com/AlexxIT/go2rtc/pkg/core"
 	"github.com/AlexxIT/go2rtc/pkg/mp4"
+	"github.com/AlexxIT/go2rtc/pkg/mpegts"
 	"github.com/AlexxIT/go2rtc/pkg/tcp"
 	"github.com/rs/zerolog"
 )
@@ -32,21 +32,12 @@ func Init() {
 	ws.HandleFunc("hls", handlerWSHLS)
 }
 
-type Consumer interface {
-	core.Consumer
-	Listen(f core.EventFunc)
-	Init() ([]byte, error)
-	MimeCodecs() string
-	Start()
-}
-
 var log zerolog.Logger
 
 const keepalive = 5 * time.Second
 
-var sessions = map[string]*Session{}
-
 // once I saw 404 on MP4 segment, so better to use mutex
+var sessions = map[string]*Session{}
 var sessionsMu sync.RWMutex
 
 func handlerStream(w http.ResponseWriter, r *http.Request) {
@@ -66,22 +57,22 @@ func handlerStream(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	var cons Consumer
+	var cons core.Consumer
 
 	// use fMP4 with codecs filter and TS without
 	medias := mp4.ParseQuery(r.URL.Query())
 	if medias != nil {
-		cons = &mp4.Consumer{
+		c := mp4.NewConsumer(medias)
-			Desc:       "HLS/HTTP",
+		c.Type = "HLS/fMP4 consumer"
-			RemoteAddr: tcp.RemoteAddr(r),
+		c.RemoteAddr = tcp.RemoteAddr(r)
-			UserAgent:  r.UserAgent(),
+		c.UserAgent = r.UserAgent()
-			Medias:     medias,
+		cons = c
-		}
 	} else {
-		//cons = &mpegts.Consumer{
+		c := mpegts.NewConsumer()
-		//	RemoteAddr: tcp.RemoteAddr(r),
+		c.Type = "HLS/TS consumer"
-		//	UserAgent:  r.UserAgent(),
+		c.RemoteAddr = tcp.RemoteAddr(r)
-		//}
+		c.UserAgent = r.UserAgent()
+		cons = c
 	}
 
 	if err := stream.AddConsumer(cons); err != nil {
@@ -89,63 +80,22 @@ func handlerStream(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	session := &Session{cons: cons}
+	session := NewSession(cons)
-
-	cons.Listen(func(msg any) {
-		if data, ok := msg.([]byte); ok {
-			session.mu.Lock()
-			session.buffer = append(session.buffer, data...)
-			session.mu.Unlock()
-		}
-	})
-
-	sid := core.RandString(8, 62)
-
 	session.alive = time.AfterFunc(keepalive, func() {
 		sessionsMu.Lock()
-		delete(sessions, sid)
+		delete(sessions, session.id)
 		sessionsMu.Unlock()
 
 		stream.RemoveConsumer(cons)
 	})
-	session.init, _ = cons.Init()
-
-	cons.Start()
-
-	// two segments important for Chromecast
-	if medias != nil {
-		session.template = `#EXTM3U
-#EXT-X-VERSION:6
-#EXT-X-TARGETDURATION:1
-#EXT-X-MEDIA-SEQUENCE:%d
-#EXT-X-MAP:URI="init.mp4?id=` + sid + `"
-#EXTINF:0.500,
-segment.m4s?id=` + sid + `&n=%d
-#EXTINF:0.500,
-segment.m4s?id=` + sid + `&n=%d`
-	} else {
-		session.template = `#EXTM3U
-#EXT-X-VERSION:3
-#EXT-X-TARGETDURATION:1
-#EXT-X-MEDIA-SEQUENCE:%d
-#EXTINF:0.500,
-segment.ts?id=` + sid + `&n=%d
-#EXTINF:0.500,
-segment.ts?id=` + sid + `&n=%d`
-	}
 
 	sessionsMu.Lock()
-	sessions[sid] = session
+	sessions[session.id] = session
 	sessionsMu.Unlock()
 
-	codecs := strings.Replace(cons.MimeCodecs(), mp4.MimeFlac, "fLaC", 1)
+	go session.Run()
 
-	// bandwidth important for Safari, codecs useful for smooth playback
+	if _, err := w.Write(session.Main()); err != nil {
-	data := []byte(`#EXTM3U
-#EXT-X-STREAM-INF:BANDWIDTH=192000,CODECS="` + codecs + `"
-hls/playlist.m3u8?id=` + sid)
-
-	if _, err := w.Write(data); err != nil {
 		log.Error().Err(err).Caller().Send()
 	}
 }
@@ -168,7 +118,7 @@ func handlerPlaylist(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if _, err := w.Write([]byte(session.Playlist())); err != nil {
+	if _, err := w.Write(session.Playlist()); err != nil {
 		log.Error().Err(err).Caller().Send()
 	}
 }
@@ -223,11 +173,8 @@ func handlerInit(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	data := session.init
+	data := session.Init()
-	session.init = nil
+	if data == nil {
-
-	session.segment0 = session.Segment()
-	if session.segment0 == nil {
 		log.Warn().Msgf("[hls] can't get init %s", r.URL.RawQuery)
 		http.NotFound(w, r)
 		return
@@ -260,14 +207,7 @@ func handlerSegmentMP4(w http.ResponseWriter, r *http.Request) {
 
 	session.alive.Reset(keepalive)
 
-	var data []byte
+	data := session.Segment()
-
-	if query.Get("n") != "0" {
-		data = session.Segment()
-	} else {
-		data = session.segment0
-	}
-
 	if data == nil {
 		log.Warn().Msgf("[hls] can't get segment %s", r.URL.RawQuery)
 		http.NotFound(w, r)

internal/hls/session.go

@@ -2,23 +2,105 @@ package hls
 
 import (
 	"fmt"
+	"io"
+	"strings"
 	"sync"
 	"time"
+
+	"github.com/AlexxIT/go2rtc/pkg/core"
+	"github.com/AlexxIT/go2rtc/pkg/mp4"
 )
 
 type Session struct {
-	cons     Consumer
+	cons     core.Consumer
+	id       string
 	template string
 	init     []byte
-	segment0 []byte
 	buffer   []byte
 	seq      int
 	alive    *time.Timer
 	mu       sync.Mutex
 }
 
-func (s *Session) Playlist() string {
+func NewSession(cons core.Consumer) *Session {
-	return fmt.Sprintf(s.template, s.seq, s.seq, s.seq+1)
+	s := &Session{
+		id:   core.RandString(8, 62),
+		cons: cons,
+	}
+
+	// two segments important for Chromecast
+	if _, ok := cons.(*mp4.Consumer); ok {
+		s.template = `#EXTM3U
+#EXT-X-VERSION:6
+#EXT-X-TARGETDURATION:1
+#EXT-X-MEDIA-SEQUENCE:%d
+#EXT-X-MAP:URI="init.mp4?id=` + s.id + `"
+#EXTINF:0.500,
+segment.m4s?id=` + s.id + `&n=%d
+#EXTINF:0.500,
+segment.m4s?id=` + s.id + `&n=%d`
+	} else {
+		s.template = `#EXTM3U
+#EXT-X-VERSION:3
+#EXT-X-TARGETDURATION:1
+#EXT-X-MEDIA-SEQUENCE:%d
+#EXTINF:0.500,
+segment.ts?id=` + s.id + `&n=%d
+#EXTINF:0.500,
+segment.ts?id=` + s.id + `&n=%d`
+	}
+
+	return s
+}
+
+func (s *Session) Write(p []byte) (n int, err error) {
+	s.mu.Lock()
+	if s.init == nil {
+		s.init = p
+	} else {
+		s.buffer = append(s.buffer, p...)
+	}
+	s.mu.Unlock()
+	return len(p), nil
+}
+
+func (s *Session) Run() {
+	_, _ = s.cons.(io.WriterTo).WriteTo(s)
+}
+
+func (s *Session) Main() []byte {
+	type withCodecs interface {
+		Codecs() []*core.Codec
+	}
+
+	codecs := mp4.MimeCodecs(s.cons.(withCodecs).Codecs())
+	codecs = strings.Replace(codecs, mp4.MimeFlac, "fLaC", 1)
+
+	// bandwidth important for Safari, codecs useful for smooth playback
+	return []byte(`#EXTM3U
+#EXT-X-STREAM-INF:BANDWIDTH=192000,CODECS="` + codecs + `"
+hls/playlist.m3u8?id=` + s.id)
+}
+
+func (s *Session) Playlist() []byte {
+	return []byte(fmt.Sprintf(s.template, s.seq, s.seq, s.seq+1))
+}
+
+func (s *Session) Init() (init []byte) {
+	for i := 0; i < 60 && init == nil; i++ {
+		if i > 0 {
+			time.Sleep(50 * time.Millisecond)
+		}
+
+		s.mu.Lock()
+		// return init only when have some buffer
+		if len(s.buffer) > 0 {
+			init = s.init
+		}
+		s.mu.Unlock()
+	}
+
+	return
 }
 
 func (s *Session) Segment() (segment []byte) {
@@ -30,8 +112,12 @@ func (s *Session) Segment() (segment []byte) {
 		s.mu.Lock()
 		if len(s.buffer) > 0 {
 			segment = s.buffer
-			// for TS important to start new segment with init
+			if _, ok := s.cons.(*mp4.Consumer); ok {
-			s.buffer = s.init
+				s.buffer = nil
+			} else {
+				// for TS important to start new segment with init
+				s.buffer = s.init
+			}
 			s.seq++
 		}
 		s.mu.Unlock()

internal/hls/ws.go

@@ -2,13 +2,11 @@ package hls
 
 import (
 	"errors"
-	"strings"
 	"time"
 
 	"github.com/AlexxIT/go2rtc/internal/api"
 	"github.com/AlexxIT/go2rtc/internal/api/ws"
 	"github.com/AlexxIT/go2rtc/internal/streams"
-	"github.com/AlexxIT/go2rtc/pkg/core"
 	"github.com/AlexxIT/go2rtc/pkg/mp4"
 	"github.com/AlexxIT/go2rtc/pkg/tcp"
 )
@@ -20,63 +18,37 @@ func handlerWSHLS(tr *ws.Transport, msg *ws.Message) error {
 	}
 
 	codecs := msg.String()
+	medias := mp4.ParseCodecs(codecs, true)
+	cons := mp4.NewConsumer(medias)
+	cons.Type = "HLS/fMP4 consumer"
+	cons.RemoteAddr = tcp.RemoteAddr(tr.Request)
+	cons.UserAgent = tr.Request.UserAgent()
 
 	log.Trace().Msgf("[hls] new ws consumer codecs=%s", codecs)
 
-	cons := &mp4.Consumer{
-		Desc:       "HLS/WebSocket",
-		RemoteAddr: tcp.RemoteAddr(tr.Request),
-		UserAgent:  tr.Request.UserAgent(),
-		Medias:     mp4.ParseCodecs(codecs, true),
-	}
-
 	if err := stream.AddConsumer(cons); err != nil {
 		log.Error().Err(err).Caller().Send()
 		return err
 	}
 
-	session := &Session{cons: cons}
+	session := NewSession(cons)
-
-	cons.Listen(func(msg any) {
-		if data, ok := msg.([]byte); ok {
-			session.mu.Lock()
-			session.buffer = append(session.buffer, data...)
-			session.mu.Unlock()
-		}
-	})
 
 	session.alive = time.AfterFunc(keepalive, func() {
+		sessionsMu.Lock()
+		delete(sessions, session.id)
+		sessionsMu.Unlock()
+
 		stream.RemoveConsumer(cons)
 	})
-	session.init, _ = cons.Init()
-
-	cons.Start()
-
-	sid := core.RandString(8, 62)
-
-	// two segments important for Chromecast
-	session.template = `#EXTM3U
-#EXT-X-VERSION:6
-#EXT-X-TARGETDURATION:1
-#EXT-X-MEDIA-SEQUENCE:%d
-#EXT-X-MAP:URI="init.mp4?id=` + sid + `"
-#EXTINF:0.500,
-segment.m4s?id=` + sid + `&n=%d
-#EXTINF:0.500,
-segment.m4s?id=` + sid + `&n=%d`
 
 	sessionsMu.Lock()
-	sessions[sid] = session
+	sessions[session.id] = session
 	sessionsMu.Unlock()
 
-	codecs = strings.Replace(cons.MimeCodecs(), mp4.MimeFlac, "fLaC", 1)
+	go session.Run()
 
-	// bandwidth important for Safari, codecs useful for smooth playback
+	main := session.Main()
-	data := `#EXTM3U
+	tr.Write(&ws.Message{Type: "hls", Value: string(main)})
-#EXT-X-STREAM-INF:BANDWIDTH=192000,CODECS="` + codecs + `"
-hls/playlist.m3u8?id=` + sid
-
-	tr.Write(&ws.Message{Type: "hls", Value: data})
 
 	return nil
 }

internal/mjpeg/init.go

@@ -180,8 +180,7 @@ func handlerWS(tr *ws.Transport, _ *ws.Message) error {
 
 	tr.Write(&ws.Message{Type: "mjpeg"})
 
-	wr := &writer2{tr: tr} // TODO: fixme
+	go cons.WriteTo(tr.Writer())
-	go cons.WriteTo(wr)
 
 	tr.OnClose(func() {
 		stream.RemoveConsumer(cons)
@@ -189,12 +188,3 @@ func handlerWS(tr *ws.Transport, _ *ws.Message) error {
 
 	return nil
 }
-
-type writer2 struct {
-	tr *ws.Transport
-}
-
-func (w *writer2) Write(p []byte) (n int, err error) {
-	w.tr.Write(p)
-	return len(p), nil
-}

internal/mp4/mp4.go

@@ -1,6 +1,7 @@
 package mp4
 
 import (
+	"io"
 	"net/http"
 	"strconv"
 	"strings"
@@ -47,18 +48,7 @@ func handlerKeyframe(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	exit := make(chan []byte, 1)
+	cons := mp4.NewKeyframe(nil)
-
-	cons := &mp4.Segment{OnlyKeyframe: true}
-	cons.Listen(func(msg any) {
-		if data, ok := msg.([]byte); ok && exit != nil {
-			select {
-			case exit <- data:
-			default:
-			}
-			exit = nil
-		}
-	})
 
 	if err := stream.AddConsumer(cons); err != nil {
 		log.Error().Err(err).Caller().Send()
@@ -66,24 +56,34 @@ func handlerKeyframe(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	data := <-exit
+	wr := &once{} // init and first frame
+	_, _ = cons.WriteTo(wr)
 
 	stream.RemoveConsumer(cons)
 
 	// Apple Safari won't show frame without length
 	header := w.Header()
-	header.Set("Content-Length", strconv.Itoa(len(data)))
+	header.Set("Content-Length", strconv.Itoa(len(wr.buf)))
-	header.Set("Content-Type", cons.MimeType)
+	header.Set("Content-Type", mp4.ContentType(cons.Codecs()))
 
 	if filename := query.Get("filename"); filename != "" {
 		header.Set("Content-Disposition", `attachment; filename="`+filename+`"`)
 	}
 
-	if _, err := w.Write(data); err != nil {
+	if _, err := w.Write(wr.buf); err != nil {
 		log.Error().Err(err).Caller().Send()
 	}
 }
 
+type once struct {
+	buf []byte
+}
+
+func (o *once) Write(p []byte) (n int, err error) {
+	o.buf = p
+	return 0, io.EOF
+}
+
 func handlerMP4(w http.ResponseWriter, r *http.Request) {
 	log.Trace().Msgf("[mp4] %s %+v", r.Method, r.Header)
 
@@ -108,29 +108,11 @@ func handlerMP4(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	exit := make(chan error, 1) // Add buffer to prevent blocking
+	medias := mp4.ParseQuery(r.URL.Query())
-
+	cons := mp4.NewConsumer(medias)
-	cons := &mp4.Consumer{
+	cons.Type = "MP4/HTTP active consumer"
-		Desc:       "MP4/HTTP",
+	cons.RemoteAddr = tcp.RemoteAddr(r)
-		RemoteAddr: tcp.RemoteAddr(r),
+	cons.UserAgent = r.UserAgent()
-		UserAgent:  r.UserAgent(),
-		Medias:     mp4.ParseQuery(r.URL.Query()),
-	}
-
-	cons.Listen(func(msg any) {
-		if exit == nil {
-			return
-		}
-		if data, ok := msg.([]byte); ok {
-			if _, err := w.Write(data); err != nil {
-				select {
-				case exit <- err:
-				default:
-				}
-				exit = nil
-			}
-		}
-	})
 
 	if err := stream.AddConsumer(cons); err != nil {
 		log.Error().Err(err).Caller().Send()
@@ -140,57 +122,34 @@ func handlerMP4(w http.ResponseWriter, r *http.Request) {
 
 	defer stream.RemoveConsumer(cons)
 
-	data, err := cons.Init()
-	if err != nil {
-		log.Error().Err(err).Caller().Send()
-		http.Error(w, err.Error(), http.StatusInternalServerError)
-		return
-	}
-
 	header := w.Header()
-	header.Set("Content-Type", cons.MimeType())
+	header.Set("Content-Type", mp4.ContentType(cons.Codecs()))
 
 	if filename := query.Get("filename"); filename != "" {
 		header.Set("Content-Disposition", `attachment; filename="`+filename+`"`)
 	}
 
 	if rotate := query.Get("rotate"); rotate != "" {
-		mp4.PatchVideoRotate(data, core.Atoi(rotate))
+		cons.Rotate = core.Atoi(rotate)
 	}
 
 	if scale := query.Get("scale"); scale != "" {
 		if sx, sy, ok := strings.Cut(scale, ":"); ok {
-			mp4.PatchVideoScale(data, core.Atoi(sx), core.Atoi(sy))
+			cons.ScaleX = core.Atoi(sx)
+			cons.ScaleY = core.Atoi(sy)
 		}
 	}
 
-	if _, err = w.Write(data); err != nil {
-		log.Error().Err(err).Caller().Send()
-		http.Error(w, err.Error(), http.StatusInternalServerError)
-		return
-	}
-
-	cons.Start()
-
 	var duration *time.Timer
 	if s := query.Get("duration"); s != "" {
 		if i, _ := strconv.Atoi(s); i > 0 {
 			duration = time.AfterFunc(time.Second*time.Duration(i), func() {
-				if exit != nil {
+				_ = cons.Stop()
-					select {
-					case exit <- nil:
-					default:
-					}
-					exit = nil
-				}
 			})
 		}
 	}
 
-	err = <-exit
+	_, _ = cons.WriteTo(w)
-	exit = nil
-
-	log.Trace().Err(err).Caller().Send()
 
 	if duration != nil {
 		duration.Stop()

internal/mp4/ws.go

@@ -6,6 +6,7 @@ import (
 	"github.com/AlexxIT/go2rtc/internal/api"
 	"github.com/AlexxIT/go2rtc/internal/api/ws"
 	"github.com/AlexxIT/go2rtc/internal/streams"
+	"github.com/AlexxIT/go2rtc/pkg/core"
 	"github.com/AlexxIT/go2rtc/pkg/mp4"
 	"github.com/AlexxIT/go2rtc/pkg/tcp"
 )
@@ -16,44 +17,30 @@ func handlerWSMSE(tr *ws.Transport, msg *ws.Message) error {
 		return errors.New(api.StreamNotFound)
 	}
 
-	cons := &mp4.Consumer{
+	var medias []*core.Media
-		Desc:       "MSE/WebSocket",
-		RemoteAddr: tcp.RemoteAddr(tr.Request),
-		UserAgent:  tr.Request.UserAgent(),
-	}
-
 	if codecs := msg.String(); codecs != "" {
 		log.Trace().Str("codecs", codecs).Msgf("[mp4] new WS/MSE consumer")
-		cons.Medias = mp4.ParseCodecs(codecs, true)
+		medias = mp4.ParseCodecs(codecs, true)
 	}
 
-	cons.Listen(func(msg any) {
+	cons := mp4.NewConsumer(medias)
-		if data, ok := msg.([]byte); ok {
+	cons.Type = "MSE/WebSocket active consumer"
-			tr.Write(data)
+	cons.RemoteAddr = tcp.RemoteAddr(tr.Request)
-		}
+	cons.UserAgent = tr.Request.UserAgent()
-	})
 
 	if err := stream.AddConsumer(cons); err != nil {
 		log.Debug().Err(err).Msg("[mp4] add consumer")
 		return err
 	}
 
+	tr.Write(&ws.Message{Type: "mse", Value: mp4.ContentType(cons.Codecs())})
+
+	go cons.WriteTo(tr.Writer())
+
 	tr.OnClose(func() {
 		stream.RemoveConsumer(cons)
 	})
 
-	tr.Write(&ws.Message{Type: "mse", Value: cons.MimeType()})
-
-	data, err := cons.Init()
-	if err != nil {
-		log.Warn().Err(err).Caller().Send()
-		return err
-	}
-
-	tr.Write(data)
-
-	cons.Start()
-
 	return nil
 }
 
@@ -63,29 +50,25 @@ func handlerWSMP4(tr *ws.Transport, msg *ws.Message) error {
 		return errors.New(api.StreamNotFound)
 	}
 
-	cons := &mp4.Segment{
+	var medias []*core.Media
-		RemoteAddr:   tcp.RemoteAddr(tr.Request),
-		UserAgent:    tr.Request.UserAgent(),
-		OnlyKeyframe: true,
-	}
-
 	if codecs := msg.String(); codecs != "" {
 		log.Trace().Str("codecs", codecs).Msgf("[mp4] new WS/MP4 consumer")
-		cons.Medias = mp4.ParseCodecs(codecs, false)
+		medias = mp4.ParseCodecs(codecs, false)
 	}
 
-	cons.Listen(func(msg any) {
+	cons := mp4.NewKeyframe(medias)
-		if data, ok := msg.([]byte); ok {
+	cons.Type = "MP4/WebSocket active consumer"
-			tr.Write(data)
+	cons.RemoteAddr = tcp.RemoteAddr(tr.Request)
-		}
+	cons.UserAgent = tr.Request.UserAgent()
-	})
 
 	if err := stream.AddConsumer(cons); err != nil {
 		log.Error().Err(err).Caller().Send()
 		return err
 	}
 
-	tr.Write(&ws.Message{Type: "mp4", Value: cons.MimeType})
+	tr.Write(&ws.Message{Type: "mse", Value: mp4.ContentType(cons.Codecs())})
+
+	go cons.WriteTo(tr.Writer())
 
 	tr.OnClose(func() {
 		stream.RemoveConsumer(cons)

internal/mpegts/mpegts.go

@@ -6,6 +6,8 @@ import (
 	"github.com/AlexxIT/go2rtc/internal/api"
 	"github.com/AlexxIT/go2rtc/internal/streams"
 	"github.com/AlexxIT/go2rtc/pkg/mpegts"
+	"github.com/AlexxIT/go2rtc/pkg/tcp"
+	"github.com/rs/zerolog/log"
 )
 
 func Init() {
@@ -14,10 +16,38 @@ func Init() {
 
 func apiHandle(w http.ResponseWriter, r *http.Request) {
 	if r.Method != "POST" {
-		http.Error(w, "", http.StatusMethodNotAllowed)
+		outputMpegTS(w, r)
+	} else {
+		inputMpegTS(w, r)
+	}
+}
+
+func outputMpegTS(w http.ResponseWriter, r *http.Request) {
+	src := r.URL.Query().Get("src")
+	stream := streams.Get(src)
+	if stream == nil {
+		http.Error(w, api.StreamNotFound, http.StatusNotFound)
 		return
 	}
 
+	cons := mpegts.NewConsumer()
+	cons.RemoteAddr = tcp.RemoteAddr(r)
+	cons.UserAgent = r.UserAgent()
+
+	if err := stream.AddConsumer(cons); err != nil {
+		log.Error().Err(err).Caller().Send()
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Add("Content-Type", "video/mp2t")
+
+	_, _ = cons.WriteTo(w)
+
+	stream.RemoveConsumer(cons)
+}
+
+func inputMpegTS(w http.ResponseWriter, r *http.Request) {
 	dst := r.URL.Query().Get("dst")
 	stream := streams.Get(dst)
 	if stream == nil {

pkg/core/core.go

@@ -138,11 +138,13 @@ func (s *SuperProducer) Close() error {
 }
 
 type SuperConsumer struct {
-	Type    string    `json:"type,omitempty"`
+	Type       string    `json:"type,omitempty"`
-	URL     string    `json:"url,omitempty"`
+	URL        string    `json:"url,omitempty"`
-	Medias  []*Media  `json:"medias,omitempty"`
+	RemoteAddr string    `json:"remote_addr,omitempty"`
-	Senders []*Sender `json:"receivers,omitempty"`
+	UserAgent  string    `json:"user_agent,omitempty"`
-	Send    int       `json:"recv,omitempty"`
+	Medias     []*Media  `json:"medias,omitempty"`
+	Senders    []*Sender `json:"receivers,omitempty"`
+	Send       int       `json:"recv,omitempty"`
 }
 
 func (s *SuperConsumer) GetMedias() []*Media {
@@ -163,3 +165,11 @@ func (s *SuperConsumer) Close() error {
 	}
 	return nil
 }
+
+func (s *SuperConsumer) Codecs() []*Codec {
+	codecs := make([]*Codec, len(s.Senders))
+	for i, sender := range s.Senders {
+		codecs[i] = sender.Codec
+	}
+	return codecs
+}

pkg/mp4/consumer.go

@@ -1,7 +1,8 @@
 package mp4
 
 import (
-	"encoding/json"
+	"errors"
+	"io"
 	"sync"
 
 	"github.com/AlexxIT/go2rtc/pkg/aac"
@@ -13,27 +14,21 @@ import (
 )
 
 type Consumer struct {
-	core.Listener
+	core.SuperConsumer
-
+	wr    *core.WriteBuffer
-	Medias []*core.Media
-
-	Desc       string
-	UserAgent  string
-	RemoteAddr string
-
-	senders []*core.Sender
-
 	muxer *Muxer
 	mu    sync.Mutex
-	state byte
+	start bool
 
-	send int
+	Rotate int `json:"-"`
+	ScaleX int `json:"-"`
+	ScaleY int `json:"-"`
 }
 
-func (c *Consumer) GetMedias() []*core.Media {
+func NewConsumer(medias []*core.Media) *Consumer {
-	if c.Medias == nil {
+	if medias == nil {
 		// default local medias
-		c.Medias = []*core.Media{
+		medias = []*core.Media{
 			{
 				Kind:      core.KindVideo,
 				Direction: core.DirectionSendonly,
@@ -52,11 +47,16 @@ func (c *Consumer) GetMedias() []*core.Media {
 		}
 	}
 
-	return c.Medias
+	cons := &Consumer{
+		muxer: &Muxer{},
+		wr:    core.NewWriteBuffer(nil),
+	}
+	cons.Medias = medias
+	return cons
 }
 
 func (c *Consumer) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiver) error {
-	trackID := byte(len(c.senders))
+	trackID := byte(len(c.Senders))
 
 	codec := track.Codec.Clone()
 	handler := core.NewSender(media, codec)
@@ -64,22 +64,19 @@ func (c *Consumer) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiv
 	switch track.Codec.Name {
 	case core.CodecH264:
 		handler.Handler = func(packet *rtp.Packet) {
-			if packet.Version != h264.RTPPacketVersionAVC {
+			if !c.start {
-				return
+				if !h264.IsKeyframe(packet.Payload) {
-			}
-
-			if c.state != stateStart {
-				if c.state != stateInit || !h264.IsKeyframe(packet.Payload) {
 					return
 				}
-				c.state = stateStart
+				c.start = true
 			}
 
 			// important to use Mutex because right fragment order
 			c.mu.Lock()
-			buf := c.muxer.Marshal(trackID, packet)
+			b := c.muxer.GetPayload(trackID, packet)
-			c.Fire(buf)
+			if n, err := c.wr.Write(b); err == nil {
-			c.send += len(buf)
+				c.Send += n
+			}
 			c.mu.Unlock()
 		}
 
@@ -91,21 +88,19 @@ func (c *Consumer) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiv
 
 	case core.CodecH265:
 		handler.Handler = func(packet *rtp.Packet) {
-			if packet.Version != h264.RTPPacketVersionAVC {
+			if !c.start {
-				return
+				if !h265.IsKeyframe(packet.Payload) {
-			}
-
-			if c.state != stateStart {
-				if c.state != stateInit || !h265.IsKeyframe(packet.Payload) {
 					return
 				}
-				c.state = stateStart
+				c.start = true
 			}
 
+			// important to use Mutex because right fragment order
 			c.mu.Lock()
-			buf := c.muxer.Marshal(trackID, packet)
+			b := c.muxer.GetPayload(trackID, packet)
-			c.Fire(buf)
+			if n, err := c.wr.Write(b); err == nil {
-			c.send += len(buf)
+				c.Send += n
+			}
 			c.mu.Unlock()
 		}
 
@@ -115,14 +110,16 @@ func (c *Consumer) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiv
 
 	default:
 		handler.Handler = func(packet *rtp.Packet) {
-			if c.state != stateStart {
+			if !c.start {
 				return
 			}
 
+			// important to use Mutex because right fragment order
 			c.mu.Lock()
-			buf := c.muxer.Marshal(trackID, packet)
+			b := c.muxer.GetPayload(trackID, packet)
-			c.Fire(buf)
+			if n, err := c.wr.Write(b); err == nil {
-			c.send += len(buf)
+				c.Send += n
+			}
 			c.mu.Unlock()
 		}
 
@@ -142,64 +139,40 @@ func (c *Consumer) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiv
 	}
 
 	if handler.Handler == nil {
-		println("ERROR: MP4 unsupported codec: " + track.Codec.String())
+		s := "mp4: unsupported codec: " + track.Codec.String()
-		return nil
+		println(s)
+		return errors.New(s)
 	}
 
+	c.muxer.AddTrack(codec)
+
 	handler.HandleRTP(track)
-	c.senders = append(c.senders, handler)
+	c.Senders = append(c.Senders, handler)
 
 	return nil
 }
 
+func (c *Consumer) WriteTo(wr io.Writer) (int64, error) {
+	init, err := c.muxer.GetInit()
+	if err != nil {
+		return 0, err
+	}
+
+	if c.Rotate != 0 {
+		PatchVideoRotate(init, c.Rotate)
+	}
+	if c.ScaleX != 0 && c.ScaleY != 0 {
+		PatchVideoScale(init, c.ScaleX, c.ScaleY)
+	}
+
+	if _, err = wr.Write(init); err != nil {
+		return 0, err
+	}
+
+	return c.wr.WriteTo(wr)
+}
+
 func (c *Consumer) Stop() error {
-	for _, sender := range c.senders {
+	_ = c.SuperConsumer.Close()
-		sender.Close()
+	return c.wr.Close()
-	}
-	return nil
-}
-
-func (c *Consumer) Codecs() []*core.Codec {
-	codecs := make([]*core.Codec, len(c.senders))
-	for i, sender := range c.senders {
-		codecs[i] = sender.Codec
-	}
-	return codecs
-}
-
-func (c *Consumer) MimeCodecs() string {
-	return c.muxer.MimeCodecs(c.Codecs())
-}
-
-func (c *Consumer) MimeType() string {
-	return `video/mp4; codecs="` + c.MimeCodecs() + `"`
-}
-
-func (c *Consumer) Init() ([]byte, error) {
-	c.muxer = &Muxer{}
-	return c.muxer.GetInit(c.Codecs())
-}
-
-func (c *Consumer) Start() {
-	for _, sender := range c.senders {
-		switch sender.Codec.Name {
-		case core.CodecH264, core.CodecH265:
-			c.state = stateInit
-			return
-		}
-	}
-
-	c.state = stateStart
-}
-
-func (c *Consumer) MarshalJSON() ([]byte, error) {
-	info := &core.Info{
-		Type:       c.Desc + " passive consumer",
-		RemoteAddr: c.RemoteAddr,
-		UserAgent:  c.UserAgent,
-		Medias:     c.Medias,
-		Senders:    c.senders,
-		Send:       c.send,
-	}
-	return json.Marshal(info)
 }

pkg/mp4/consumer_test.go

@@ -1,77 +0,0 @@
-package mp4
-
-import (
-	"github.com/AlexxIT/go2rtc/pkg/core"
-	"github.com/AlexxIT/go2rtc/pkg/h264"
-	"github.com/pion/rtp"
-	"github.com/stretchr/testify/require"
-	"testing"
-	"time"
-)
-
-func TestStartH264(t *testing.T) {
-	codec := &core.Codec{Name: core.CodecH264}
-	track := core.NewReceiver(nil, codec)
-
-	packetKey := &rtp.Packet{
-		Header:  rtp.Header{Marker: true},
-		Payload: []byte{h264.NALUTypeIFrame, 0, 0},
-	}
-
-	packetNotKey := &rtp.Packet{
-		Header:  rtp.Header{Marker: true},
-		Payload: []byte{h264.NALUTypePFrame, 0, 0},
-	}
-
-	cons := &Consumer{}
-	err := cons.AddTrack(nil, nil, track)
-	require.Nil(t, err)
-
-	track.WriteRTP(packetKey)
-	time.Sleep(time.Millisecond)
-
-	_, err = cons.Init()
-	require.Nil(t, err)
-
-	cons.Start()
-
-	track.WriteRTP(packetNotKey)
-	time.Sleep(time.Millisecond)
-
-	require.Zero(t, cons.send)
-
-	track.WriteRTP(packetKey)
-	time.Sleep(time.Millisecond)
-
-	require.NotZero(t, cons.send)
-}
-
-func TestStartOPUS(t *testing.T) {
-	// Test for fix this issue
-	// https://github.com/AlexxIT/go2rtc/issues/404
-	codec := &core.Codec{Name: core.CodecOpus}
-	track := core.NewReceiver(nil, codec)
-
-	cons := &Consumer{}
-	err := cons.AddTrack(nil, nil, track)
-	require.Nil(t, err)
-
-	track.WriteRTP(&rtp.Packet{
-		Payload: []byte{0},
-	})
-	time.Sleep(time.Millisecond)
-
-	require.Zero(t, cons.send)
-
-	_, err = cons.Init()
-	require.Nil(t, err)
-
-	cons.Start()
-
-	track.WriteRTP(&rtp.Packet{
-		Payload: []byte{0},
-	})
-	time.Sleep(time.Millisecond)
-
-	require.NotZero(t, cons.send)
-}

pkg/mp4/keyframe.go

@@ -0,0 +1,102 @@
+package mp4
+
+import (
+	"io"
+
+	"github.com/AlexxIT/go2rtc/pkg/core"
+	"github.com/AlexxIT/go2rtc/pkg/h264"
+	"github.com/AlexxIT/go2rtc/pkg/h265"
+	"github.com/pion/rtp"
+)
+
+type Keyframe struct {
+	core.SuperConsumer
+	wr    *core.WriteBuffer
+	muxer *Muxer
+}
+
+func NewKeyframe(medias []*core.Media) *Keyframe {
+	if medias == nil {
+		medias = []*core.Media{
+			{
+				Kind:      core.KindVideo,
+				Direction: core.DirectionSendonly,
+				Codecs: []*core.Codec{
+					{Name: core.CodecH264},
+					{Name: core.CodecH265},
+				},
+			},
+		}
+	}
+
+	cons := &Keyframe{
+		muxer: &Muxer{},
+		wr:    core.NewWriteBuffer(nil),
+	}
+	cons.Medias = medias
+	return cons
+}
+
+func (c *Keyframe) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiver) error {
+	c.muxer.AddTrack(track.Codec)
+	init, err := c.muxer.GetInit()
+	if err != nil {
+		return err
+	}
+
+	handler := core.NewSender(media, track.Codec)
+
+	switch track.Codec.Name {
+	case core.CodecH264:
+		handler.Handler = func(packet *rtp.Packet) {
+			if !h264.IsKeyframe(packet.Payload) {
+				return
+			}
+
+			// important to use Mutex because right fragment order
+			b := c.muxer.GetPayload(0, packet)
+			b = append(init, b...)
+			if n, err := c.wr.Write(b); err == nil {
+				c.Send += n
+			}
+		}
+
+		if track.Codec.IsRTP() {
+			handler.Handler = h264.RTPDepay(track.Codec, handler.Handler)
+		} else {
+			handler.Handler = h264.RepairAVCC(track.Codec, handler.Handler)
+		}
+
+	case core.CodecH265:
+		handler.Handler = func(packet *rtp.Packet) {
+			if !h265.IsKeyframe(packet.Payload) {
+				return
+			}
+
+			// important to use Mutex because right fragment order
+			b := c.muxer.GetPayload(0, packet)
+			b = append(init, b...)
+			if n, err := c.wr.Write(b); err == nil {
+				c.Send += n
+			}
+		}
+
+		if track.Codec.IsRTP() {
+			handler.Handler = h265.RTPDepay(track.Codec, handler.Handler)
+		}
+	}
+
+	handler.HandleRTP(track)
+	c.Senders = append(c.Senders, handler)
+
+	return nil
+}
+
+func (c *Keyframe) WriteTo(wr io.Writer) (int64, error) {
+	return c.wr.WriteTo(wr)
+}
+
+func (c *Keyframe) Stop() error {
+	_ = c.SuperConsumer.Close()
+	return c.wr.Close()
+}

pkg/mp4/mime.go

@@ -0,0 +1,45 @@
+package mp4
+
+import (
+	"github.com/AlexxIT/go2rtc/pkg/core"
+	"github.com/AlexxIT/go2rtc/pkg/h264"
+)
+
+const (
+	MimeH264 = "avc1.640029"
+	MimeH265 = "hvc1.1.6.L153.B0"
+	MimeAAC  = "mp4a.40.2"
+	MimeFlac = "flac"
+	MimeOpus = "opus"
+)
+
+func MimeCodecs(codecs []*core.Codec) string {
+	var s string
+
+	for i, codec := range codecs {
+		if i > 0 {
+			s += ","
+		}
+
+		switch codec.Name {
+		case core.CodecH264:
+			s += "avc1." + h264.GetProfileLevelID(codec.FmtpLine)
+		case core.CodecH265:
+			// H.265 profile=main level=5.1
+			// hvc1 - supported in Safari, hev1 - doesn't, both supported in Chrome
+			s += MimeH265
+		case core.CodecAAC:
+			s += MimeAAC
+		case core.CodecOpus:
+			s += MimeOpus
+		case core.CodecFLAC:
+			s += MimeFlac
+		}
+	}
+
+	return s
+}
+
+func ContentType(codecs []*core.Codec) string {
+	return `video/mp4; codecs="` + MimeCodecs(codecs) + `"`
+}

pkg/mp4/muxer.go

@@ -12,55 +12,26 @@ import (
 )
 
 type Muxer struct {
-	fragIndex uint32
+	index  uint32
-	dts       []uint64
+	dts    []uint64
-	pts       []uint32
+	pts    []uint32
-	codecs    []*core.Codec
+	codecs []*core.Codec
 }
 
-const (
+func (m *Muxer) AddTrack(codec *core.Codec) {
-	MimeH264 = "avc1.640029"
+	m.dts = append(m.dts, 0)
-	MimeH265 = "hvc1.1.6.L153.B0"
+	m.pts = append(m.pts, 0)
-	MimeAAC  = "mp4a.40.2"
+	m.codecs = append(m.codecs, codec)
-	MimeFlac = "flac"
-	MimeOpus = "opus"
-)
-
-func (m *Muxer) MimeCodecs(codecs []*core.Codec) string {
-	var s string
-
-	for i, codec := range codecs {
-		if i > 0 {
-			s += ","
-		}
-
-		switch codec.Name {
-		case core.CodecH264:
-			s += "avc1." + h264.GetProfileLevelID(codec.FmtpLine)
-		case core.CodecH265:
-			// H.265 profile=main level=5.1
-			// hvc1 - supported in Safari, hev1 - doesn't, both supported in Chrome
-			s += MimeH265
-		case core.CodecAAC:
-			s += MimeAAC
-		case core.CodecOpus:
-			s += MimeOpus
-		case core.CodecFLAC:
-			s += MimeFlac
-		}
-	}
-
-	return s
 }
 
-func (m *Muxer) GetInit(codecs []*core.Codec) ([]byte, error) {
+func (m *Muxer) GetInit() ([]byte, error) {
 	mv := iso.NewMovie(1024)
 	mv.WriteFileType()
 
 	mv.StartAtom(iso.Moov)
 	mv.WriteMovieHeader()
 
-	for i, codec := range codecs {
+	for i, codec := range m.codecs {
 		switch codec.Name {
 		case core.CodecH264:
 			sps, pps := h264.GetParameterSet(codec.FmtpLine)
@@ -119,14 +90,10 @@ func (m *Muxer) GetInit(codecs []*core.Codec) ([]byte, error) {
 				uint32(i+1), codec.Name, codec.ClockRate, codec.Channels, nil,
 			)
 		}
-
-		m.dts = append(m.dts, 0)
-		m.pts = append(m.pts, 0)
-		m.codecs = append(m.codecs, codec)
 	}
 
 	mv.StartAtom(iso.MoovMvex)
-	for i := range codecs {
+	for i := range m.codecs {
 		mv.WriteTrackExtend(uint32(i + 1))
 	}
 	mv.EndAtom() // MVEX
@@ -137,17 +104,17 @@ func (m *Muxer) GetInit(codecs []*core.Codec) ([]byte, error) {
 }
 
 func (m *Muxer) Reset() {
-	m.fragIndex = 0
+	m.index = 0
 	for i := range m.dts {
 		m.dts[i] = 0
 		m.pts[i] = 0
 	}
 }
 
-func (m *Muxer) Marshal(trackID byte, packet *rtp.Packet) []byte {
+func (m *Muxer) GetPayload(trackID byte, packet *rtp.Packet) []byte {
 	codec := m.codecs[trackID]
 
-	m.fragIndex++
+	m.index++
 
 	duration := packet.Timestamp - m.pts[trackID]
 	m.pts[trackID] = packet.Timestamp
@@ -185,11 +152,11 @@ func (m *Muxer) Marshal(trackID byte, packet *rtp.Packet) []byte {
 
 	mv := iso.NewMovie(1024 + size)
 	mv.WriteMovieFragment(
-		m.fragIndex, uint32(trackID+1), duration, uint32(size), flags, m.dts[trackID],
+		m.index, uint32(trackID+1), duration, uint32(size), flags, m.dts[trackID],
 	)
 	mv.WriteData(packet.Payload)
 
-	//log.Printf("[MP4] track=%d ts=%6d dur=%5d idx=%3d len=%d", trackID+1, m.dts[trackID], duration, m.fragIndex, len(packet.Payload))
+	//log.Printf("[MP4] track=%d ts=%6d dur=%5d idx=%3d len=%d", trackID+1, m.dts[trackID], duration, m.index, len(packet.Payload))
 
 	m.dts[trackID] += uint64(duration)
 

pkg/mp4/segment.go

@@ -1,151 +0,0 @@
-package mp4
-
-import (
-	"encoding/json"
-
-	"github.com/AlexxIT/go2rtc/pkg/core"
-	"github.com/AlexxIT/go2rtc/pkg/h264"
-	"github.com/AlexxIT/go2rtc/pkg/h265"
-	"github.com/pion/rtp"
-)
-
-type Segment struct {
-	core.Listener
-
-	Medias     []*core.Media
-	UserAgent  string
-	RemoteAddr string
-
-	senders []*core.Sender
-
-	MimeType     string
-	OnlyKeyframe bool
-
-	send int
-}
-
-func (c *Segment) GetMedias() []*core.Media {
-	if c.Medias != nil {
-		return c.Medias
-	}
-
-	// default local medias
-	return []*core.Media{
-		{
-			Kind:      core.KindVideo,
-			Direction: core.DirectionSendonly,
-			Codecs: []*core.Codec{
-				{Name: core.CodecH264},
-				{Name: core.CodecH265},
-			},
-		},
-	}
-}
-
-func (c *Segment) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiver) error {
-	muxer := &Muxer{}
-
-	codecs := []*core.Codec{track.Codec}
-
-	init, err := muxer.GetInit(codecs)
-	if err != nil {
-		return nil
-	}
-
-	c.MimeType = `video/mp4; codecs="` + muxer.MimeCodecs(codecs) + `"`
-
-	handler := core.NewSender(media, track.Codec)
-
-	switch track.Codec.Name {
-	case core.CodecH264:
-
-		if c.OnlyKeyframe {
-			handler.Handler = func(packet *rtp.Packet) {
-				if !h264.IsKeyframe(packet.Payload) {
-					return
-				}
-
-				buf := muxer.Marshal(0, packet)
-				c.Fire(append(init, buf...))
-
-				c.send += len(buf)
-			}
-		} else {
-			var buf []byte
-
-			handler.Handler = func(packet *rtp.Packet) {
-				if h264.IsKeyframe(packet.Payload) {
-					// fist frame - send only IFrame
-					// other frames - send IFrame and all PFrames
-					if buf == nil {
-						buf = append(buf, init...)
-						b := muxer.Marshal(0, packet)
-						buf = append(buf, b...)
-					}
-
-					c.Fire(buf)
-
-					c.send += len(buf)
-
-					buf = buf[:0]
-					buf = append(buf, init...)
-					muxer.Reset()
-				}
-
-				if buf != nil {
-					b := muxer.Marshal(0, packet)
-					buf = append(buf, b...)
-				}
-			}
-		}
-
-		if track.Codec.IsRTP() {
-			handler.Handler = h264.RTPDepay(track.Codec, handler.Handler)
-		} else {
-			handler.Handler = h264.RepairAVCC(track.Codec, handler.Handler)
-		}
-
-	case core.CodecH265:
-		handler.Handler = func(packet *rtp.Packet) {
-			if !h265.IsKeyframe(packet.Payload) {
-				return
-			}
-
-			buf := muxer.Marshal(0, packet)
-			c.Fire(append(init, buf...))
-
-			c.send += len(buf)
-		}
-
-		if track.Codec.IsRTP() {
-			handler.Handler = h265.RTPDepay(track.Codec, handler.Handler)
-		}
-
-	default:
-		panic(core.UnsupportedCodec)
-	}
-
-	handler.HandleRTP(track)
-	c.senders = append(c.senders, handler)
-
-	return nil
-}
-
-func (c *Segment) Stop() error {
-	for _, sender := range c.senders {
-		sender.Close()
-	}
-	return nil
-}
-
-func (c *Segment) MarshalJSON() ([]byte, error) {
-	info := &core.Info{
-		Type:       "MP4/WebSocket passive consumer",
-		RemoteAddr: c.RemoteAddr,
-		UserAgent:  c.UserAgent,
-		Medias:     c.Medias,
-		Senders:    c.senders,
-		Send:       c.send,
-	}
-	return json.Marshal(info)
-}

pkg/mpegts/consumer.go

@@ -107,7 +107,7 @@ func (c *Consumer) WriteTo(wr io.Writer) (int64, error) {
 	return c.wr.WriteTo(wr)
 }
 
-func (c *Consumer) Close() error {
+func (c *Consumer) Stop() error {
 	_ = c.SuperConsumer.Close()
 	return c.wr.Close()
 }

pkg/mpegts/muxer.go

@@ -45,17 +45,26 @@ func (m *Muxer) GetHeader() []byte {
 func (m *Muxer) GetPayload(pid uint16, pts uint32, payload []byte) []byte {
 	pes := m.pes[pid]
 
+	size := 8 + len(payload)
+
 	b := make([]byte, 14+len(payload))
 	_ = b[14] // bounds
+
 	b[0] = 0
 	b[1] = 0
 	b[2] = 1
 	b[3] = pes.StreamID
-	binary.BigEndian.PutUint16(b[4:], 8+uint16(len(payload)))
 	b[6] = 0x80 // Marker bits (binary)
 	b[7] = 0x80 // PTS indicator
 	b[8] = 5    // PES header length
+
+	// zero size is OK for video stream
+	if size <= 0xFFFF {
+		binary.BigEndian.PutUint16(b[4:], uint16(size))
+	}
+
 	WriteTime(b[9:], pts)
+
 	copy(b[14:], payload)
 
 	switch pes.StreamType {