diff --git a/internal/api/ws/ws.go b/internal/api/ws/ws.go index e70e745a..5b8a35c3 100644 --- a/internal/api/ws/ws.go +++ b/internal/api/ws/ws.go @@ -1,15 +1,17 @@ package ws import ( - "github.com/AlexxIT/go2rtc/internal/api" - "github.com/AlexxIT/go2rtc/internal/app" - "github.com/gorilla/websocket" - "github.com/rs/zerolog/log" + "io" "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 +} diff --git a/internal/hls/hls.go b/internal/hls/hls.go index 66e467c9..5d3cd918 100644 --- a/internal/hls/hls.go +++ b/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{ - Desc: "HLS/HTTP", - RemoteAddr: tcp.RemoteAddr(r), - UserAgent: r.UserAgent(), - Medias: medias, - } + c := mp4.NewConsumer(medias) + c.Type = "HLS/fMP4 consumer" + c.RemoteAddr = tcp.RemoteAddr(r) + c.UserAgent = r.UserAgent() + cons = c } else { - //cons = &mpegts.Consumer{ - // RemoteAddr: tcp.RemoteAddr(r), - // UserAgent: r.UserAgent(), - //} + c := mpegts.NewConsumer() + c.Type = "HLS/TS consumer" + 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} - - 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 := NewSession(cons) 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 - data := []byte(`#EXTM3U -#EXT-X-STREAM-INF:BANDWIDTH=192000,CODECS="` + codecs + `" -hls/playlist.m3u8?id=` + sid) - - if _, err := w.Write(data); err != nil { + if _, err := w.Write(session.Main()); 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 - session.init = nil - - session.segment0 = session.Segment() - if session.segment0 == nil { + data := session.Init() + if data == 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 - - if query.Get("n") != "0" { - data = session.Segment() - } else { - data = session.segment0 - } - + data := session.Segment() if data == nil { log.Warn().Msgf("[hls] can't get segment %s", r.URL.RawQuery) http.NotFound(w, r) diff --git a/internal/hls/session.go b/internal/hls/session.go index e28ab184..9f07ad41 100644 --- a/internal/hls/session.go +++ b/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 { - return fmt.Sprintf(s.template, s.seq, s.seq, s.seq+1) +func NewSession(cons core.Consumer) *Session { + 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 - s.buffer = s.init + if _, ok := s.cons.(*mp4.Consumer); ok { + s.buffer = nil + } else { + // for TS important to start new segment with init + s.buffer = s.init + } s.seq++ } s.mu.Unlock() diff --git a/internal/hls/ws.go b/internal/hls/ws.go index c8aebc1b..ea1f5a3a 100644 --- a/internal/hls/ws.go +++ b/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} - - cons.Listen(func(msg any) { - if data, ok := msg.([]byte); ok { - session.mu.Lock() - session.buffer = append(session.buffer, data...) - session.mu.Unlock() - } - }) + session := NewSession(cons) 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 - data := `#EXTM3U -#EXT-X-STREAM-INF:BANDWIDTH=192000,CODECS="` + codecs + `" -hls/playlist.m3u8?id=` + sid - - tr.Write(&ws.Message{Type: "hls", Value: data}) + main := session.Main() + tr.Write(&ws.Message{Type: "hls", Value: string(main)}) return nil } diff --git a/internal/mjpeg/init.go b/internal/mjpeg/init.go index e9d6dbee..19f8f6c9 100644 --- a/internal/mjpeg/init.go +++ b/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(wr) + go cons.WriteTo(tr.Writer()) 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 -} diff --git a/internal/mp4/mp4.go b/internal/mp4/mp4.go index 52f51f5b..15c51f71 100644 --- a/internal/mp4/mp4.go +++ b/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.Segment{OnlyKeyframe: true} - cons.Listen(func(msg any) { - if data, ok := msg.([]byte); ok && exit != nil { - select { - case exit <- data: - default: - } - exit = nil - } - }) + cons := mp4.NewKeyframe(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-Type", cons.MimeType) + header.Set("Content-Length", strconv.Itoa(len(wr.buf))) + 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 - - cons := &mp4.Consumer{ - Desc: "MP4/HTTP", - RemoteAddr: tcp.RemoteAddr(r), - 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 - } - } - }) + medias := mp4.ParseQuery(r.URL.Query()) + cons := mp4.NewConsumer(medias) + cons.Type = "MP4/HTTP active consumer" + cons.RemoteAddr = tcp.RemoteAddr(r) + cons.UserAgent = r.UserAgent() 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 { - select { - case exit <- nil: - default: - } - exit = nil - } + _ = cons.Stop() }) } } - err = <-exit - exit = nil - - log.Trace().Err(err).Caller().Send() + _, _ = cons.WriteTo(w) if duration != nil { duration.Stop() diff --git a/internal/mp4/ws.go b/internal/mp4/ws.go index b9071f31..060ff5f6 100644 --- a/internal/mp4/ws.go +++ b/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{ - Desc: "MSE/WebSocket", - RemoteAddr: tcp.RemoteAddr(tr.Request), - UserAgent: tr.Request.UserAgent(), - } - + var medias []*core.Media 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) { - if data, ok := msg.([]byte); ok { - tr.Write(data) - } - }) + cons := mp4.NewConsumer(medias) + cons.Type = "MSE/WebSocket active consumer" + 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{ - RemoteAddr: tcp.RemoteAddr(tr.Request), - UserAgent: tr.Request.UserAgent(), - OnlyKeyframe: true, - } - + var medias []*core.Media 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) { - if data, ok := msg.([]byte); ok { - tr.Write(data) - } - }) + cons := mp4.NewKeyframe(medias) + cons.Type = "MP4/WebSocket active consumer" + 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) diff --git a/internal/mpegts/mpegts.go b/internal/mpegts/mpegts.go index c359d204..b435e021 100644 --- a/internal/mpegts/mpegts.go +++ b/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 { diff --git a/pkg/core/core.go b/pkg/core/core.go index 10a435d3..6d7aadef 100644 --- a/pkg/core/core.go +++ b/pkg/core/core.go @@ -138,11 +138,13 @@ func (s *SuperProducer) Close() error { } type SuperConsumer struct { - Type string `json:"type,omitempty"` - URL string `json:"url,omitempty"` - Medias []*Media `json:"medias,omitempty"` - Senders []*Sender `json:"receivers,omitempty"` - Send int `json:"recv,omitempty"` + Type string `json:"type,omitempty"` + URL string `json:"url,omitempty"` + RemoteAddr string `json:"remote_addr,omitempty"` + UserAgent string `json:"user_agent,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 +} diff --git a/pkg/mp4/consumer.go b/pkg/mp4/consumer.go index 9b08fb0c..b1392e29 100644 --- a/pkg/mp4/consumer.go +++ b/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 - - Medias []*core.Media - - Desc string - UserAgent string - RemoteAddr string - - senders []*core.Sender - + core.SuperConsumer + wr *core.WriteBuffer 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 { - if c.Medias == nil { +func NewConsumer(medias []*core.Media) *Consumer { + 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 { - return - } - - if c.state != stateStart { - if c.state != stateInit || !h264.IsKeyframe(packet.Payload) { + if !c.start { + if !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) - c.Fire(buf) - c.send += len(buf) + b := c.muxer.GetPayload(trackID, packet) + if n, err := c.wr.Write(b); err == nil { + 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 { - return - } - - if c.state != stateStart { - if c.state != stateInit || !h265.IsKeyframe(packet.Payload) { + if !c.start { + if !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) - c.Fire(buf) - c.send += len(buf) + b := c.muxer.GetPayload(trackID, packet) + if n, err := c.wr.Write(b); err == nil { + 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) - c.Fire(buf) - c.send += len(buf) + b := c.muxer.GetPayload(trackID, packet) + if n, err := c.wr.Write(b); err == nil { + 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()) - return nil + s := "mp4: unsupported codec: " + track.Codec.String() + 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 { - sender.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) + _ = c.SuperConsumer.Close() + return c.wr.Close() } diff --git a/pkg/mp4/consumer_test.go b/pkg/mp4/consumer_test.go deleted file mode 100644 index 4991c044..00000000 --- a/pkg/mp4/consumer_test.go +++ /dev/null @@ -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) -} diff --git a/pkg/mp4/keyframe.go b/pkg/mp4/keyframe.go new file mode 100644 index 00000000..25a6983d --- /dev/null +++ b/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() +} diff --git a/pkg/mp4/mime.go b/pkg/mp4/mime.go new file mode 100644 index 00000000..f1a74e4e --- /dev/null +++ b/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) + `"` +} diff --git a/pkg/mp4/muxer.go b/pkg/mp4/muxer.go index f70475fc..f48887a8 100644 --- a/pkg/mp4/muxer.go +++ b/pkg/mp4/muxer.go @@ -12,55 +12,26 @@ import ( ) type Muxer struct { - fragIndex uint32 - dts []uint64 - pts []uint32 - codecs []*core.Codec + index uint32 + dts []uint64 + pts []uint32 + codecs []*core.Codec } -const ( - MimeH264 = "avc1.640029" - MimeH265 = "hvc1.1.6.L153.B0" - MimeAAC = "mp4a.40.2" - 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) AddTrack(codec *core.Codec) { + m.dts = append(m.dts, 0) + m.pts = append(m.pts, 0) + m.codecs = append(m.codecs, codec) } -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) diff --git a/pkg/mp4/segment.go b/pkg/mp4/segment.go deleted file mode 100644 index ad1559cd..00000000 --- a/pkg/mp4/segment.go +++ /dev/null @@ -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) -} diff --git a/pkg/mpegts/consumer.go b/pkg/mpegts/consumer.go index 13a98d09..9dd3c75a 100644 --- a/pkg/mpegts/consumer.go +++ b/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() } diff --git a/pkg/mpegts/muxer.go b/pkg/mpegts/muxer.go index 39c96ac6..7ea22847 100644 --- a/pkg/mpegts/muxer.go +++ b/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 {