Update version to 1.2.0

cmd/app/app.go

@@ -14,7 +14,7 @@ import (
 	"time"
 )
 
-var Version = "1.1.2"
+var Version = "1.2.0"
 var UserAgent = "go2rtc/" + Version
 
 var ConfigPath string

cmd/dvrip/dvrip.go

@@ -0,0 +1,25 @@
+package dvrip
+
+import (
+	"github.com/AlexxIT/go2rtc/cmd/streams"
+	"github.com/AlexxIT/go2rtc/pkg/dvrip"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+)
+
+func Init() {
+	streams.HandleFunc("dvrip", handle)
+}
+
+func handle(url string) (streamer.Producer, error) {
+	conn := dvrip.NewClient(url)
+	if err := conn.Dial(); err != nil {
+		return nil, err
+	}
+	if err := conn.Play(); err != nil {
+		return nil, err
+	}
+	if err := conn.Handle(); err != nil {
+		return nil, err
+	}
+	return conn, nil
+}

cmd/ffmpeg/ffmpeg.go

@@ -44,10 +44,12 @@ var defaults = map[string]string{
 	"global": "-hide_banner",
 
 	// inputs
-	"file": "-re -stream_loop -1 -i {input}",
+	"file": "-re -i {input}",
 	"http": "-fflags nobuffer -flags low_delay -i {input}",
 	"rtsp": "-fflags nobuffer -flags low_delay -timeout 5000000 -user_agent go2rtc/ffmpeg -rtsp_transport tcp -i {input}",
 
+	"rtsp/udp": "-fflags nobuffer -flags low_delay -timeout 5000000 -user_agent go2rtc/ffmpeg -i {input}",
+
 	// output
 	"output": "-user_agent ffmpeg/go2rtc -rtsp_transport tcp -f rtsp {output}",
 
@@ -95,6 +97,21 @@ var defaults = map[string]string{
 	"h265/videotoolbox": "-c:v hevc_videotoolbox -g 50 -bf 0 -profile:v high -level:v 5.1",
 }
 
+// inputTemplate - select input template from YAML config by template name
+// if query has input param - select another tempalte by this name
+// if there is no another template - use input param as template
+func inputTemplate(name, s string, query url.Values) string {
+	var template string
+	if input := query.Get("input"); input != "" {
+		if template = defaults[input]; template == "" {
+			template = input
+		}
+	} else {
+		template = defaults[name]
+	}
+	return strings.Replace(template, "{input}", s, 1)
+}
+
 func parseArgs(s string) *Args {
 	// init FFmpeg arguments
 	args := &Args{
@@ -118,7 +135,7 @@ func parseArgs(s string) *Args {
 	if i := strings.Index(s, "://"); i > 0 {
 		switch s[:i] {
 		case "http", "https", "rtmp":
-			args.input = strings.Replace(defaults["http"], "{input}", s, 1)
+			args.input = inputTemplate("http", s, query)
 		case "rtsp", "rtsps":
 			// https://ffmpeg.org/ffmpeg-protocols.html#rtsp
 			// skip unnecessary input tracks
@@ -131,7 +148,7 @@ func parseArgs(s string) *Args {
 				args.input = "-allowed_media_types audio "
 			}
 
-			args.input += strings.Replace(defaults["rtsp"], "{input}", s, 1)
+			args.input += inputTemplate("rtsp", s, query)
 		default:
 			args.input = "-i " + s
 		}
@@ -145,7 +162,7 @@ func parseArgs(s string) *Args {
 		default:
 			s += "?video&audio"
 		}
-		args.input = strings.Replace(defaults["rtsp"], "{input}", s, 1)
+		args.input = inputTemplate("rtsp", s, query)
 	} else if strings.HasPrefix(s, "device?") {
 		var err error
 		args.input, err = device.GetInput(s)
@@ -153,7 +170,7 @@ func parseArgs(s string) *Args {
 			return nil
 		}
 	} else {
-		args.input = strings.Replace(defaults["file"], "{input}", s, 1)
+		args.input = inputTemplate("file", s, query)
 	}
 
 	if query["async"] != nil {
@@ -208,7 +225,11 @@ func parseArgs(s string) *Args {
 		if args.video > 0 {
 			for _, video := range query["video"] {
 				if video != "copy" {
-					args.AddCodec(defaults[video])
+					if codec := defaults[video]; codec != "" {
+						args.AddCodec(codec)
+					} else {
+						args.AddCodec(video)
+					}
 				} else {
 					args.AddCodec("-c:v copy")
 				}
@@ -221,7 +242,11 @@ func parseArgs(s string) *Args {
 		if args.audio > 0 {
 			for _, audio := range query["audio"] {
 				if audio != "copy" {
-					args.AddCodec(defaults[audio])
+					if codec := defaults[audio]; codec != "" {
+						args.AddCodec(codec)
+					} else {
+						args.AddCodec(audio)
+					}
 				} else {
 					args.AddCodec("-c:a copy")
 				}

cmd/hls/hls.go

@@ -5,8 +5,8 @@ import (
 	"github.com/AlexxIT/go2rtc/cmd/api"
 	"github.com/AlexxIT/go2rtc/cmd/streams"
 	"github.com/AlexxIT/go2rtc/pkg/mp4"
+	"github.com/AlexxIT/go2rtc/pkg/mpegts"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
-	"github.com/AlexxIT/go2rtc/pkg/ts"
 	"github.com/rs/zerolog/log"
 	"net/http"
 	"strconv"
@@ -75,7 +75,7 @@ func handlerStream(w http.ResponseWriter, r *http.Request) {
 			Medias:     medias,
 		}
 	} else {
-		cons = &ts.Consumer{
+		cons = &mpegts.Consumer{
 			RemoteAddr: r.RemoteAddr,
 			UserAgent:  r.UserAgent(),
 		}

cmd/http/http.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/AlexxIT/go2rtc/cmd/streams"
 	"github.com/AlexxIT/go2rtc/pkg/mjpeg"
+	"github.com/AlexxIT/go2rtc/pkg/mpegts"
 	"github.com/AlexxIT/go2rtc/pkg/rtmp"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/AlexxIT/go2rtc/pkg/tcp"
@@ -41,6 +42,7 @@ func handle(url string) (streamer.Producer, error) {
 	switch ct {
 	case "image/jpeg", "multipart/x-mixed-replace":
 		return mjpeg.NewClient(res), nil
+
 	case "video/x-flv":
 		var conn *rtmp.Client
 		if conn, err = rtmp.Accept(res); err != nil {
@@ -50,6 +52,13 @@ func handle(url string) (streamer.Producer, error) {
 			return nil, err
 		}
 		return conn, nil
+
+	case "video/mpeg":
+		client := mpegts.NewClient(res)
+		if err = client.Handle(); err != nil {
+			return nil, err
+		}
+		return client, nil
 	}
 
 	return nil, fmt.Errorf("unsupported Content-Type: %s", ct)

cmd/mjpeg/mjpeg.go

@@ -6,6 +6,7 @@ import (
 	"github.com/AlexxIT/go2rtc/cmd/streams"
 	"github.com/AlexxIT/go2rtc/pkg/mjpeg"
 	"github.com/rs/zerolog/log"
+	"io"
 	"net/http"
 	"strconv"
 )
@@ -62,6 +63,14 @@ func handlerKeyframe(w http.ResponseWriter, r *http.Request) {
 const header = "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: "
 
 func handlerStream(w http.ResponseWriter, r *http.Request) {
+	if r.Method != "POST" {
+		outputMjpeg(w, r)
+	} else {
+		inputMjpeg(w, r)
+	}
+}
+
+func outputMjpeg(w http.ResponseWriter, r *http.Request) {
 	src := r.URL.Query().Get("src")
 	stream := streams.GetOrNew(src)
 	if stream == nil {
@@ -108,6 +117,27 @@ func handlerStream(w http.ResponseWriter, r *http.Request) {
 	//log.Trace().Msg("[api.mjpeg] close")
 }
 
+func inputMjpeg(w http.ResponseWriter, r *http.Request) {
+	dst := r.URL.Query().Get("dst")
+	stream := streams.Get(dst)
+	if stream == nil {
+		http.Error(w, api.StreamNotFound, http.StatusNotFound)
+		return
+	}
+
+	res := &http.Response{Body: r.Body, Header: r.Header, Request: r}
+	res.Header.Set("Content-Type", "multipart/mixed;boundary=")
+
+	client := mjpeg.NewClient(res)
+	stream.AddProducer(client)
+
+	if err := client.Start(); err != nil && err != io.EOF {
+		log.Warn().Err(err).Caller().Send()
+	}
+
+	stream.RemoveProducer(client)
+}
+
 func handlerWS(tr *api.Transport, _ *api.Message) error {
 	src := tr.Request.URL.Query().Get("src")
 	stream := streams.GetOrNew(src)

cmd/mpegts/mpegts.go

@@ -0,0 +1,43 @@
+package mpegts
+
+import (
+	"github.com/AlexxIT/go2rtc/cmd/api"
+	"github.com/AlexxIT/go2rtc/cmd/streams"
+	"github.com/AlexxIT/go2rtc/pkg/mpegts"
+	"net/http"
+)
+
+func Init() {
+	api.HandleFunc("api/stream.ts", apiHandle)
+}
+
+func apiHandle(w http.ResponseWriter, r *http.Request) {
+	if r.Method != "POST" {
+		http.Error(w, "", http.StatusMethodNotAllowed)
+		return
+	}
+
+	dst := r.URL.Query().Get("dst")
+	stream := streams.Get(dst)
+	if stream == nil {
+		http.Error(w, api.StreamNotFound, http.StatusNotFound)
+		return
+	}
+
+	res := &http.Response{Body: r.Body, Request: r}
+	client := mpegts.NewClient(res)
+
+	if err := client.Handle(); err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	stream.AddProducer(client)
+
+	if err := client.Handle(); err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	stream.RemoveProducer(client)
+}

cmd/rtmp/rtmp.go

@@ -1,16 +1,22 @@
 package rtmp
 
 import (
+	"github.com/AlexxIT/go2rtc/cmd/api"
 	"github.com/AlexxIT/go2rtc/cmd/streams"
 	"github.com/AlexxIT/go2rtc/pkg/rtmp"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/rs/zerolog/log"
+	"io"
+	"net/http"
 )
 
 func Init() {
-	streams.HandleFunc("rtmp", handle)
+	streams.HandleFunc("rtmp", streamsHandle)
+
+	api.HandleFunc("api/stream.flv", apiHandle)
 }
 
-func handle(url string) (streamer.Producer, error) {
+func streamsHandle(url string) (streamer.Producer, error) {
 	conn := rtmp.NewClient(url)
 	if err := conn.Dial(); err != nil {
 		return nil, err
@@ -20,3 +26,37 @@ func handle(url string) (streamer.Producer, error) {
 	}
 	return conn, nil
 }
+
+func apiHandle(w http.ResponseWriter, r *http.Request) {
+	if r.Method != "POST" {
+		http.Error(w, "", http.StatusMethodNotAllowed)
+		return
+	}
+
+	dst := r.URL.Query().Get("dst")
+	stream := streams.Get(dst)
+	if stream == nil {
+		http.Error(w, api.StreamNotFound, http.StatusNotFound)
+		return
+	}
+
+	res := &http.Response{Body: r.Body, Request: r}
+	client, err := rtmp.Accept(res)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	if err = client.Describe(); err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	stream.AddProducer(client)
+
+	if err = client.Start(); err != nil && err != io.EOF {
+		log.Warn().Err(err).Caller().Send()
+	}
+
+	stream.RemoveProducer(client)
+}

cmd/streams/play.go

@@ -0,0 +1,116 @@
+package streams
+
+import (
+	"errors"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+)
+
+func (s *Stream) Play(source string) error {
+	s.mu.Lock()
+	for _, producer := range s.producers {
+		if producer.state == stateInternal && producer.element != nil {
+			_ = producer.element.Stop()
+		}
+	}
+	s.mu.Unlock()
+
+	if source == "" {
+		return nil
+	}
+
+	for _, producer := range s.producers {
+		// start new client
+		client, err := GetProducer(producer.url)
+		if err != nil {
+			return err
+		}
+
+		// check if client support consumer interface
+		cons, ok := client.(streamer.Consumer)
+		if !ok {
+			continue
+		}
+
+		// start new producer
+		prod, err := GetProducer(source)
+		if err != nil {
+			return err
+		}
+
+		if !matchMedia(prod, cons) {
+			return errors.New("can't match media")
+		}
+
+		s.AddInternalProducer(prod)
+		s.AddInternalConsumer(cons)
+
+		go func() {
+			_ = prod.Start()
+			_ = client.Stop()
+			s.RemoveProducer(prod)
+		}()
+
+		go func() {
+			_ = client.Start()
+			_ = prod.Stop()
+			s.RemoveInternalConsumer(cons)
+		}()
+
+		return nil
+	}
+
+	return errors.New("can't find consumer")
+}
+
+func (s *Stream) AddInternalProducer(prod streamer.Producer) {
+	producer := &Producer{element: prod, state: stateInternal}
+	s.mu.Lock()
+	s.producers = append(s.producers, producer)
+	s.mu.Unlock()
+}
+
+func (s *Stream) AddInternalConsumer(cons streamer.Consumer) {
+	consumer := &Consumer{element: cons}
+	s.mu.Lock()
+	s.consumers = append(s.consumers, consumer)
+	s.mu.Unlock()
+}
+
+func (s *Stream) RemoveInternalConsumer(cons streamer.Consumer) {
+	s.mu.Lock()
+	for i, consumer := range s.consumers {
+		if consumer.element == cons {
+			s.removeConsumer(i)
+			break
+		}
+	}
+	s.mu.Unlock()
+}
+
+func matchMedia(prod streamer.Producer, cons streamer.Consumer) bool {
+	for _, consMedia := range cons.GetMedias() {
+		for _, prodMedia := range prod.GetMedias() {
+			// codec negotiation
+			prodCodec := prodMedia.MatchMedia(consMedia)
+			if prodCodec == nil {
+				continue
+			}
+
+			// setup producer track
+			prodTrack := prod.GetTrack(prodMedia, prodCodec)
+			if prodTrack == nil {
+				return false
+			}
+
+			// setup consumer track
+			consTrack := cons.AddTrack(consMedia, prodTrack)
+			if consTrack == nil {
+				return false
+			}
+
+			return true
+		}
+	}
+
+	return false
+}

cmd/streams/producer.go

@@ -16,6 +16,7 @@ const (
 	stateTracks
 	stateStart
 	stateExternal
+	stateInternal
 )
 
 type Producer struct {

cmd/streams/stream_test.go

@@ -1,135 +0,0 @@
-package streams
-
-import (
-	"github.com/AlexxIT/go2rtc/pkg/fake"
-	"github.com/AlexxIT/go2rtc/pkg/rtsp"
-	"github.com/AlexxIT/go2rtc/pkg/streamer"
-	"github.com/stretchr/testify/assert"
-	"testing"
-	"time"
-)
-
-// Google Chrome 104.0.5112.79
-const chrome = `v=0
-o=- 0 0 IN IP4 0.0.0.0
-s=-
-t=0 0
-m=audio 9 UDP/TLS/RTP/SAVPF 111 63 103 104 9 0 8 110 112 113 126
-a=sendrecv
-a=rtpmap:111 opus/48000/2
-a=rtpmap:63 red/48000/2
-a=rtpmap:103 ISAC/16000
-a=rtpmap:104 ISAC/32000
-a=rtpmap:9 G722/8000
-a=rtpmap:0 PCMU/8000
-a=rtpmap:8 PCMA/8000
-a=rtpmap:110 telephone-event/48000
-a=rtpmap:112 telephone-event/32000
-a=rtpmap:113 telephone-event/16000
-a=rtpmap:126 telephone-event/8000
-m=video 9 UDP/TLS/RTP/SAVPF 96 97 98 99 100 101 102 122 127 121 125 107 108 109 124 120 123 119 35 36 37 38 39 40 41 42 114 115 116 117 118 43
-a=recvonly
-a=rtpmap:96 VP8/90000
-a=rtpmap:97 rtx/90000
-a=rtpmap:98 VP9/90000
-a=rtpmap:99 rtx/90000
-a=rtpmap:100 VP9/90000
-a=rtpmap:101 rtx/90000
-a=rtpmap:102 VP9/90000
-a=rtpmap:122 rtx/90000
-a=rtpmap:127 H264/90000
-a=fmtp:127 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42001f
-a=rtpmap:121 rtx/90000
-a=rtpmap:125 H264/90000
-a=fmtp:125 level-asymmetry-allowed=1;packetization-mode=0;profile-level-id=42001f
-a=rtpmap:107 rtx/90000
-a=rtpmap:108 H264/90000
-a=fmtp:108 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f
-a=rtpmap:109 rtx/90000
-a=rtpmap:124 H264/90000
-a=fmtp:124 level-asymmetry-allowed=1;packetization-mode=0;profile-level-id=42e01f
-a=rtpmap:120 rtx/90000
-a=rtpmap:123 H264/90000
-a=fmtp:123 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=4d001f
-a=rtpmap:119 rtx/90000
-a=rtpmap:35 H264/90000
-a=fmtp:35 level-asymmetry-allowed=1;packetization-mode=0;profile-level-id=4d001f
-a=rtpmap:36 rtx/90000
-a=rtpmap:37 H264/90000
-a=fmtp:37 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=f4001f
-a=rtpmap:38 rtx/90000
-a=rtpmap:39 H264/90000
-a=fmtp:39 level-asymmetry-allowed=1;packetization-mode=0;profile-level-id=f4001f
-a=rtpmap:40 rtx/90000
-a=rtpmap:41 AV1/90000
-a=rtpmap:42 rtx/90000
-a=rtpmap:114 H264/90000
-a=fmtp:114 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=64001f
-a=rtpmap:115 rtx/90000
-a=rtpmap:116 red/90000
-a=rtpmap:117 rtx/90000
-a=rtpmap:118 ulpfec/90000
-a=rtpmap:43 flexfec-03/90000
-`
-
-const dahuaSimple = `v=0
-o=- 0 0 IN IP4 0.0.0.0
-s=-
-t=0 0
-m=video 0 RTP/AVP 96
-a=control:trackID=0
-a=rtpmap:96 H264/90000
-a=fmtp:96 packetization-mode=1;profile-level-id=42401E;sprop-parameter-sets=Z0JAHqaAoD2QAA==,aM48gAA=
-a=recvonly
-m=audio 0 RTP/AVP 97
-a=control:trackID=1
-a=rtpmap:97 MPEG4-GENERIC/16000
-a=fmtp:97 streamtype=5;profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3;config=1408
-a=recvonly
-m=audio 0 RTP/AVP 8
-a=control:trackID=5
-a=rtpmap:8 PCMA/8000
-a=sendonly
-`
-
-const ffmpegPCMU48000 = `v=0
-o=- 0 0 IN IP4 127.0.0.1
-s=-
-t=0 0
-m=audio 0 RTP/AVP 96
-b=AS:384
-a=rtpmap:96 PCMU/48000/1
-a=control:streamid=0
-`
-
-func TestRouting(t *testing.T) {
-	prod := &fake.Producer{}
-	prod.Medias, _ = rtsp.UnmarshalSDP([]byte(dahuaSimple))
-	assert.Len(t, prod.Medias, 3)
-
-	HandleFunc("fake", func(url string) (streamer.Producer, error) {
-		return prod, nil
-	})
-
-	cons := &fake.Consumer{}
-	cons.Medias, _ = streamer.UnmarshalSDP([]byte(chrome))
-	assert.Len(t, cons.Medias, 3)
-
-	// setup stream with one producer
-	stream := NewStream("fake:")
-
-	// main check:
-	err := stream.AddConsumer(cons)
-	assert.Nil(t, err)
-
-	assert.Len(t, prod.Tracks, 2)
-	assert.Len(t, cons.Tracks, 2)
-
-	time.Sleep(time.Second)
-
-	assert.Greater(t, prod.SendPackets,0)
-	assert.Greater(t, cons.RecvPackets,0)
-
-	assert.Greater(t, prod.RecvPackets,0)
-	assert.Greater(t, cons.SendPackets,0)
-}

cmd/streams/streams.go

@@ -54,27 +54,61 @@ func GetOrNew(src string) *Stream {
 }
 
 func streamsHandler(w http.ResponseWriter, r *http.Request) {
-	src := r.URL.Query().Get("src")
+	query := r.URL.Query()
+	src := query.Get("src")
 
-	switch r.Method {
-	case "PUT":
-		name := r.URL.Query().Get("name")
-		if name == "" {
-			name = src
-		}
-		New(name, src)
-		return
-	case "DELETE":
-		delete(streams, src)
+	// without source - return all streams list
+	if src == "" && r.Method != "POST" {
+		_ = json.NewEncoder(w).Encode(streams)
 		return
 	}
 
-	if src != "" {
+	// Not sure about all this API. Should be rewrited...
+	switch r.Method {
+	case "GET":
 		e := json.NewEncoder(w)
 		e.SetIndent("", "  ")
 		_ = e.Encode(streams[src])
-	} else {
-		_ = json.NewEncoder(w).Encode(streams)
+
+	case "PUT":
+		name := query.Get("name")
+		if name == "" {
+			name = src
+		}
+
+		New(name, src)
+
+	case "PATCH":
+		name := query.Get("name")
+		if name == "" {
+			http.Error(w, "", http.StatusBadRequest)
+			return
+		}
+
+		if stream := Get(name); stream != nil {
+			stream.SetSource(src)
+		} else {
+			New(name, src)
+		}
+
+	case "POST":
+		// with dst - redirect source to dst
+		if dst := query.Get("dst"); dst != "" {
+			if stream := Get(dst); stream != nil {
+				if err := stream.Play(src); err != nil {
+					http.Error(w, err.Error(), http.StatusInternalServerError)
+				} else {
+					_ = json.NewEncoder(w).Encode(stream)
+				}
+			} else {
+				http.Error(w, "", http.StatusNotFound)
+			}
+		} else {
+			http.Error(w, "", http.StatusBadRequest)
+		}
+
+	case "DELETE":
+		delete(streams, src)
 	}
 }
 

cmd/tapo/tapo.go

@@ -0,0 +1,19 @@
+package tapo
+
+import (
+	"github.com/AlexxIT/go2rtc/cmd/streams"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/AlexxIT/go2rtc/pkg/tapo"
+)
+
+func Init() {
+	streams.HandleFunc("tapo", handle)
+}
+
+func handle(url string) (streamer.Producer, error) {
+	conn := tapo.NewClient(url)
+	if err := conn.Dial(); err != nil {
+		return nil, err
+	}
+	return conn, nil
+}

main.go

@@ -4,6 +4,7 @@ import (
 	"github.com/AlexxIT/go2rtc/cmd/api"
 	"github.com/AlexxIT/go2rtc/cmd/app"
 	"github.com/AlexxIT/go2rtc/cmd/debug"
+	"github.com/AlexxIT/go2rtc/cmd/dvrip"
 	"github.com/AlexxIT/go2rtc/cmd/echo"
 	"github.com/AlexxIT/go2rtc/cmd/exec"
 	"github.com/AlexxIT/go2rtc/cmd/ffmpeg"
@@ -14,11 +15,13 @@ import (
 	"github.com/AlexxIT/go2rtc/cmd/ivideon"
 	"github.com/AlexxIT/go2rtc/cmd/mjpeg"
 	"github.com/AlexxIT/go2rtc/cmd/mp4"
+	"github.com/AlexxIT/go2rtc/cmd/mpegts"
 	"github.com/AlexxIT/go2rtc/cmd/ngrok"
 	"github.com/AlexxIT/go2rtc/cmd/rtmp"
 	"github.com/AlexxIT/go2rtc/cmd/rtsp"
 	"github.com/AlexxIT/go2rtc/cmd/srtp"
 	"github.com/AlexxIT/go2rtc/cmd/streams"
+	"github.com/AlexxIT/go2rtc/cmd/tapo"
 	"github.com/AlexxIT/go2rtc/cmd/webrtc"
 	"os"
 	"os/signal"
@@ -37,6 +40,10 @@ func main() {
 	hass.Init()   // add support hass scheme
 	echo.Init()
 	ivideon.Init()
+	http.Init()
+	dvrip.Init()
+	tapo.Init()
+	mpegts.Init()
 
 	srtp.Init()
 	homekit.Init()
@@ -45,7 +52,6 @@ func main() {
 	mp4.Init()
 	hls.Init()
 	mjpeg.Init()
-	http.Init()
 
 	ngrok.Init()
 	debug.Init()

pkg/dvrip/client.go

@@ -0,0 +1,427 @@
+package dvrip
+
+import (
+	"bufio"
+	"crypto/md5"
+	"encoding/base64"
+	"encoding/binary"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"github.com/AlexxIT/go2rtc/pkg/h264"
+	"github.com/AlexxIT/go2rtc/pkg/h265"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/pion/rtp"
+	"io"
+	"net"
+	"net/url"
+	"time"
+)
+
+type Client struct {
+	streamer.Element
+
+	uri     string
+	conn    net.Conn
+	reader  *bufio.Reader
+	session uint32
+	seq     uint32
+	stream  string
+
+	medias     []*streamer.Media
+	videoTrack *streamer.Track
+	audioTrack *streamer.Track
+
+	videoTS  uint32
+	videoDT  uint32
+	audioTS  uint32
+	audioSeq uint16
+}
+
+type Response map[string]interface{}
+
+const Login = uint16(1000)
+const OPMonitorClaim = uint16(1413)
+const OPMonitorStart = uint16(1410)
+
+func NewClient(url string) *Client {
+	return &Client{uri: url}
+}
+
+func (c *Client) Dial() (err error) {
+	u, err := url.Parse(c.uri)
+	if err != nil {
+		return
+	}
+
+	if u.Port() == "" {
+		// add default TCP port
+		u.Host += ":34567"
+	}
+
+	c.conn, err = net.DialTimeout("tcp", u.Host, time.Second*3)
+	if err != nil {
+		return
+	}
+
+	c.reader = bufio.NewReader(c.conn)
+
+	query := u.Query()
+	channel := query.Get("channel")
+	if channel == "" {
+		channel = "0"
+	}
+
+	subtype := query.Get("subtype")
+	switch subtype {
+	case "", "0":
+		subtype = "Main"
+	case "1":
+		subtype = "Extra1"
+	}
+
+	c.stream = fmt.Sprintf(
+		`{"Channel":%s,"CombinMode":"NONE","StreamType":"%s","TransMode":"TCP"}`,
+		channel, subtype,
+	)
+
+	if u.User != nil {
+		pass, _ := u.User.Password()
+		return c.Login(u.User.Username(), pass)
+	} else {
+		return c.Login("admin", "admin")
+	}
+}
+
+func (c *Client) Login(user, pass string) (err error) {
+	data := fmt.Sprintf(
+		`{"EncryptType":"MD5","LoginType":"DVRIP-Web","PassWord":"%s","UserName":"%s"}`,
+		SofiaHash(pass), user,
+	)
+
+	if err = c.Request(Login, data); err != nil {
+		return
+	}
+
+	_, err = c.ResponseJSON()
+	return
+}
+
+func (c *Client) Play() (err error) {
+	format := `{"Name":"OPMonitor","SessionID":"0x%08X","OPMonitor":{"Action":"%s","Parameter":%s}}`
+
+	data := fmt.Sprintf(format, c.session, "Claim", c.stream)
+	if err = c.Request(OPMonitorClaim, data); err != nil {
+		return
+	}
+	if _, err = c.ResponseJSON(); err != nil {
+		return
+	}
+
+	data = fmt.Sprintf(format, c.session, "Start", c.stream)
+	return c.Request(OPMonitorStart, data)
+}
+
+func (c *Client) Handle() error {
+	var buf []byte
+	var size int
+
+	var probe byte
+	if c.medias == nil {
+		probe = 1
+	}
+
+	for {
+		b, err := c.Response()
+		if err != nil {
+			return err
+		}
+
+		// collect data from multiple packets
+		if size > 0 {
+			buf = append(buf, b...)
+			if len(buf) < size {
+				continue
+			}
+			if len(buf) > size {
+				return errors.New("wrong size")
+			}
+			b = buf
+		}
+
+		dataType := binary.BigEndian.Uint32(b)
+		switch dataType {
+		case 0x1FC, 0x1FE:
+			size = int(binary.LittleEndian.Uint32(b[12:])) + 16
+		case 0x1FD: // PFrame
+			size = int(binary.LittleEndian.Uint32(b[4:])) + 8
+		case 0x1FA, 0x1F9:
+			size = int(binary.LittleEndian.Uint16(b[6:])) + 8
+		default:
+			return fmt.Errorf("unknown type: %X", dataType)
+		}
+
+		if len(b) < size {
+			buf = b
+			continue // need to collect data from next packets
+		}
+
+		//log.Printf("[DVR] type: %d, len: %d", dataType, len(b))
+
+		switch dataType {
+		case 0x1FC, 0x1FE: // video IFrame
+			payload := h264.AnnexB2AVC(b[16:])
+
+			if c.videoTrack == nil {
+				fps := b[5]
+				//width := uint16(b[6]) * 8
+				//height := uint16(b[7]) * 8
+				//println(width, height)
+				ts := b[8:]
+
+				// the exact value of the start TS does not matter
+				c.videoTS = binary.LittleEndian.Uint32(ts)
+				c.videoDT = 90000 / uint32(fps)
+
+				c.AddVideoTrack(b[4], payload)
+			}
+
+			if c.videoTrack != nil {
+				c.videoTS += c.videoDT
+
+				packet := &rtp.Packet{
+					Header:  rtp.Header{Timestamp: c.videoTS},
+					Payload: payload,
+				}
+
+				//log.Printf("[AVC] %v, len: %d, ts: %10d", h265.Types(payload), len(payload), packet.Timestamp)
+
+				_ = c.videoTrack.WriteRTP(packet)
+			}
+
+		case 0x1FD: // PFrame
+			if c.videoTrack != nil {
+				c.videoTS += c.videoDT
+
+				packet := &rtp.Packet{
+					Header:  rtp.Header{Timestamp: c.videoTS},
+					Payload: h264.AnnexB2AVC(b[8:]),
+				}
+
+				//log.Printf("[DVR] %v, len: %d, ts: %10d", h265.Types(packet.Payload), len(packet.Payload), packet.Timestamp)
+
+				_ = c.videoTrack.WriteRTP(packet)
+			}
+
+		case 0x1FA, 0x1F9: // audio
+			if c.audioTrack == nil {
+				// the exact value of the start TS does not matter
+				c.audioTS = c.videoTS
+
+				c.AddAudioTrack(b[4], b[5])
+			}
+
+			if c.audioTrack != nil {
+				for b != nil {
+					payload := b[8:size]
+					if len(b) > size {
+						b = b[size:]
+					} else {
+						b = nil
+					}
+
+					c.audioTS += uint32(len(payload))
+					c.audioSeq++
+
+					packet := &rtp.Packet{
+						Header: rtp.Header{
+							Version:        2,
+							Marker:         true,
+							SequenceNumber: c.audioSeq,
+							Timestamp:      c.audioTS,
+						},
+						Payload: payload,
+					}
+
+					//log.Printf("[DVR] len: %d, ts: %10d", len(packet.Payload), packet.Timestamp)
+
+					_ = c.audioTrack.WriteRTP(packet)
+				}
+			}
+		}
+
+		if probe != 0 {
+			probe++
+			if (c.videoTS > 0 && c.audioTS > 0) || probe == 20 {
+				return nil
+			}
+		}
+
+		size = 0
+	}
+}
+
+func (c *Client) Close() error {
+	return c.conn.Close()
+}
+
+func (c *Client) Request(cmd uint16, data string) (err error) {
+	b := make([]byte, 20, 128)
+	b[0] = 255
+	binary.LittleEndian.PutUint32(b[4:], c.session)
+	binary.LittleEndian.PutUint32(b[8:], c.seq)
+	binary.LittleEndian.PutUint16(b[14:], cmd)
+	binary.LittleEndian.PutUint32(b[16:], uint32(len(data))+2)
+	b = append(b, data...)
+	b = append(b, 0x0A, 0x00)
+
+	c.seq++
+
+	if err = c.conn.SetWriteDeadline(time.Now().Add(time.Second * 5)); err != nil {
+		return
+	}
+
+	_, err = c.conn.Write(b)
+	return
+}
+
+func (c *Client) Response() (b []byte, err error) {
+	if err = c.conn.SetReadDeadline(time.Now().Add(time.Second * 5)); err != nil {
+		return
+	}
+
+	b = make([]byte, 20)
+	if _, err = io.ReadFull(c.reader, b); err != nil {
+		return
+	}
+
+	if b[0] != 255 {
+		return nil, errors.New("read error")
+	}
+
+	c.session = binary.LittleEndian.Uint32(b[4:])
+	size := binary.LittleEndian.Uint32(b[16:])
+
+	b = make([]byte, size)
+	if _, err = io.ReadFull(c.reader, b); err != nil {
+		return
+	}
+
+	return
+}
+
+func (c *Client) ResponseJSON() (res Response, err error) {
+	b, err := c.Response()
+	if err != nil {
+		return
+	}
+
+	res = Response{}
+	if err = json.Unmarshal(b[:len(b)-2], &res); err != nil {
+		return
+	}
+
+	if v, ok := res["Ret"].(float64); !ok || (v != 100 && v != 515) {
+		err = fmt.Errorf("wrong response: %s", b)
+	}
+	return
+}
+
+func (c *Client) AddVideoTrack(mediaCode byte, payload []byte) {
+	var codec *streamer.Codec
+	switch mediaCode {
+	case 2:
+		codec = &streamer.Codec{
+			Name:        streamer.CodecH264,
+			ClockRate:   90000,
+			PayloadType: streamer.PayloadTypeRAW,
+			FmtpLine:    h264.GetFmtpLine(payload),
+		}
+
+	case 0x03, 0x13:
+		codec = &streamer.Codec{
+			Name:        streamer.CodecH265,
+			ClockRate:   90000,
+			PayloadType: streamer.PayloadTypeRAW,
+			FmtpLine:    "profile-id=1",
+		}
+
+		for {
+			size := 4 + int(binary.BigEndian.Uint32(payload))
+
+			switch h265.NALUType(payload) {
+			case h265.NALUTypeVPS:
+				codec.FmtpLine += ";sprop-vps=" + base64.StdEncoding.EncodeToString(payload[4:size])
+			case h265.NALUTypeSPS:
+				codec.FmtpLine += ";sprop-sps=" + base64.StdEncoding.EncodeToString(payload[4:size])
+			case h265.NALUTypePPS:
+				codec.FmtpLine += ";sprop-pps=" + base64.StdEncoding.EncodeToString(payload[4:size])
+			}
+
+			if size < len(payload) {
+				payload = payload[size:]
+			} else {
+				break
+			}
+		}
+	default:
+		println("[DVRIP] unsupported video codec:", mediaCode)
+		return
+	}
+
+	media := &streamer.Media{
+		Kind:      streamer.KindVideo,
+		Direction: streamer.DirectionSendonly,
+		Codecs:    []*streamer.Codec{codec},
+	}
+	c.medias = append(c.medias, media)
+
+	c.videoTrack = streamer.NewTrack(codec, media.Direction)
+}
+
+var sampleRates = []uint32{4000, 8000, 11025, 16000, 20000, 22050, 32000, 44100, 48000}
+
+func (c *Client) AddAudioTrack(mediaCode byte, sampleRate byte) {
+	// https://github.com/vigoss30611/buildroot-ltc/blob/master/system/qm/ipc/ProtocolService/src/ZhiNuo/inc/zn_dh_base_type.h
+	// PCM8 = 7, G729, IMA_ADPCM, G711U, G721, PCM8_VWIS, MS_ADPCM, G711A, PCM16
+	var codec *streamer.Codec
+	switch mediaCode {
+	case 10: // G711U
+		codec = &streamer.Codec{
+			Name: streamer.CodecPCMU,
+		}
+	case 14: // G711A
+		codec = &streamer.Codec{
+			Name: streamer.CodecPCMA,
+		}
+	default:
+		println("[DVRIP] unsupported audio codec:", mediaCode)
+		return
+	}
+
+	if sampleRate <= byte(len(sampleRates)) {
+		codec.ClockRate = sampleRates[sampleRate-1]
+	}
+
+	media := &streamer.Media{
+		Kind:      streamer.KindAudio,
+		Direction: streamer.DirectionSendonly,
+		Codecs:    []*streamer.Codec{codec},
+	}
+	c.medias = append(c.medias, media)
+
+	c.audioTrack = streamer.NewTrack(codec, media.Direction)
+}
+
+func SofiaHash(password string) string {
+	const chars = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
+
+	sofia := make([]byte, 0, 8)
+	hash := md5.Sum([]byte(password))
+	for i := 0; i < md5.Size; i += 2 {
+		j := uint16(hash[i]) + uint16(hash[i+1])
+		sofia = append(sofia, chars[j%62])
+	}
+
+	return string(sofia)
+}

pkg/dvrip/producer.go

@@ -0,0 +1,25 @@
+package dvrip
+
+import "github.com/AlexxIT/go2rtc/pkg/streamer"
+
+func (c *Client) GetMedias() []*streamer.Media {
+	return c.medias
+}
+
+func (c *Client) GetTrack(media *streamer.Media, codec *streamer.Codec) *streamer.Track {
+	if c.videoTrack != nil && c.videoTrack.Codec == codec {
+		return c.videoTrack
+	}
+	if c.audioTrack != nil && c.audioTrack.Codec == codec {
+		return c.audioTrack
+	}
+	return nil
+}
+
+func (c *Client) Start() error {
+	return c.Handle()
+}
+
+func (c *Client) Stop() error {
+	return c.Close()
+}

pkg/h264/README.md

@@ -1,38 +1,6 @@
 # H264
 
-Access Unit (AU) can contain one or multiple NAL Unit:
-
-1. [SEI,] SPS, PPS, IFrame, [IFrame...]
-2. BFrame, [BFrame...]
-3. IFrame, [IFrame...]
-
-## RTP H264
-
-Camera | NALu
--------|-----
-EZVIZ C3S          | 7f, 8f, 28:28:28 -> 5t, 28:28:28 -> 1t, 1t, 1t, 1t
-Sonoff GK-200MP2-B | 28:28:28 -> 5t, 1t, 1t, 1t
-Dahua IPC-K42      | 7f, 8f, 28:28:28 -> 5t, 28:28:28 -> 1t, 28:28:28 -> 1t
-FFmpeg copy        | 5t, 1t, 1t, 28:28:28 -> 1t, 28:28:28 -> 1t
-FFmpeg h264        | 24 -> 6:5:5:5:5t, 24 -> 1:1:1:1t, 28:28:28 -> 5f, 28:28:28 -> 5f, 28:28:28 -> 5t
-FFmpeg resize      | 6f, 28:28:28 -> 5f, 28... -> 5t, 24 -> 1:1f, 24 -> 1:1t
-
-## WebRTC
-
-Video codec	    | Media string | Device
-----------------|--------------|-------
-H.264/baseline! | avc1.42E0xx  | Chromecast
-H.264/baseline! | avc1.42E0xx  | Chrome/Safari WebRTC
-H.264/baseline! | avc1.42C0xx  | FFmpeg ultrafast
-H.264/baseline! | avc1.4240xx  | Dahua H264B
-H.264/baseline  | avc1.4200xx  | Chrome WebRTC
-H.264/main!     | avc1.4D40xx  | Chromecast
-H.264/main!     | avc1.4D40xx  | FFmpeg superfast main
-H.264/main!     | avc1.4D40xx  | Dahua H264
-H.264/main      | avc1.4D00xx  | Chrome WebRTC
-H.264/high!     | avc1.640Cxx  | Safari WebRTC
-H.264/high      | avc1.6400xx  | Chromecast
-H.264/high      | avc1.6400xx  | FFmpeg superfast
+Payloader code taken from [pion](https://github.com/pion/rtp) library. And changed to AVC packets support.
 
 ## Useful Links
 

pkg/h264/avc.go

@@ -1,11 +1,31 @@
 package h264
 
 import (
+	"bytes"
 	"encoding/binary"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/pion/rtp"
 )
 
+func AnnexB2AVC(b []byte) []byte {
+	for i := 0; i < len(b); {
+		if i+4 >= len(b) {
+			break
+		}
+
+		size := bytes.Index(b[i+4:], []byte{0, 0, 0, 1})
+		if size < 0 {
+			size = len(b) - (i + 4)
+		}
+
+		binary.BigEndian.PutUint32(b[i:], uint32(size))
+
+		i += size + 4
+	}
+
+	return b
+}
+
 func EncodeAVC(nals ...[]byte) (avc []byte) {
 	var i, n int
 

pkg/h264/helper.go

@@ -104,3 +104,26 @@ func GetParameterSet(fmtp string) (sps, pps []byte) {
 
 	return
 }
+
+// GetFmtpLine from SPS+PPS+IFrame in AVC format
+func GetFmtpLine(avc []byte) string {
+	s := "packetization-mode=1"
+
+	for {
+		size := 4 + int(binary.BigEndian.Uint32(avc))
+
+		switch NALUType(avc) {
+		case NALUTypeSPS:
+			s += ";profile-level-id=" + hex.EncodeToString(avc[5:8])
+			s += ";sprop-parameter-sets=" + base64.StdEncoding.EncodeToString(avc[4:size])
+		case NALUTypePPS:
+			s += "," + base64.StdEncoding.EncodeToString(avc[4:size])
+		}
+
+		if size < len(avc) {
+			avc = avc[size:]
+		} else {
+			return s
+		}
+	}
+}

pkg/h264/payloader.go

@@ -31,7 +31,7 @@ const (
 
 //func annexbNALUStartCode() []byte { return []byte{0x00, 0x00, 0x00, 0x01} }
 
-func emitNalus(nals []byte, isAVC bool, emit func([]byte)) {
+func EmitNalus(nals []byte, isAVC bool, emit func([]byte)) {
 	if !isAVC {
 		nextInd := func(nalu []byte, start int) (indStart int, indLen int) {
 			zeroCount := 0
@@ -84,7 +84,7 @@ func (p *Payloader) Payload(mtu uint16, payload []byte) [][]byte {
 		return payloads
 	}
 
-	emitNalus(payload, p.IsAVC, func(nalu []byte) {
+	EmitNalus(payload, p.IsAVC, func(nalu []byte) {
 		if len(nalu) == 0 {
 			return
 		}

pkg/h264/rtp.go

@@ -1,7 +1,6 @@
 package h264
 
 import (
-	"bytes"
 	"encoding/binary"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/pion/rtp"
@@ -83,23 +82,7 @@ func RTPDepay(track *streamer.Track) streamer.WrapperFunc {
 				// some Chinese buggy cameras has single packet with SPS+PPS+IFrame separated by 00 00 00 01
 				// https://github.com/AlexxIT/WebRTC/issues/391
 				// https://github.com/AlexxIT/WebRTC/issues/392
-				for i := 0; i < len(payload); {
-					if i+4 >= len(payload) {
-						break
-					}
-
-					size := bytes.Index(payload[i+4:], []byte{0, 0, 0, 1})
-					if size < 0 {
-						if i == 0 {
-							break
-						}
-						size = len(payload) - (i + 4)
-					}
-
-					binary.BigEndian.PutUint32(payload[i:], uint32(size))
-
-					i += size + 4
-				}
+				AnnexB2AVC(payload)
 			}
 
 			//log.Printf("[AVC] %v, len: %d, ts: %10d, seq: %d", Types(payload), len(payload), packet.Timestamp, packet.SequenceNumber)

pkg/h265/README.md

@@ -1,3 +1,7 @@
+# H265
+
+Payloader code taken from [pion](https://github.com/pion/rtp) library branch [h265](https://github.com/pion/rtp/tree/h265). Because it's still not in release. Thanks to [@kevmo314](https://github.com/kevmo314).
+
 ## Useful links
 
 - https://datatracker.ietf.org/doc/html/rfc7798

pkg/h265/payloader.go

@@ -0,0 +1,300 @@
+package h265
+
+import (
+	"encoding/binary"
+	"github.com/AlexxIT/go2rtc/pkg/h264"
+	"math"
+)
+
+//
+// Network Abstraction Unit Header implementation
+//
+
+const (
+	// sizeof(uint16)
+	h265NaluHeaderSize = 2
+	// https://datatracker.ietf.org/doc/html/rfc7798#section-4.4.2
+	h265NaluAggregationPacketType = 48
+	// https://datatracker.ietf.org/doc/html/rfc7798#section-4.4.3
+	h265NaluFragmentationUnitType = 49
+	// https://datatracker.ietf.org/doc/html/rfc7798#section-4.4.4
+	h265NaluPACIPacketType = 50
+)
+
+// H265NALUHeader is a H265 NAL Unit Header
+// https://datatracker.ietf.org/doc/html/rfc7798#section-1.1.4
+// +---------------+---------------+
+//
+//	|0|1|2|3|4|5|6|7|0|1|2|3|4|5|6|7|
+//	+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+//	|F|   Type    |  LayerID  | TID |
+//	+-------------+-----------------+
+type H265NALUHeader uint16
+
+func newH265NALUHeader(highByte, lowByte uint8) H265NALUHeader {
+	return H265NALUHeader((uint16(highByte) << 8) | uint16(lowByte))
+}
+
+// F is the forbidden bit, should always be 0.
+func (h H265NALUHeader) F() bool {
+	return (uint16(h) >> 15) != 0
+}
+
+// Type of NAL Unit.
+func (h H265NALUHeader) Type() uint8 {
+	// 01111110 00000000
+	const mask = 0b01111110 << 8
+	return uint8((uint16(h) & mask) >> (8 + 1))
+}
+
+// IsTypeVCLUnit returns whether or not the NAL Unit type is a VCL NAL unit.
+func (h H265NALUHeader) IsTypeVCLUnit() bool {
+	// Type is coded on 6 bits
+	const msbMask = 0b00100000
+	return (h.Type() & msbMask) == 0
+}
+
+// LayerID should always be 0 in non-3D HEVC context.
+func (h H265NALUHeader) LayerID() uint8 {
+	// 00000001 11111000
+	const mask = (0b00000001 << 8) | 0b11111000
+	return uint8((uint16(h) & mask) >> 3)
+}
+
+// TID is the temporal identifier of the NAL unit +1.
+func (h H265NALUHeader) TID() uint8 {
+	const mask = 0b00000111
+	return uint8(uint16(h) & mask)
+}
+
+// IsAggregationPacket returns whether or not the packet is an Aggregation packet.
+func (h H265NALUHeader) IsAggregationPacket() bool {
+	return h.Type() == h265NaluAggregationPacketType
+}
+
+// IsFragmentationUnit returns whether or not the packet is a Fragmentation Unit packet.
+func (h H265NALUHeader) IsFragmentationUnit() bool {
+	return h.Type() == h265NaluFragmentationUnitType
+}
+
+// IsPACIPacket returns whether or not the packet is a PACI packet.
+func (h H265NALUHeader) IsPACIPacket() bool {
+	return h.Type() == h265NaluPACIPacketType
+}
+
+//
+// Fragmentation Unit implementation
+//
+
+const (
+	// sizeof(uint8)
+	h265FragmentationUnitHeaderSize = 1
+)
+
+// H265FragmentationUnitHeader is a H265 FU Header
+// +---------------+
+// |0|1|2|3|4|5|6|7|
+// +-+-+-+-+-+-+-+-+
+// |S|E|  FuType   |
+// +---------------+
+type H265FragmentationUnitHeader uint8
+
+// S represents the start of a fragmented NAL unit.
+func (h H265FragmentationUnitHeader) S() bool {
+	const mask = 0b10000000
+	return ((h & mask) >> 7) != 0
+}
+
+// E represents the end of a fragmented NAL unit.
+func (h H265FragmentationUnitHeader) E() bool {
+	const mask = 0b01000000
+	return ((h & mask) >> 6) != 0
+}
+
+// FuType MUST be equal to the field Type of the fragmented NAL unit.
+func (h H265FragmentationUnitHeader) FuType() uint8 {
+	const mask = 0b00111111
+	return uint8(h) & mask
+}
+
+// Payloader payloads H265 packets
+type Payloader struct {
+	AddDONL         bool
+	SkipAggregation bool
+	donl            uint16
+}
+
+// Payload fragments a H265 packet across one or more byte arrays
+func (p *Payloader) Payload(mtu uint16, payload []byte) [][]byte {
+	var payloads [][]byte
+	if len(payload) == 0 {
+		return payloads
+	}
+
+	bufferedNALUs := make([][]byte, 0)
+	aggregationBufferSize := 0
+
+	flushBufferedNals := func() {
+		if len(bufferedNALUs) == 0 {
+			return
+		}
+		if len(bufferedNALUs) == 1 {
+			// emit this as a single NALU packet
+			nalu := bufferedNALUs[0]
+
+			if p.AddDONL {
+				buf := make([]byte, len(nalu)+2)
+
+				// copy the NALU header to the payload header
+				copy(buf[0:h265NaluHeaderSize], nalu[0:h265NaluHeaderSize])
+
+				// copy the DONL into the header
+				binary.BigEndian.PutUint16(buf[h265NaluHeaderSize:h265NaluHeaderSize+2], p.donl)
+
+				// write the payload
+				copy(buf[h265NaluHeaderSize+2:], nalu[h265NaluHeaderSize:])
+
+				p.donl++
+
+				payloads = append(payloads, buf)
+			} else {
+				// write the nalu directly to the payload
+				payloads = append(payloads, nalu)
+			}
+		} else {
+			// construct an aggregation packet
+			aggregationPacketSize := aggregationBufferSize + 2
+			buf := make([]byte, aggregationPacketSize)
+
+			layerID := uint8(math.MaxUint8)
+			tid := uint8(math.MaxUint8)
+			for _, nalu := range bufferedNALUs {
+				header := newH265NALUHeader(nalu[0], nalu[1])
+				headerLayerID := header.LayerID()
+				headerTID := header.TID()
+				if headerLayerID < layerID {
+					layerID = headerLayerID
+				}
+				if headerTID < tid {
+					tid = headerTID
+				}
+			}
+
+			binary.BigEndian.PutUint16(buf[0:2], (uint16(h265NaluAggregationPacketType)<<9)|(uint16(layerID)<<3)|uint16(tid))
+
+			index := 2
+			for i, nalu := range bufferedNALUs {
+				if p.AddDONL {
+					if i == 0 {
+						binary.BigEndian.PutUint16(buf[index:index+2], p.donl)
+						index += 2
+					} else {
+						buf[index] = byte(i - 1)
+						index++
+					}
+				}
+				binary.BigEndian.PutUint16(buf[index:index+2], uint16(len(nalu)))
+				index += 2
+				index += copy(buf[index:], nalu)
+			}
+			payloads = append(payloads, buf)
+		}
+		// clear the buffered NALUs
+		bufferedNALUs = make([][]byte, 0)
+		aggregationBufferSize = 0
+	}
+
+	h264.EmitNalus(payload, true, func(nalu []byte) {
+		if len(nalu) == 0 {
+			return
+		}
+
+		if len(nalu) <= int(mtu) {
+			// this nalu fits into a single packet, either it can be emitted as
+			// a single nalu or appended to the previous aggregation packet
+
+			marginalAggregationSize := len(nalu) + 2
+			if p.AddDONL {
+				marginalAggregationSize += 1
+			}
+
+			if aggregationBufferSize+marginalAggregationSize > int(mtu) {
+				flushBufferedNals()
+			}
+			bufferedNALUs = append(bufferedNALUs, nalu)
+			aggregationBufferSize += marginalAggregationSize
+			if p.SkipAggregation {
+				// emit this immediately.
+				flushBufferedNals()
+			}
+		} else {
+			// if this nalu doesn't fit in the current mtu, it needs to be fragmented
+			fuPacketHeaderSize := h265FragmentationUnitHeaderSize + 2 /* payload header size */
+			if p.AddDONL {
+				fuPacketHeaderSize += 2
+			}
+
+			// then, fragment the nalu
+			maxFUPayloadSize := int(mtu) - fuPacketHeaderSize
+
+			naluHeader := newH265NALUHeader(nalu[0], nalu[1])
+
+			// the nalu header is omitted from the fragmentation packet payload
+			nalu = nalu[h265NaluHeaderSize:]
+
+			if maxFUPayloadSize == 0 || len(nalu) == 0 {
+				return
+			}
+
+			// flush any buffered aggregation packets.
+			flushBufferedNals()
+
+			fullNALUSize := len(nalu)
+			for len(nalu) > 0 {
+				curentFUPayloadSize := len(nalu)
+				if curentFUPayloadSize > maxFUPayloadSize {
+					curentFUPayloadSize = maxFUPayloadSize
+				}
+
+				out := make([]byte, fuPacketHeaderSize+curentFUPayloadSize)
+
+				// write the payload header
+				binary.BigEndian.PutUint16(out[0:2], uint16(naluHeader))
+				out[0] = (out[0] & 0b10000001) | h265NaluFragmentationUnitType<<1
+
+				// write the fragment header
+				out[2] = byte(H265FragmentationUnitHeader(naluHeader.Type()))
+				if len(nalu) == fullNALUSize {
+					// Set start bit
+					out[2] |= 1 << 7
+				} else if len(nalu)-curentFUPayloadSize == 0 {
+					// Set end bit
+					out[2] |= 1 << 6
+				}
+
+				if p.AddDONL {
+					// write the DONL header
+					binary.BigEndian.PutUint16(out[3:5], p.donl)
+
+					p.donl++
+
+					// copy the fragment payload
+					copy(out[5:], nalu[0:curentFUPayloadSize])
+				} else {
+					// copy the fragment payload
+					copy(out[3:], nalu[0:curentFUPayloadSize])
+				}
+
+				// append the fragment to the payload
+				payloads = append(payloads, out)
+
+				// advance the nalu data pointer
+				nalu = nalu[curentFUPayloadSize:]
+			}
+		}
+	})
+
+	flushBufferedNals()
+
+	return payloads
+}

pkg/h265/rtp.go

@@ -77,6 +77,39 @@ func RTPDepay(track *streamer.Track) streamer.WrapperFunc {
 	}
 }
 
+func RTPPay(mtu uint16) streamer.WrapperFunc {
+	payloader := &Payloader{}
+	sequencer := rtp.NewRandomSequencer()
+	mtu -= 12 // rtp.Header size
+
+	return func(push streamer.WriterFunc) streamer.WriterFunc {
+		return func(packet *rtp.Packet) error {
+			if packet.Version != h264.RTPPacketVersionAVC {
+				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
+		}
+	}
+}
+
 // SafariPay - generate Safari friendly payload for H265
 // https://github.com/AlexxIT/Blog/issues/5
 func SafariPay(mtu uint16) streamer.WrapperFunc {

pkg/mjpeg/client.go

@@ -2,7 +2,6 @@ package mjpeg
 
 import (
 	"bufio"
-	"encoding/json"
 	"errors"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/AlexxIT/go2rtc/pkg/tcp"
@@ -25,63 +24,13 @@ type Client struct {
 	closed bool
 	res    *http.Response
 
-	track *streamer.Track
-	recv  uint32
+	medias []*streamer.Media
+	track  *streamer.Track
+	recv   uint32
 }
 
 func NewClient(res *http.Response) *Client {
-	codec := &streamer.Codec{
-		Name: streamer.CodecJPEG, ClockRate: 90000, PayloadType: streamer.PayloadTypeRAW,
-	}
-	return &Client{
-		res:   res,
-		track: streamer.NewTrack(codec, streamer.DirectionSendonly),
-	}
-}
-
-func (c *Client) GetMedias() []*streamer.Media {
-	return []*streamer.Media{{
-		Kind:      streamer.KindVideo,
-		Direction: streamer.DirectionSendonly,
-		Codecs:    []*streamer.Codec{c.track.Codec},
-	}}
-}
-
-func (c *Client) GetTrack(media *streamer.Media, codec *streamer.Codec) *streamer.Track {
-	return c.track
-}
-
-func (c *Client) Start() error {
-	ct := c.res.Header.Get("Content-Type")
-
-	if ct == "image/jpeg" {
-		return c.startJPEG()
-	}
-
-	// added in go1.18
-	if _, s, ok := strings.Cut(ct, "boundary="); ok {
-		return c.startMJPEG(s)
-	}
-
-	return errors.New("wrong Content-Type: " + ct)
-}
-
-func (c *Client) Stop() error {
-	// important for close reader/writer gorutines
-	_ = c.res.Body.Close()
-	c.closed = true
-	return nil
-}
-
-func (c *Client) MarshalJSON() ([]byte, error) {
-	info := &streamer.Info{
-		Type:       "MJPEG source",
-		URL:        c.res.Request.URL.String(),
-		RemoteAddr: c.RemoteAddr,
-		UserAgent:  c.UserAgent,
-		Recv:       atomic.LoadUint32(&c.recv),
-	}
-	return json.Marshal(info)
+	return &Client{res: res}
 }
 
 func (c *Client) startJPEG() error {
@@ -136,7 +85,7 @@ func (c *Client) startMJPEG(boundary string) error {
 		if err != nil {
 			return err
 		}
-		if s != boundary {
+		if !strings.HasPrefix(s, boundary) {
 			return errors.New("wrong boundary: " + s)
 		}
 

pkg/mjpeg/producer.go

@@ -0,0 +1,64 @@
+package mjpeg
+
+import (
+	"encoding/json"
+	"errors"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"strings"
+	"sync/atomic"
+)
+
+func (c *Client) GetMedias() []*streamer.Media {
+	if c.medias == nil {
+		c.medias = []*streamer.Media{{
+			Kind:      streamer.KindVideo,
+			Direction: streamer.DirectionSendonly,
+			Codecs: []*streamer.Codec{
+				{
+					Name: streamer.CodecJPEG, ClockRate: 90000, PayloadType: streamer.PayloadTypeRAW,
+				},
+			},
+		}}
+	}
+	return c.medias
+}
+
+func (c *Client) GetTrack(media *streamer.Media, codec *streamer.Codec) *streamer.Track {
+	if c.track == nil {
+		c.track = streamer.NewTrack(codec, streamer.DirectionSendonly)
+	}
+	return c.track
+}
+
+func (c *Client) Start() error {
+	ct := c.res.Header.Get("Content-Type")
+
+	if ct == "image/jpeg" {
+		return c.startJPEG()
+	}
+
+	// added in go1.18
+	if _, s, ok := strings.Cut(ct, "boundary="); ok {
+		return c.startMJPEG(s)
+	}
+
+	return errors.New("wrong Content-Type: " + ct)
+}
+
+func (c *Client) Stop() error {
+	// important for close reader/writer gorutines
+	_ = c.res.Body.Close()
+	c.closed = true
+	return nil
+}
+
+func (c *Client) MarshalJSON() ([]byte, error) {
+	info := &streamer.Info{
+		Type:       "MJPEG source",
+		URL:        c.res.Request.URL.String(),
+		RemoteAddr: c.RemoteAddr,
+		UserAgent:  c.UserAgent,
+		Recv:       atomic.LoadUint32(&c.recv),
+	}
+	return json.Marshal(info)
+}

pkg/mjpeg/rtp.go

@@ -1,8 +1,12 @@
 package mjpeg
 
 import (
+	"bytes"
+	"encoding/binary"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/pion/rtp"
+	"image"
+	"image/jpeg"
 )
 
 func RTPDepay(track *streamer.Track) streamer.WrapperFunc {
@@ -82,144 +86,142 @@ func RTPDepay(track *streamer.Track) streamer.WrapperFunc {
 	}
 }
 
+func cutSize(size uint16) uint16 {
+	return ((size >> 3) & 0xFF) << 3
+}
+
 func RTPPay() streamer.WrapperFunc {
+	const packetSize = 1436
+
+	sequencer := rtp.NewRandomSequencer()
+
 	return func(push streamer.WriterFunc) streamer.WriterFunc {
 		return func(packet *rtp.Packet) error {
+			// reincode image to more common form
+			p, err := Transcode(packet.Payload)
+			if err != nil {
+				return err
+			}
+
+			h1 := make([]byte, 8)
+			h1[4] = 1   // Type
+			h1[5] = 255 // Q
+
+			// MBZ=0, Precision=0, Length=128
+			h2 := make([]byte, 4, 132)
+			h2[3] = 128
+
+			var jpgData []byte
+			for jpgData == nil {
+				// 2 bytes h1
+				if p[0] != 0xFF {
+					return nil
+				}
+
+				size := binary.BigEndian.Uint16(p[2:]) + 2
+
+				// 2 bytes payload size (include 2 bytes)
+				switch p[1] {
+				case 0xD8: // 0. Start Of Image (size=0)
+					p = p[2:]
+					continue
+				case 0xDB: // 1. Define Quantization Table (size=130)
+					for i := uint16(4 + 1); i < size; i += 1 + 64 {
+						h2 = append(h2, p[i:i+64]...)
+					}
+				case 0xC0: // 2. Start Of Frame (size=15)
+					if p[4] != 8 {
+						return nil
+					}
+					h := binary.BigEndian.Uint16(p[5:])
+					w := binary.BigEndian.Uint16(p[7:])
+					h1[6] = uint8(w >> 3)
+					h1[7] = uint8(h >> 3)
+				case 0xC4: // 3. Define Huffman Table (size=416)
+				case 0xDA: // 4. Start Of Scan (size=10)
+					jpgData = p[size:]
+				}
+
+				p = p[size:]
+			}
+
+			offset := 0
+			p = make([]byte, 0)
+
+			for jpgData != nil {
+				p = p[:0]
+
+				if offset > 0 {
+					h1[1] = byte(offset >> 16)
+					h1[2] = byte(offset >> 8)
+					h1[3] = byte(offset)
+					p = append(p, h1...)
+				} else {
+					p = append(p, h1...)
+					p = append(p, h2...)
+				}
+
+				dataLen := packetSize - len(p)
+				if dataLen < len(jpgData) {
+					p = append(p, jpgData[:dataLen]...)
+					jpgData = jpgData[dataLen:]
+					offset += dataLen
+				} else {
+					p = append(p, jpgData...)
+					jpgData = nil
+				}
+
+				clone := rtp.Packet{
+					Header: rtp.Header{
+						Version:        2,
+						Marker:         jpgData == nil,
+						SequenceNumber: sequencer.NextSequenceNumber(),
+						Timestamp:      packet.Timestamp,
+					},
+					Payload: p,
+				}
+				if err := push(&clone); err != nil {
+					return err
+				}
+			}
+
 			return nil
 		}
 	}
 }
 
-func cutSize(size uint16) uint16 {
-	return ((size >> 3) & 0xFF) << 3
-}
+func Transcode(b []byte) ([]byte, error) {
+	img, err := jpeg.Decode(bytes.NewReader(b))
+	if err != nil {
+		return nil, err
+	}
 
-//func RTPPay() streamer.WrapperFunc {
-//	const packetSize = 1436
-//
-//	sequencer := rtp.NewRandomSequencer()
-//
-//	return func(push streamer.WriterFunc) streamer.WriterFunc {
-//		return func(packet *rtp.Packet) error {
-//			// reincode image to more common form
-//			img, err := jpeg.Decode(bytes.NewReader(packet.Payload))
-//			if err != nil {
-//				return err
-//			}
-//
-//			wh := img.Bounds().Size()
-//			w := wh.X
-//			h := wh.Y
-//
-//			if w > 2040 {
-//				w = 2040
-//			} else if w&3 > 0 {
-//				w &= 3
-//			}
-//			if h > 2040 {
-//				h = 2040
-//			} else if h&3 > 0 {
-//				h &= 3
-//			}
-//
-//			if w != wh.X || h != wh.Y {
-//				x0 := (wh.X - w) / 2
-//				y0 := (wh.Y - h) / 2
-//				rect := image.Rect(x0, y0, x0+w, y0+h)
-//				img = img.(*image.YCbCr).SubImage(rect)
-//			}
-//
-//			buf := bytes.NewBuffer(nil)
-//			if err = jpeg.Encode(buf, img, nil); err != nil {
-//				return err
-//			}
-//
-//			h1 := make([]byte, 8)
-//			h1[4] = 1   // Type
-//			h1[5] = 255 // Q
-//
-//			// MBZ=0, Precision=0, Length=128
-//			h2 := make([]byte, 4, 132)
-//			h2[3] = 128
-//
-//			var jpgData []byte
-//
-//			p := buf.Bytes()
-//
-//			for jpgData == nil {
-//				// 2 bytes h1
-//				if p[0] != 0xFF {
-//					return nil
-//				}
-//
-//				size := binary.BigEndian.Uint16(p[2:]) + 2
-//
-//				// 2 bytes payload size (include 2 bytes)
-//				switch p[1] {
-//				case 0xD8: // 0. Start Of Image (size=0)
-//					p = p[2:]
-//					continue
-//				case 0xDB: // 1. Define Quantization Table (size=130)
-//					for i := uint16(4 + 1); i < size; i += 1 + 64 {
-//						h2 = append(h2, p[i:i+64]...)
-//					}
-//				case 0xC0: // 2. Start Of Frame (size=15)
-//					if p[4] != 8 {
-//						return nil
-//					}
-//					h := binary.BigEndian.Uint16(p[5:])
-//					w := binary.BigEndian.Uint16(p[7:])
-//					h1[6] = uint8(w >> 3)
-//					h1[7] = uint8(h >> 3)
-//				case 0xC4: // 3. Define Huffman Table (size=416)
-//				case 0xDA: // 4. Start Of Scan (size=10)
-//					jpgData = p[size:]
-//				}
-//
-//				p = p[size:]
-//			}
-//
-//			offset := 0
-//			p = make([]byte, 0)
-//
-//			for jpgData != nil {
-//				p = p[:0]
-//
-//				if offset > 0 {
-//					h1[1] = byte(offset >> 16)
-//					h1[2] = byte(offset >> 8)
-//					h1[3] = byte(offset)
-//					p = append(p, h1...)
-//				} else {
-//					p = append(p, h1...)
-//					p = append(p, h2...)
-//				}
-//
-//				dataLen := packetSize - len(p)
-//				if dataLen < len(jpgData) {
-//					p = append(p, jpgData[:dataLen]...)
-//					jpgData = jpgData[dataLen:]
-//					offset += dataLen
-//				} else {
-//					p = append(p, jpgData...)
-//					jpgData = nil
-//				}
-//
-//				clone := rtp.Packet{
-//					Header: rtp.Header{
-//						Version:        2,
-//						Marker:         jpgData == nil,
-//						SequenceNumber: sequencer.NextSequenceNumber(),
-//						Timestamp:      packet.Timestamp,
-//					},
-//					Payload: p,
-//				}
-//				if err := push(&clone); err != nil {
-//					return err
-//				}
-//			}
-//
-//			return nil
-//		}
-//	}
-//}
+	wh := img.Bounds().Size()
+	w := wh.X
+	h := wh.Y
+
+	if w > 2040 {
+		w = 2040
+	} else if w&3 > 0 {
+		w &= 3
+	}
+	if h > 2040 {
+		h = 2040
+	} else if h&3 > 0 {
+		h &= 3
+	}
+
+	if w != wh.X || h != wh.Y {
+		x0 := (wh.X - w) / 2
+		y0 := (wh.Y - h) / 2
+		rect := image.Rect(x0, y0, x0+w, y0+h)
+		img = img.(*image.YCbCr).SubImage(rect)
+	}
+
+	buf := bytes.NewBuffer(nil)
+	if err = jpeg.Encode(buf, img, nil); err != nil {
+		return nil, err
+	}
+
+	return buf.Bytes(), nil
+}

pkg/mpegts/README.md

@@ -0,0 +1,5 @@
+## Useful links
+
+- https://github.com/theREDspace/video-onboarding/blob/main/MPEGTS%20Knowledge.md
+- https://en.wikipedia.org/wiki/MPEG_transport_stream
+- https://en.wikipedia.org/wiki/Program-specific_information

pkg/mpegts/checksum.go

@@ -0,0 +1,54 @@
+package mpegts
+
+var ieeeCrc32Tbl = []uint32{
+	0x00000000, 0xB71DC104, 0x6E3B8209, 0xD926430D, 0xDC760413, 0x6B6BC517,
+	0xB24D861A, 0x0550471E, 0xB8ED0826, 0x0FF0C922, 0xD6D68A2F, 0x61CB4B2B,
+	0x649B0C35, 0xD386CD31, 0x0AA08E3C, 0xBDBD4F38, 0x70DB114C, 0xC7C6D048,
+	0x1EE09345, 0xA9FD5241, 0xACAD155F, 0x1BB0D45B, 0xC2969756, 0x758B5652,
+	0xC836196A, 0x7F2BD86E, 0xA60D9B63, 0x11105A67, 0x14401D79, 0xA35DDC7D,
+	0x7A7B9F70, 0xCD665E74, 0xE0B62398, 0x57ABE29C, 0x8E8DA191, 0x39906095,
+	0x3CC0278B, 0x8BDDE68F, 0x52FBA582, 0xE5E66486, 0x585B2BBE, 0xEF46EABA,
+	0x3660A9B7, 0x817D68B3, 0x842D2FAD, 0x3330EEA9, 0xEA16ADA4, 0x5D0B6CA0,
+	0x906D32D4, 0x2770F3D0, 0xFE56B0DD, 0x494B71D9, 0x4C1B36C7, 0xFB06F7C3,
+	0x2220B4CE, 0x953D75CA, 0x28803AF2, 0x9F9DFBF6, 0x46BBB8FB, 0xF1A679FF,
+	0xF4F63EE1, 0x43EBFFE5, 0x9ACDBCE8, 0x2DD07DEC, 0x77708634, 0xC06D4730,
+	0x194B043D, 0xAE56C539, 0xAB068227, 0x1C1B4323, 0xC53D002E, 0x7220C12A,
+	0xCF9D8E12, 0x78804F16, 0xA1A60C1B, 0x16BBCD1F, 0x13EB8A01, 0xA4F64B05,
+	0x7DD00808, 0xCACDC90C, 0x07AB9778, 0xB0B6567C, 0x69901571, 0xDE8DD475,
+	0xDBDD936B, 0x6CC0526F, 0xB5E61162, 0x02FBD066, 0xBF469F5E, 0x085B5E5A,
+	0xD17D1D57, 0x6660DC53, 0x63309B4D, 0xD42D5A49, 0x0D0B1944, 0xBA16D840,
+	0x97C6A5AC, 0x20DB64A8, 0xF9FD27A5, 0x4EE0E6A1, 0x4BB0A1BF, 0xFCAD60BB,
+	0x258B23B6, 0x9296E2B2, 0x2F2BAD8A, 0x98366C8E, 0x41102F83, 0xF60DEE87,
+	0xF35DA999, 0x4440689D, 0x9D662B90, 0x2A7BEA94, 0xE71DB4E0, 0x500075E4,
+	0x892636E9, 0x3E3BF7ED, 0x3B6BB0F3, 0x8C7671F7, 0x555032FA, 0xE24DF3FE,
+	0x5FF0BCC6, 0xE8ED7DC2, 0x31CB3ECF, 0x86D6FFCB, 0x8386B8D5, 0x349B79D1,
+	0xEDBD3ADC, 0x5AA0FBD8, 0xEEE00C69, 0x59FDCD6D, 0x80DB8E60, 0x37C64F64,
+	0x3296087A, 0x858BC97E, 0x5CAD8A73, 0xEBB04B77, 0x560D044F, 0xE110C54B,
+	0x38368646, 0x8F2B4742, 0x8A7B005C, 0x3D66C158, 0xE4408255, 0x535D4351,
+	0x9E3B1D25, 0x2926DC21, 0xF0009F2C, 0x471D5E28, 0x424D1936, 0xF550D832,
+	0x2C769B3F, 0x9B6B5A3B, 0x26D61503, 0x91CBD407, 0x48ED970A, 0xFFF0560E,
+	0xFAA01110, 0x4DBDD014, 0x949B9319, 0x2386521D, 0x0E562FF1, 0xB94BEEF5,
+	0x606DADF8, 0xD7706CFC, 0xD2202BE2, 0x653DEAE6, 0xBC1BA9EB, 0x0B0668EF,
+	0xB6BB27D7, 0x01A6E6D3, 0xD880A5DE, 0x6F9D64DA, 0x6ACD23C4, 0xDDD0E2C0,
+	0x04F6A1CD, 0xB3EB60C9, 0x7E8D3EBD, 0xC990FFB9, 0x10B6BCB4, 0xA7AB7DB0,
+	0xA2FB3AAE, 0x15E6FBAA, 0xCCC0B8A7, 0x7BDD79A3, 0xC660369B, 0x717DF79F,
+	0xA85BB492, 0x1F467596, 0x1A163288, 0xAD0BF38C, 0x742DB081, 0xC3307185,
+	0x99908A5D, 0x2E8D4B59, 0xF7AB0854, 0x40B6C950, 0x45E68E4E, 0xF2FB4F4A,
+	0x2BDD0C47, 0x9CC0CD43, 0x217D827B, 0x9660437F, 0x4F460072, 0xF85BC176,
+	0xFD0B8668, 0x4A16476C, 0x93300461, 0x242DC565, 0xE94B9B11, 0x5E565A15,
+	0x87701918, 0x306DD81C, 0x353D9F02, 0x82205E06, 0x5B061D0B, 0xEC1BDC0F,
+	0x51A69337, 0xE6BB5233, 0x3F9D113E, 0x8880D03A, 0x8DD09724, 0x3ACD5620,
+	0xE3EB152D, 0x54F6D429, 0x7926A9C5, 0xCE3B68C1, 0x171D2BCC, 0xA000EAC8,
+	0xA550ADD6, 0x124D6CD2, 0xCB6B2FDF, 0x7C76EEDB, 0xC1CBA1E3, 0x76D660E7,
+	0xAFF023EA, 0x18EDE2EE, 0x1DBDA5F0, 0xAAA064F4, 0x738627F9, 0xC49BE6FD,
+	0x09FDB889, 0xBEE0798D, 0x67C63A80, 0xD0DBFB84, 0xD58BBC9A, 0x62967D9E,
+	0xBBB03E93, 0x0CADFF97, 0xB110B0AF, 0x060D71AB, 0xDF2B32A6, 0x6836F3A2,
+	0x6D66B4BC, 0xDA7B75B8, 0x035D36B5, 0xB440F7B1, 0x00000001,
+}
+
+func calcCRC32(crc uint32, data []byte) uint32 {
+	for _, b := range data {
+		crc = ieeeCrc32Tbl[b^byte(crc)] ^ (crc >> 8)
+	}
+	return crc
+}

pkg/mpegts/client.go

@@ -0,0 +1,73 @@
+package mpegts
+
+import (
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"net/http"
+)
+
+type Client struct {
+	streamer.Element
+
+	medias []*streamer.Media
+	tracks map[byte]*streamer.Track
+
+	res *http.Response
+}
+
+func NewClient(res *http.Response) *Client {
+	return &Client{res: res}
+}
+
+func (c *Client) Handle() error {
+	if c.tracks == nil {
+		c.tracks = map[byte]*streamer.Track{}
+	}
+
+	reader := NewReader()
+
+	b := make([]byte, 1024*1024*256) // 256K
+
+	probe := streamer.NewProbe(c.medias == nil)
+	for probe == nil || probe.Active() {
+		n, err := c.res.Body.Read(b)
+		if err != nil {
+			return err
+		}
+
+		reader.AppendBuffer(b[:n])
+
+		for {
+			packet := reader.GetPacket()
+			if packet == nil {
+				break
+			}
+
+			track := c.tracks[packet.PayloadType]
+			if track == nil {
+				// count track on probe state even if not support it
+				probe.Append(packet.PayloadType)
+
+				media := GetMedia(packet)
+				if media == nil {
+					continue // unsupported codec
+				}
+
+				track = streamer.NewTrack2(media, nil)
+
+				c.medias = append(c.medias, media)
+				c.tracks[packet.PayloadType] = track
+			}
+
+			_ = track.WriteRTP(packet)
+
+			//log.Printf("[AVC] %v, len: %d, pts: %d ts: %10d", h264.Types(packet.Payload), len(packet.Payload), pkt.PTS, packet.Timestamp)
+		}
+	}
+
+	return nil
+}
+
+func (c *Client) Close() error {
+	_ = c.res.Body.Close()
+	return nil
+}

pkg/mpegts/helpers.go

@@ -0,0 +1,251 @@
+package mpegts
+
+import (
+	"bytes"
+	"github.com/AlexxIT/go2rtc/pkg/h264"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/pion/rtp"
+	"time"
+)
+
+const (
+	PacketSize = 188
+	SyncByte   = 0x47
+)
+
+const (
+	StreamTypePrivate  = 0x06 // PCMU or PCMA from FFmpeg
+	StreamTypeAAC      = 0x0F
+	StreamTypeH264     = 0x1B
+	StreamTypePCMATapo = 0x90
+)
+
+type Packet struct {
+	StreamType byte
+	PTS        time.Duration
+	DTS        time.Duration
+	Payload    []byte
+}
+
+// PES - Packetized Elementary Stream
+type PES struct {
+	StreamType byte
+	StreamID   byte
+	Payload    []byte
+	Mode       byte
+	Size       int
+
+	Sequence  uint16
+	Timestamp uint32
+}
+
+const (
+	ModeUnknown = iota
+	ModeSize
+	ModeStream
+)
+
+// parse Optional PES header
+const minHeaderSize = 3
+
+func (p *PES) SetBuffer(size uint16, b []byte) {
+	if size == 0 {
+		optSize := b[2] // optional fields
+		b = b[minHeaderSize+optSize:]
+
+		if p.StreamType == StreamTypeH264 {
+			if bytes.HasPrefix(b, []byte{0, 0, 0, 1, h264.NALUTypeAUD}) {
+				p.Mode = ModeStream
+				b = b[5:]
+			}
+		}
+
+		if p.Mode == ModeUnknown {
+			println("WARNING: mpegts: unknown zero-size stream")
+		}
+	} else {
+		p.Mode = ModeSize
+		p.Size = int(size)
+	}
+
+	p.Payload = make([]byte, 0, size)
+	p.Payload = append(p.Payload, b...)
+}
+
+func (p *PES) AppendBuffer(b []byte) {
+	p.Payload = append(p.Payload, b...)
+}
+
+func (p *PES) GetPacket() (pkt *rtp.Packet) {
+	switch p.Mode {
+	case ModeSize:
+		left := p.Size - len(p.Payload)
+		if left > 0 {
+			return
+		}
+
+		if left < 0 {
+			println("WARNING: mpegts: buffer overflow")
+			p.Payload = nil
+			return
+		}
+
+		// fist byte also flags
+		flags := p.Payload[1]
+		optSize := p.Payload[2] // optional fields
+
+		payload := p.Payload[minHeaderSize+optSize:]
+
+		switch p.StreamType {
+		case StreamTypeH264:
+			var ts uint32
+
+			const hasPTS = 0b1000_0000
+			if flags&hasPTS != 0 {
+				ts = ParseTime(p.Payload[minHeaderSize:])
+			}
+
+			pkt = &rtp.Packet{
+				Header: rtp.Header{
+					PayloadType: p.StreamType,
+					Timestamp:   ts,
+				},
+				Payload: h264.AnnexB2AVC(payload),
+			}
+
+		case StreamTypePCMATapo:
+			p.Sequence++
+			p.Timestamp += uint32(len(payload))
+
+			pkt = &rtp.Packet{
+				Header: rtp.Header{
+					Version:        2,
+					PayloadType:    p.StreamType,
+					SequenceNumber: p.Sequence,
+					Timestamp:      p.Timestamp,
+				},
+				Payload: payload,
+			}
+		}
+
+		p.Payload = nil
+
+	case ModeStream:
+		i := bytes.Index(p.Payload, []byte{0, 0, 0, 1, h264.NALUTypeAUD})
+		if i < 0 {
+			return
+		}
+		if i2 := IndexFrom(p.Payload, []byte{0, 0, 1}, i); i2 < 0 && i2 > 9 {
+			return
+		}
+
+		pkt = &rtp.Packet{
+			Header: rtp.Header{
+				PayloadType: p.StreamType,
+				Timestamp:   uint32(time.Duration(time.Now().UnixNano()) * 90000 / time.Second),
+			},
+			Payload: DecodeAnnex3B(p.Payload[:i]),
+		}
+
+		p.Payload = p.Payload[i+5:]
+
+	default:
+		p.Payload = nil
+	}
+
+	return
+}
+
+func ParseTime(b []byte) uint32 {
+	return (uint32(b[0]&0x0E) << 29) | (uint32(b[1]) << 22) | (uint32(b[2]&0xFE) << 14) | (uint32(b[3]) << 7) | (uint32(b[4]) >> 1)
+}
+
+func GetMedia(pkt *rtp.Packet) *streamer.Media {
+	var codec *streamer.Codec
+	var kind string
+
+	switch pkt.PayloadType {
+	case StreamTypeH264:
+		codec = &streamer.Codec{
+			Name:        streamer.CodecH264,
+			ClockRate:   90000,
+			PayloadType: streamer.PayloadTypeRAW,
+			FmtpLine:    h264.GetFmtpLine(pkt.Payload),
+		}
+		kind = streamer.KindVideo
+
+	case StreamTypePCMATapo:
+		codec = &streamer.Codec{
+			Name:      streamer.CodecPCMA,
+			ClockRate: 8000,
+		}
+		kind = streamer.KindAudio
+
+	default:
+		return nil
+	}
+
+	return &streamer.Media{
+		Kind:      kind,
+		Direction: streamer.DirectionSendonly,
+		Codecs:    []*streamer.Codec{codec},
+	}
+}
+
+func DecodeAnnex3B(annexb []byte) (avc []byte) {
+	// depends on AU delimeter size
+	i0 := bytes.Index(annexb, []byte{0, 0, 1})
+	if i0 < 0 || i0 > 9 {
+		return nil
+	}
+
+	annexb = annexb[i0+3:] // skip first separator
+	i0 = 0
+
+	for {
+		// search next separato
+		iN := IndexFrom(annexb, []byte{0, 0, 1}, i0)
+		if iN < 0 {
+			break
+		}
+
+		// move i0 to next AU
+		if i0 = iN + 3; i0 >= len(annexb) {
+			break
+		}
+
+		// check if AU type valid
+		octet := annexb[i0]
+		const forbiddenZeroBit = 0x80
+		if octet&forbiddenZeroBit == 0 {
+			const nalUnitType = 0x1F
+			switch octet & nalUnitType {
+			case h264.NALUTypePFrame, h264.NALUTypeIFrame, h264.NALUTypeSPS, h264.NALUTypePPS:
+				// add AU in AVC format
+				avc = append(avc, byte(iN>>24), byte(iN>>16), byte(iN>>8), byte(iN))
+				avc = append(avc, annexb[:iN]...)
+
+				// cut search to next AU start
+				annexb = annexb[i0:]
+				i0 = 0
+			}
+		}
+	}
+
+	size := len(annexb)
+	avc = append(avc, byte(size>>24), byte(size>>16), byte(size>>8), byte(size))
+	return append(avc, annexb...)
+}
+
+func IndexFrom(b []byte, sep []byte, from int) int {
+	if from > 0 {
+		if from < len(b) {
+			if i := bytes.Index(b[from:], sep); i >= 0 {
+				return from + i
+			}
+		}
+		return -1
+	}
+
+	return bytes.Index(b, sep)
+}

pkg/mpegts/mpegts_test.go

@@ -0,0 +1,14 @@
+package mpegts
+
+import (
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestTime(t *testing.T) {
+	w := NewWriter()
+	w.WriteTime(0xFFFFFFFF)
+	assert.Equal(t, []byte{0x27, 0xFF, 0xFF, 0xFF, 0xFF}, w.Bytes())
+	ts := ParseTime(w.Bytes())
+	assert.Equal(t, uint32(0xFFFFFFFF), ts)
+}

pkg/mpegts/producer.go

@@ -0,0 +1,26 @@
+package mpegts
+
+import (
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+)
+
+func (c *Client) GetMedias() []*streamer.Media {
+	return c.medias
+}
+
+func (c *Client) GetTrack(media *streamer.Media, codec *streamer.Codec) *streamer.Track {
+	for _, track := range c.tracks {
+		if track.Codec == codec {
+			return track
+		}
+	}
+	return nil
+}
+
+func (c *Client) Start() error {
+	return c.Handle()
+}
+
+func (c *Client) Stop() error {
+	return c.Close()
+}

pkg/mpegts/reader.go

@@ -0,0 +1,201 @@
+package mpegts
+
+import "github.com/pion/rtp"
+
+type Reader struct {
+	b []byte // packets buffer
+	i byte   // read position
+	s byte   // end position
+
+	pmt uint16 // Program Map Table (PMT) PID
+	pes map[uint16]*PES
+}
+
+func NewReader() *Reader {
+	return &Reader{}
+}
+
+func (r *Reader) SetBuffer(b []byte) {
+	r.b = b
+	r.i = 0
+	r.s = PacketSize
+}
+
+func (r *Reader) AppendBuffer(b []byte) {
+	r.b = append(r.b, b...)
+}
+
+func (r *Reader) GetPacket() *rtp.Packet {
+	for r.Sync() {
+		r.Skip(1) // Sync byte
+
+		pid := r.ReadUint16() & 0x1FFF // PID
+		flag := r.ReadByte()           // flags...
+
+		const pidNullPacket = 0x1FFF
+		if pid == pidNullPacket {
+			continue
+		}
+
+		const hasAdaptionField = 0b0010_0000
+		if flag&hasAdaptionField != 0 {
+			adSize := r.ReadByte() // Adaptation field length
+			if adSize > PacketSize-6 {
+				println("WARNING: mpegts: wrong adaptation size")
+				continue
+			}
+			r.Skip(adSize)
+		}
+
+		// PAT: Program Association Table
+		const pidPAT = 0
+		if pid == pidPAT {
+			// already processed
+			if r.pmt != 0 {
+				continue
+			}
+
+			r.ReadPSIHeader()
+
+			const CRCSize = 4
+			for r.Left() > CRCSize {
+				pNum := r.ReadUint16()
+				pPID := r.ReadUint16() & 0x1FFF
+				if pNum != 0 {
+					r.pmt = pPID
+				}
+			}
+
+			r.Skip(4) // CRC32
+			continue
+		}
+
+		// PMT : Program Map Table
+		if pid == r.pmt {
+			// already processed
+			if r.pes != nil {
+				continue
+			}
+
+			r.ReadPSIHeader()
+
+			pesPID := r.ReadUint16() & 0x1FFF // ? PCR PID
+			pSize := r.ReadUint16() & 0x03FF  // ? 0x0FFF
+			r.Skip(byte(pSize))
+
+			r.pes = map[uint16]*PES{}
+
+			const CRCSize = 4
+			for r.Left() > CRCSize {
+				streamType := r.ReadByte()
+				pesPID = r.ReadUint16() & 0x1FFF // Elementary PID
+				iSize := r.ReadUint16() & 0x03FF // ? 0x0FFF
+				r.Skip(byte(iSize))
+
+				r.pes[pesPID] = &PES{StreamType: streamType}
+			}
+
+			r.Skip(4) // ? CRC32
+			continue
+		}
+
+		if r.pes == nil {
+			continue
+		}
+
+		pes := r.pes[pid]
+		if pes == nil {
+			continue // unknown PID
+		}
+
+		if pes.Payload == nil {
+			// PES Packet start code prefix
+			if r.ReadByte() != 0 || r.ReadByte() != 0 || r.ReadByte() != 1 {
+				continue
+			}
+
+			// read stream ID and total payload size
+			pes.StreamID = r.ReadByte()
+			pes.SetBuffer(r.ReadUint16(), r.Bytes())
+		} else {
+			pes.AppendBuffer(r.Bytes())
+		}
+
+		if pkt := pes.GetPacket(); pkt != nil {
+			return pkt
+		}
+	}
+
+	return nil
+}
+
+// Sync - search sync byte
+func (r *Reader) Sync() bool {
+	// drop previous readed packet
+	if r.i != 0 {
+		r.b = r.b[PacketSize:]
+		r.i = 0
+		r.s = PacketSize
+	}
+
+	// if packet available
+	if len(r.b) < PacketSize {
+		return false
+	}
+
+	// if data starts from sync byte
+	if r.b[0] == SyncByte {
+		return true
+	}
+
+	for len(r.b) >= PacketSize {
+		if r.b[0] == SyncByte {
+			return true
+		}
+		r.b = r.b[1:]
+	}
+
+	return false
+}
+
+func (r *Reader) ReadPSIHeader() {
+	pointer := r.ReadByte() // Pointer field
+	r.Skip(pointer)         // Pointer filler bytes
+
+	r.Skip(1)                       // Table ID
+	size := r.ReadUint16() & 0x03FF // Section length
+	r.SetSize(byte(size))
+
+	r.Skip(2) // Table ID extension
+	r.Skip(1) // flags...
+	r.Skip(1) // Section number
+	r.Skip(1) // Last section number
+}
+
+func (r *Reader) Skip(i byte) {
+	r.i += i
+}
+
+func (r *Reader) ReadByte() byte {
+	b := r.b[r.i]
+	r.i++
+	return b
+}
+
+func (r *Reader) ReadUint16() uint16 {
+	i := (uint16(r.b[r.i]) << 8) | uint16(r.b[r.i+1])
+	r.i += 2
+	return i
+}
+
+func (r *Reader) Bytes() []byte {
+	return r.b[r.i:PacketSize]
+}
+
+func (r *Reader) Left() byte {
+	return r.s - r.i
+}
+
+func (r *Reader) SetSize(size byte) {
+	r.s = r.i + size
+}

pkg/mpegts/ts.go

@@ -1,4 +1,4 @@
-package ts
+package mpegts
 
 import (
 	"bytes"

pkg/mpegts/writer.go

@@ -0,0 +1,219 @@
+package mpegts
+
+type Writer struct {
+	b []byte // packets buffer
+	m int    // crc start
+
+	pid        []uint16
+	counter    []byte
+	streamType []byte
+	timestamp  []uint32
+}
+
+func NewWriter() *Writer {
+	return &Writer{}
+}
+
+func (w *Writer) AddPES(pid uint16, streamType byte) {
+	w.pid = append(w.pid, pid)
+	w.streamType = append(w.streamType, streamType)
+	w.counter = append(w.counter, 0)
+	w.timestamp = append(w.timestamp, 0)
+}
+
+func (w *Writer) WriteByte(b byte) {
+	w.b = append(w.b, b)
+}
+
+func (w *Writer) WriteUint16(i uint16) {
+	w.b = append(w.b, byte(i>>8), byte(i))
+}
+
+func (w *Writer) WriteTime(t uint32) {
+	const onlyPTS = 0x20
+	// [>>32 <<3] [>>24 <<2] [>>16 <<2] [>>8 <<1] [<<1]
+	w.b = append(w.b, onlyPTS|byte(t>>29)|1, byte(t>>22), byte(t>>14)|1, byte(t>>7), byte(t<<1)|1)
+}
+
+func (w *Writer) WriteBytes(b []byte) {
+	w.b = append(w.b, b...)
+}
+
+func (w *Writer) MarkChecksum() {
+	w.m = len(w.b)
+}
+
+func (w *Writer) WriteChecksum() {
+	crc := calcCRC32(0xFFFFFFFF, w.b[w.m:])
+	w.b = append(w.b, byte(crc), byte(crc>>8), byte(crc>>16), byte(crc>>24))
+}
+
+func (w *Writer) FinishPacket() {
+	if n := len(w.b) % PacketSize; n != 0 {
+		w.b = append(w.b, make([]byte, PacketSize-n)...)
+	}
+}
+
+func (w *Writer) Bytes() []byte {
+	if len(w.b)%PacketSize != 0 {
+		panic("wrong packet size")
+	}
+	return w.b
+}
+
+func (w *Writer) Reset() {
+	w.b = nil
+}
+
+const isUnitStart = 0x4000
+const flagHasAdaptation = 0x20
+const flagHasPayload = 0x10
+const lenIsProgramTable = 0xB000
+const tableFlags = 0xC1
+const tableHeader = 0xE000
+const tableLength = 0xF000
+
+const patPID = 0
+const patTableID = 0
+const patTableExtID = 1
+
+func (w *Writer) WritePAT() {
+	w.WriteByte(SyncByte)
+	w.WriteUint16(isUnitStart | patPID) // PAT PID
+	w.WriteByte(flagHasPayload)         // flags...
+
+	w.WriteByte(0) // Pointer field
+
+	w.MarkChecksum()
+	w.WriteByte(patTableID)               // Table ID
+	w.WriteUint16(lenIsProgramTable | 13) // Section length
+	w.WriteUint16(patTableExtID)          // Table ID extension
+	w.WriteByte(tableFlags)               // flags...
+	w.WriteByte(0)                        // Section number
+	w.WriteByte(0)                        // Last section number
+
+	w.WriteUint16(1) // Program num (usual 1)
+	w.WriteUint16(tableHeader + pmtPID)
+
+	w.WriteChecksum()
+
+	w.FinishPacket()
+}
+
+const pmtPID = 18
+const pmtTableID = 2
+const pmtTableExtID = 1
+
+func (w *Writer) WritePMT() {
+	w.WriteByte(SyncByte)
+	w.WriteUint16(isUnitStart | pmtPID) // PMT PID
+	w.WriteByte(flagHasPayload)         // flags...
+
+	w.WriteByte(0) // Pointer field
+
+	tableLen := 13 + uint16(len(w.pid))*5
+
+	w.MarkChecksum()
+	w.WriteByte(pmtTableID)                     // Table ID
+	w.WriteUint16(lenIsProgramTable | tableLen) // Section length
+	w.WriteUint16(pmtTableExtID)                // Table ID extension
+	w.WriteByte(tableFlags)                     // flags...
+	w.WriteByte(0)                              // Section number
+	w.WriteByte(0)                              // Last section number
+
+	w.WriteUint16(tableHeader | w.pid[0]) // PID
+	w.WriteUint16(tableLength | 0)        // Info length
+
+	for i, pid := range w.pid {
+		w.WriteByte(w.streamType[i])
+		w.WriteUint16(tableHeader | pid) // PID
+		w.WriteUint16(tableLength | 0)   // Info len
+	}
+
+	w.WriteChecksum()
+
+	w.FinishPacket()
+}
+
+const pesHeaderSize = PacketSize - 18
+
+func (w *Writer) WritePES(pid uint16, streamID byte, payload []byte) {
+	w.WriteByte(SyncByte)
+	w.WriteUint16(isUnitStart | pid)
+
+	// check if payload lower then max first packet size
+	if len(payload) < PacketSize-18 {
+		w.WriteByte(flagHasAdaptation | flagHasPayload)
+
+		// for 183 payload will be zero
+		adSize := PacketSize - 18 - 1 - byte(len(payload))
+		w.WriteByte(adSize)
+		w.WriteBytes(make([]byte, adSize))
+	} else {
+		w.WriteByte(flagHasPayload)
+	}
+
+	w.WriteByte(0)
+	w.WriteByte(0)
+	w.WriteByte(1)
+
+	w.WriteByte(streamID)
+	w.WriteUint16(uint16(8 + len(payload)))
+
+	w.WriteByte(0x80)
+	w.WriteByte(0x80) // only PTS
+	w.WriteByte(5)    // optional size
+
+	switch w.streamType[0] {
+	case StreamTypePCMATapo:
+		w.timestamp[0] += uint32(len(payload) * 45 / 8)
+	}
+
+	w.WriteTime(w.timestamp[0])
+
+	if len(payload) < PacketSize-18 {
+		w.WriteBytes(payload)
+		return
+	}
+
+	w.WriteBytes(payload[:pesHeaderSize])
+
+	payload = payload[pesHeaderSize:]
+	var counter byte
+
+	for {
+		counter++
+
+		if len(payload) > PacketSize-4 {
+			// payload more then maximum size
+			w.WriteByte(SyncByte)
+			w.WriteUint16(pid)
+			w.WriteByte(flagHasPayload | counter&0xF)
+			w.WriteBytes(payload[:PacketSize-4])
+
+			payload = payload[PacketSize-4:]
+		} else if len(payload) == PacketSize-4 {
+			// payload equal maximum size (last packet)
+			w.WriteByte(SyncByte)
+			w.WriteUint16(pid)
+			w.WriteByte(flagHasPayload | counter&0xF)
+			w.WriteBytes(payload)
+
+			break
+		} else {
+			// payload lower than maximum size (last packet)
+			w.WriteByte(SyncByte)
+			w.WriteUint16(pid)
+			w.WriteByte(flagHasAdaptation | flagHasPayload | counter&0xF)
+
+			// for 183 payload will be zero
+			adSize := PacketSize - 4 - 1 - byte(len(payload))
+			w.WriteByte(adSize)
+			w.WriteBytes(make([]byte, adSize))
+
+			w.WriteBytes(payload)
+
+			break
+		}
+	}
+}

pkg/rtsp/client.go

@@ -6,9 +6,6 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
-	"github.com/AlexxIT/go2rtc/pkg/aac"
-	"github.com/AlexxIT/go2rtc/pkg/h264"
-	"github.com/AlexxIT/go2rtc/pkg/mjpeg"
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/AlexxIT/go2rtc/pkg/tcp"
 	"github.com/pion/rtcp"
@@ -38,11 +35,11 @@ const (
 type Mode byte
 
 const (
-	ModeUnknown Mode = iota
-	ModeClientProducer
+	ModeUnknown        Mode = iota
+	ModeClientProducer      // conn act as RTSP client that receive data from RTSP server (ex. camera)
 	ModeServerUnknown
-	ModeServerProducer
-	ModeServerConsumer
+	ModeServerProducer // conn act as RTSP server that reseive data from RTSP client (ex. ffmpeg output)
+	ModeServerConsumer // conn act as RTSP server that send data to RTSP client (ex. ffmpeg input)
 )
 
 type State byte
@@ -111,19 +108,6 @@ func NewClient(uri string) (*Conn, error) {
 	return c, c.parseURI()
 }
 
-func NewServer(conn net.Conn) *Conn {
-	c := new(Conn)
-	c.conn = conn
-	c.mode = ModeServerUnknown
-	c.reader = bufio.NewReader(conn)
-	return c
-}
-
-func (c *Conn) Auth(username, password string) {
-	info := url.UserPassword(username, password)
-	c.auth = tcp.NewAuth(info)
-}
-
 func (c *Conn) parseURI() (err error) {
 	c.URL, err = url.Parse(c.uri)
 	if err != nil {
@@ -220,12 +204,15 @@ func (c *Conn) Do(req *tcp.Request) (*tcp.Response, error) {
 	if res.StatusCode == http.StatusUnauthorized {
 		switch c.auth.Method {
 		case tcp.AuthNone:
+			if c.auth.ReadNone(res) {
+				return c.Do(req)
+			}
 			return nil, errors.New("user/pass not provided")
 		case tcp.AuthUnknown:
 			if c.auth.Read(res) {
 				return c.Do(req)
 			}
-		case tcp.AuthBasic, tcp.AuthDigest:
+		default:
 			return nil, errors.New("wrong user/pass")
 		}
 	}
@@ -332,24 +319,26 @@ func (c *Conn) Describe() error {
 	return nil
 }
 
-//func (c *Conn) Announce() (err error) {
-//	req := &tcp.Request{
-//		Method: MethodAnnounce,
-//		URL:    c.URL,
-//		Header: map[string][]string{
-//			"Content-Type": {"application/sdp"},
-//		},
-//	}
-//
-//	//req.Body, err = c.sdp.Marshal()
-//	if err != nil {
-//		return
-//	}
-//
-//	_, err = c.Do(req)
-//
-//	return
-//}
+func (c *Conn) Announce() (err error) {
+	req := &tcp.Request{
+		Method: MethodAnnounce,
+		URL:    c.URL,
+		Header: map[string][]string{
+			"Content-Type": {"application/sdp"},
+		},
+	}
+
+	req.Body, err = streamer.MarshalSDP(c.SessionName, c.Medias)
+	if err != nil {
+		return err
+	}
+
+	res, err := c.Do(req)
+
+	_ = res
+
+	return
+}
 
 func (c *Conn) Setup() error {
 	for _, media := range c.Medias {
@@ -528,151 +517,6 @@ func (c *Conn) Close() error {
 	return c.conn.Close()
 }
 
-const transport = "RTP/AVP/TCP;unicast;interleaved="
-
-func (c *Conn) Accept() error {
-	for {
-		req, err := tcp.ReadRequest(c.reader)
-		if err != nil {
-			return err
-		}
-
-		if c.URL == nil {
-			c.URL = req.URL
-			c.UserAgent = req.Header.Get("User-Agent")
-		}
-
-		c.Fire(req)
-
-		if !c.auth.Validate(req) {
-			res := &tcp.Response{
-				Status: "401 Unauthorized",
-				Header: map[string][]string{"Www-Authenticate": {`Basic realm="go2rtc"`}},
-			}
-			if err = c.Response(res); err != nil {
-				return err
-			}
-			continue
-		}
-
-		// Receiver: OPTIONS > DESCRIBE > SETUP... > PLAY > TEARDOWN
-		// Sender: OPTIONS > ANNOUNCE > SETUP... > RECORD > TEARDOWN
-		switch req.Method {
-		case MethodOptions:
-			res := &tcp.Response{
-				Header: map[string][]string{
-					"Public": {"OPTIONS, SETUP, TEARDOWN, DESCRIBE, PLAY, PAUSE, ANNOUNCE, RECORD"},
-				},
-				Request: req,
-			}
-			if err = c.Response(res); err != nil {
-				return err
-			}
-
-		case MethodAnnounce:
-			if req.Header.Get("Content-Type") != "application/sdp" {
-				return errors.New("wrong content type")
-			}
-
-			c.Medias, err = UnmarshalSDP(req.Body)
-			if err != nil {
-				return err
-			}
-
-			// TODO: fix someday...
-			c.channels = map[byte]*streamer.Track{}
-			for i, media := range c.Medias {
-				track := streamer.NewTrack(media.Codecs[0], media.Direction)
-				c.tracks = append(c.tracks, track)
-				c.channels[byte(i<<1)] = track
-			}
-
-			c.mode = ModeServerProducer
-			c.Fire(MethodAnnounce)
-
-			res := &tcp.Response{Request: req}
-			if err = c.Response(res); err != nil {
-				return err
-			}
-
-		case MethodDescribe:
-			c.mode = ModeServerConsumer
-			c.Fire(MethodDescribe)
-
-			if c.tracks == nil {
-				res := &tcp.Response{
-					Status:  "404 Not Found",
-					Request: req,
-				}
-				return c.Response(res)
-			}
-
-			res := &tcp.Response{
-				Header: map[string][]string{
-					"Content-Type": {"application/sdp"},
-				},
-				Request: req,
-			}
-
-			// convert tracks to real output medias medias
-			var medias []*streamer.Media
-			for _, track := range c.tracks {
-				media := &streamer.Media{
-					Kind:      streamer.GetKind(track.Codec.Name),
-					Direction: streamer.DirectionSendonly,
-					Codecs:    []*streamer.Codec{track.Codec},
-				}
-				medias = append(medias, media)
-			}
-
-			res.Body, err = streamer.MarshalSDP(c.SessionName, medias)
-			if err != nil {
-				return err
-			}
-
-			if err = c.Response(res); err != nil {
-				return err
-			}
-
-		case MethodSetup:
-			tr := req.Header.Get("Transport")
-
-			res := &tcp.Response{
-				Header:  map[string][]string{},
-				Request: req,
-			}
-
-			if strings.HasPrefix(tr, transport) {
-				c.Session = "1" // TODO: fixme
-				c.state = StateSetup
-				res.Header.Set("Transport", tr[:len(transport)+3])
-			} else {
-				res.Status = "461 Unsupported transport"
-			}
-
-			if err = c.Response(res); err != nil {
-				return err
-			}
-
-		case MethodRecord, MethodPlay:
-			res := &tcp.Response{Request: req}
-			if err = c.Response(res); err == nil {
-				c.state = StatePlay
-			}
-			return err
-
-		case MethodTeardown:
-			res := &tcp.Response{Request: req}
-			_ = c.Response(res)
-			c.state = StateNone
-			return c.conn.Close()
-
-		default:
-			return fmt.Errorf("unsupported method: %s", req.Method)
-		}
-	}
-}
-
 func (c *Conn) Handle() (err error) {
 	c.stateMu.Lock()
 
@@ -715,9 +559,16 @@ func (c *Conn) Handle() (err error) {
 	switch c.mode {
 	case ModeClientProducer:
 		// polling frames from remote RTSP Server (ex Camera)
-		timeout = time.Second * 5
 		go c.keepalive()
 
+		if c.HasSendTracks() {
+			// if we receiving video/audio from camera
+			timeout = time.Second * 5
+		} else {
+			// if we only send audio to camera
+			timeout = time.Second * 30
+		}
+
 	case ModeServerProducer:
 		// polling frames from remote RTSP Client (ex FFmpeg)
 		timeout = time.Second * 15
@@ -878,50 +729,11 @@ func (c *Conn) GetChannel(media *streamer.Media) int {
 	return -1
 }
 
-func (c *Conn) bindTrack(
-	track *streamer.Track, channel uint8, payloadType uint8,
-) *streamer.Track {
-	push := func(packet *rtp.Packet) error {
-		if c.state == StateNone {
-			return nil
-		}
-		packet.Header.PayloadType = payloadType
-
-		size := packet.MarshalSize()
-
-		//log.Printf("[RTP] codec: %s, size: %6d, ts: %10d, pt: %2d, ssrc: %d, seq: %d, mark: %v", track.Codec.Name, len(packet.Payload), packet.Timestamp, packet.PayloadType, packet.SSRC, packet.SequenceNumber, packet.Marker)
-
-		data := make([]byte, 4+size)
-		data[0] = '$'
-		data[1] = channel
-		binary.BigEndian.PutUint16(data[2:], uint16(size))
-
-		if _, err := packet.MarshalTo(data[4:]); err != nil {
-			return nil
-		}
-
-		if _, err := c.conn.Write(data); err != nil {
-			return err
-		}
-
-		c.send += size
-
-		return nil
-	}
-
-	if !track.Codec.IsRTP() {
-		switch track.Codec.Name {
-		case streamer.CodecH264:
-			wrapper := h264.RTPPay(1500)
-			push = wrapper(push)
-		case streamer.CodecAAC:
-			wrapper := aac.RTPPay(1500)
-			push = wrapper(push)
-		case streamer.CodecJPEG:
-			wrapper := mjpeg.RTPPay()
-			push = wrapper(push)
+func (c *Conn) HasSendTracks() bool {
+	for _, track := range c.tracks {
+		if track.Direction == streamer.DirectionSendonly {
+			return true
 		}
 	}
-
-	return track.Bind(push)
+	return false
 }

pkg/rtsp/consumer.go

@@ -0,0 +1,112 @@
+package rtsp
+
+import (
+	"encoding/binary"
+	"github.com/AlexxIT/go2rtc/pkg/aac"
+	"github.com/AlexxIT/go2rtc/pkg/h264"
+	"github.com/AlexxIT/go2rtc/pkg/h265"
+	"github.com/AlexxIT/go2rtc/pkg/mjpeg"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/pion/rtp"
+)
+
+func (c *Conn) AddTrack(media *streamer.Media, track *streamer.Track) *streamer.Track {
+	switch c.mode {
+	// send our track to RTSP consumer (ex. FFmpeg)
+	case ModeServerConsumer:
+		i := len(c.tracks)
+		channelID := byte(i << 1)
+
+		codec := track.Codec.Clone()
+		codec.PayloadType = uint8(96 + i)
+
+		if media.MatchAll() {
+			// fill consumer medias list
+			c.Medias = append(c.Medias, &streamer.Media{
+				Kind: media.Kind, Direction: media.Direction,
+				Codecs: []*streamer.Codec{codec},
+			})
+		} else {
+			// find consumer media and replace codec with right one
+			for i, m := range c.Medias {
+				if m == media {
+					media.Codecs = []*streamer.Codec{codec}
+					c.Medias[i] = media
+					break
+				}
+			}
+		}
+
+		track = c.bindTrack(track, channelID, codec.PayloadType)
+		track.Codec = codec
+		c.tracks = append(c.tracks, track)
+
+		return track
+
+	// camera with backchannel support
+	case ModeClientProducer:
+		consCodec := media.MatchCodec(track.Codec)
+		consTrack := c.GetTrack(media, consCodec)
+		if consTrack == nil {
+			return nil
+		}
+
+		return track.Bind(func(packet *rtp.Packet) error {
+			return consTrack.WriteRTP(packet)
+		})
+	}
+
+	println("WARNING: rtsp: AddTrack to wrong mode")
+	return nil
+}
+
+func (c *Conn) bindTrack(
+	track *streamer.Track, channel uint8, payloadType uint8,
+) *streamer.Track {
+	push := func(packet *rtp.Packet) error {
+		if c.state == StateNone {
+			return nil
+		}
+		packet.Header.PayloadType = payloadType
+
+		size := packet.MarshalSize()
+
+		//log.Printf("[RTP] codec: %s, size: %6d, ts: %10d, pt: %2d, ssrc: %d, seq: %d, mark: %v", track.Codec.Name, len(packet.Payload), packet.Timestamp, packet.PayloadType, packet.SSRC, packet.SequenceNumber, packet.Marker)
+
+		data := make([]byte, 4+size)
+		data[0] = '$'
+		data[1] = channel
+		binary.BigEndian.PutUint16(data[2:], uint16(size))
+
+		if _, err := packet.MarshalTo(data[4:]); err != nil {
+			return nil
+		}
+
+		if _, err := c.conn.Write(data); err != nil {
+			return err
+		}
+
+		c.send += size
+
+		return nil
+	}
+
+	if !track.Codec.IsRTP() {
+		switch track.Codec.Name {
+		case streamer.CodecH264:
+			wrapper := h264.RTPPay(1500)
+			push = wrapper(push)
+		case streamer.CodecH265:
+			wrapper := h265.RTPPay(1500)
+			push = wrapper(push)
+		case streamer.CodecAAC:
+			wrapper := aac.RTPPay(1500)
+			push = wrapper(push)
+		case streamer.CodecJPEG:
+			wrapper := mjpeg.RTPPay()
+			push = wrapper(push)
+		}
+	}
+
+	return track.Bind(push)
+}

pkg/rtsp/helpers.go

@@ -29,6 +29,11 @@ func UnmarshalSDP(rawSDP []byte) ([]*streamer.Media, error) {
 		rawSDP[i+10] = '\n'
 	}
 
+	// fix bug from Ezviz C6N
+	if i := bytes.Index(rawSDP, []byte("H265/90000\r\na=fmtp:96 profile-level-id=420029;")); i > 0 {
+		rawSDP[i+3] = '4'
+	}
+
 	sd := &sdp.SessionDescription{}
 	if err := sd.Unmarshal(rawSDP); err != nil {
 		// fix multiple `s=` https://github.com/AlexxIT/WebRTC/issues/417

pkg/rtsp/producer.go

@@ -6,8 +6,6 @@ import (
 	"github.com/AlexxIT/go2rtc/pkg/streamer"
 )
 
-// Element Producer
-
 func (c *Conn) GetMedias() []*streamer.Media {
 	if c.Medias != nil {
 		return c.Medias
@@ -70,50 +68,6 @@ func (c *Conn) Stop() error {
 	return c.Close()
 }
 
-// Consumer
-
-func (c *Conn) AddTrack(media *streamer.Media, track *streamer.Track) *streamer.Track {
-	switch track.Direction {
-	// send our track to RTSP consumer (ex. FFmpeg)
-	case streamer.DirectionSendonly:
-		i := len(c.tracks)
-		channelID := byte(i << 1)
-
-		codec := track.Codec.Clone()
-		codec.PayloadType = uint8(96 + i)
-
-		if media.MatchAll() {
-			// fill consumer medias list
-			c.Medias = append(c.Medias, &streamer.Media{
-				Kind: media.Kind, Direction: media.Direction,
-				Codecs: []*streamer.Codec{codec},
-			})
-		} else {
-			// find consumer media and replace codec with right one
-			for i, m := range c.Medias {
-				if m == media {
-					media.Codecs = []*streamer.Codec{codec}
-					c.Medias[i] = media
-					break
-				}
-			}
-		}
-
-		track = c.bindTrack(track, channelID, codec.PayloadType)
-		track.Codec = codec
-		c.tracks = append(c.tracks, track)
-
-		return track
-
-	case streamer.DirectionRecvonly:
-		panic("not implemented")
-	}
-
-	panic("wrong direction")
-}
-
-//
-
 func (c *Conn) MarshalJSON() ([]byte, error) {
 	info := &streamer.Info{
 		UserAgent: c.UserAgent,

pkg/rtsp/rtsp_test.go

@@ -1,7 +1,10 @@
 package rtsp
 
 import (
+	"github.com/AlexxIT/go2rtc/pkg/h264"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
 	"github.com/stretchr/testify/assert"
+	"strings"
 	"testing"
 )
 
@@ -106,3 +109,33 @@ a=control:track3
 	assert.Nil(t, err)
 	assert.Len(t, medias, 3)
 }
+
+func TestBugSDP4(t *testing.T) {
+	s := `v=0
+o=- 1676583297494652 1676583297494652 IN IP4 192.168.1.58
+s=Media Presentation
+e=NONE
+b=AS:5050
+t=0 0
+a=control:rtsp://192.168.1.58:554/h264_stream/
+m=video 0 RTP/AVP 96
+b=AS:5000
+a=control:rtsp://192.168.1.58:554/h264_stream/trackID=1
+a=rtpmap:96 H265/90000
+a=fmtp:96 profile-level-id=420029; packetization-mode=1; sprop-parameter-sets=
+a=Media_header:MEDIAINFO=494D4B48010100000400050000000000000000000000000000000000000000000000000000000000;
+a=appversion:1.0
+`
+	s = strings.ReplaceAll(s, "\n", "\r\n")
+	medias, err := UnmarshalSDP([]byte(s))
+	assert.Nil(t, err)
+
+	codec := medias[0].Codecs[0]
+	assert.Equal(t, streamer.CodecH264, codec.Name)
+
+	sps, _ := h264.GetParameterSet(codec.FmtpLine)
+	assert.Nil(t, sps)
+
+	profile := h264.GetProfileLevelID(codec.FmtpLine)
+	assert.Equal(t, "420029", profile)
+}

pkg/rtsp/server.go

@@ -0,0 +1,170 @@
+package rtsp
+
+import (
+	"bufio"
+	"errors"
+	"fmt"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/AlexxIT/go2rtc/pkg/tcp"
+	"net"
+	"net/url"
+	"strings"
+)
+
+func NewServer(conn net.Conn) *Conn {
+	c := new(Conn)
+	c.conn = conn
+	c.mode = ModeServerUnknown
+	c.reader = bufio.NewReader(conn)
+	return c
+}
+
+func (c *Conn) Auth(username, password string) {
+	info := url.UserPassword(username, password)
+	c.auth = tcp.NewAuth(info)
+}
+
+const transport = "RTP/AVP/TCP;unicast;interleaved="
+
+func (c *Conn) Accept() error {
+	for {
+		req, err := tcp.ReadRequest(c.reader)
+		if err != nil {
+			return err
+		}
+
+		if c.URL == nil {
+			c.URL = req.URL
+			c.UserAgent = req.Header.Get("User-Agent")
+		}
+
+		c.Fire(req)
+
+		if !c.auth.Validate(req) {
+			res := &tcp.Response{
+				Status: "401 Unauthorized",
+				Header: map[string][]string{"Www-Authenticate": {`Basic realm="go2rtc"`}},
+			}
+			if err = c.Response(res); err != nil {
+				return err
+			}
+			continue
+		}
+
+		// Receiver: OPTIONS > DESCRIBE > SETUP... > PLAY > TEARDOWN
+		// Sender: OPTIONS > ANNOUNCE > SETUP... > RECORD > TEARDOWN
+		switch req.Method {
+		case MethodOptions:
+			res := &tcp.Response{
+				Header: map[string][]string{
+					"Public": {"OPTIONS, SETUP, TEARDOWN, DESCRIBE, PLAY, PAUSE, ANNOUNCE, RECORD"},
+				},
+				Request: req,
+			}
+			if err = c.Response(res); err != nil {
+				return err
+			}
+
+		case MethodAnnounce:
+			if req.Header.Get("Content-Type") != "application/sdp" {
+				return errors.New("wrong content type")
+			}
+
+			c.Medias, err = UnmarshalSDP(req.Body)
+			if err != nil {
+				return err
+			}
+
+			// TODO: fix someday...
+			c.channels = map[byte]*streamer.Track{}
+			for i, media := range c.Medias {
+				track := streamer.NewTrack(media.Codecs[0], media.Direction)
+				c.tracks = append(c.tracks, track)
+				c.channels[byte(i<<1)] = track
+			}
+
+			c.mode = ModeServerProducer
+			c.Fire(MethodAnnounce)
+
+			res := &tcp.Response{Request: req}
+			if err = c.Response(res); err != nil {
+				return err
+			}
+
+		case MethodDescribe:
+			c.mode = ModeServerConsumer
+			c.Fire(MethodDescribe)
+
+			if c.tracks == nil {
+				res := &tcp.Response{
+					Status:  "404 Not Found",
+					Request: req,
+				}
+				return c.Response(res)
+			}
+
+			res := &tcp.Response{
+				Header: map[string][]string{
+					"Content-Type": {"application/sdp"},
+				},
+				Request: req,
+			}
+
+			// convert tracks to real output medias medias
+			var medias []*streamer.Media
+			for _, track := range c.tracks {
+				media := &streamer.Media{
+					Kind:      streamer.GetKind(track.Codec.Name),
+					Direction: streamer.DirectionSendonly,
+					Codecs:    []*streamer.Codec{track.Codec},
+				}
+				medias = append(medias, media)
+			}
+
+			res.Body, err = streamer.MarshalSDP(c.SessionName, medias)
+			if err != nil {
+				return err
+			}
+
+			if err = c.Response(res); err != nil {
+				return err
+			}
+
+		case MethodSetup:
+			tr := req.Header.Get("Transport")
+
+			res := &tcp.Response{
+				Header:  map[string][]string{},
+				Request: req,
+			}
+
+			if strings.HasPrefix(tr, transport) {
+				c.Session = "1" // TODO: fixme
+				c.state = StateSetup
+				res.Header.Set("Transport", tr[:len(transport)+3])
+			} else {
+				res.Status = "461 Unsupported transport"
+			}
+
+			if err = c.Response(res); err != nil {
+				return err
+			}
+
+		case MethodRecord, MethodPlay:
+			res := &tcp.Response{Request: req}
+			if err = c.Response(res); err == nil {
+				c.state = StatePlay
+			}
+			return err
+
+		case MethodTeardown:
+			res := &tcp.Response{Request: req}
+			_ = c.Response(res)
+			c.state = StateNone
+			return c.conn.Close()
+
+		default:
+			return fmt.Errorf("unsupported method: %s", req.Method)
+		}
+	}
+}

pkg/streamer/helpers.go

@@ -2,6 +2,7 @@ package streamer
 
 import (
 	"strings"
+	"time"
 )
 
 type Info struct {
@@ -50,3 +51,31 @@ func Contains(medias []*Media, media *Media, codec *Codec) bool {
 	}
 	return ok1 && ok2
 }
+
+type Probe struct {
+	deadline time.Time
+	items    map[interface{}]struct{}
+}
+
+func NewProbe(enable bool) *Probe {
+	if enable {
+		return &Probe{
+			deadline: time.Now().Add(time.Second * 3),
+			items:    map[interface{}]struct{}{},
+		}
+	} else {
+		return nil
+	}
+}
+
+// Active return true if probe enabled and not finish
+func (p *Probe) Active() bool {
+	return len(p.items) < 2 && time.Now().Before(p.deadline)
+}
+
+// Append safe to run if Probe is nil
+func (p *Probe) Append(v interface{}) {
+	if p != nil {
+		p.items[v] = struct{}{}
+	}
+}

pkg/streamer/track.go

@@ -21,6 +21,13 @@ func NewTrack(codec *Codec, direction string) *Track {
 	return &Track{Codec: codec, Direction: direction, sinkMu: new(sync.RWMutex)}
 }
 
+func NewTrack2(media *Media, codec *Codec) *Track {
+	if codec == nil {
+		codec = media.Codecs[0]
+	}
+	return &Track{Codec: codec, Direction: media.Direction, sinkMu: new(sync.RWMutex)}
+}
+
 func (t *Track) String() string {
 	s := t.Codec.String()
 	if t.sinkMu.TryRLock() {

pkg/tapo/backchannel.go

@@ -0,0 +1,53 @@
+package tapo
+
+import (
+	"bytes"
+	"github.com/AlexxIT/go2rtc/pkg/mpegts"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/pion/rtp"
+	"strconv"
+)
+
+func (c *Client) backchannelWriter() streamer.WriterFunc {
+	w := mpegts.NewWriter()
+	w.AddPES(68, mpegts.StreamTypePCMATapo)
+	w.WritePAT()
+	w.WritePMT()
+
+	return func(packet *rtp.Packet) (err error) {
+		// don't know why 68 and 192
+		w.WritePES(68, 192, packet.Payload)
+		err = c.WriteBackchannel(w.Bytes())
+		w.Reset()
+		return
+	}
+}
+
+func (c *Client) SetupBackchannel() (err error) {
+	// if conn1 is not used - we will use it for backchannel
+	// or we need to start another conn for session2
+	if c.session1 != "" {
+		if c.conn2, err = c.newConn(); err != nil {
+			return
+		}
+	} else {
+		c.conn2 = c.conn1
+	}
+
+	c.session2, err = c.Request(c.conn2, []byte(`{"params":{"talk":{"mode":"aec"},"method":"get"},"seq":3,"type":"request"}`))
+	return
+}
+
+func (c *Client) WriteBackchannel(body []byte) (err error) {
+	// TODO: fixme (size)
+	buf := bytes.NewBuffer(nil)
+	buf.WriteString("----client-stream-boundary--\r\n")
+	buf.WriteString("Content-Type: audio/mp2t\r\n")
+	buf.WriteString("X-If-Encrypt: 0\r\n")
+	buf.WriteString("X-Session-Id: " + c.session2 + "\r\n")
+	buf.WriteString("Content-Length: " + strconv.Itoa(len(body)) + "\r\n\r\n")
+	buf.Write(body)
+
+	_, err = buf.WriteTo(c.conn2)
+	return
+}

pkg/tapo/client.go

@@ -0,0 +1,229 @@
+package tapo
+
+import (
+	"bytes"
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/md5"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"github.com/AlexxIT/go2rtc/pkg/mpegts"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/AlexxIT/go2rtc/pkg/tcp"
+	"mime/multipart"
+	"net"
+	"net/http"
+	"net/url"
+	"strconv"
+)
+
+type Client struct {
+	streamer.Element
+
+	url string
+
+	medias []*streamer.Media
+	tracks map[byte]*streamer.Track
+
+	conn1 net.Conn
+	conn2 net.Conn
+
+	decrypt func(b []byte) []byte
+
+	session1 string
+	session2 string
+}
+
+// block ciphers using cipher block chaining.
+type cbcMode interface {
+	cipher.BlockMode
+	SetIV([]byte)
+}
+
+func NewClient(url string) *Client {
+	return &Client{url: url}
+}
+
+func (c *Client) Dial() (err error) {
+	c.conn1, err = c.newConn()
+	return
+}
+
+func (c *Client) newConn() (net.Conn, error) {
+	u, err := url.Parse(c.url)
+	if err != nil {
+		return nil, err
+	}
+
+	// support raw username/password
+	username := u.User.Username()
+	password, _ := u.User.Password()
+
+	// or cloud password in place of username
+	if password == "" {
+		password = fmt.Sprintf("%16X", md5.Sum([]byte(username)))
+		username = "admin"
+		u.User = url.UserPassword(username, password)
+	}
+
+	u.Scheme = "http"
+	u.Path = "/stream"
+	if u.Port() == "" {
+		u.Host += ":8800"
+	}
+
+	req, err := http.NewRequest("POST", u.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+
+	req.Header.Set("Content-Type", "multipart/mixed; boundary=--client-stream-boundary--")
+
+	res, err := tcp.Do(req)
+	if err != nil {
+		return nil, err
+	}
+
+	if res.StatusCode != http.StatusOK {
+		return nil, errors.New(res.Status)
+	}
+
+	if c.decrypt == nil {
+		c.newDectypter(res, username, password)
+	}
+
+	return res.Body.(net.Conn), nil
+}
+
+func (c *Client) newDectypter(res *http.Response, username, password string) {
+	// extract nonce from response
+	// cipher="AES_128_CBC" username="admin" padding="PKCS7_16" algorithm="MD5" nonce="***"
+	nonce := res.Header.Get("Key-Exchange")
+	nonce = streamer.Between(nonce, `nonce="`, `"`)
+
+	key := md5.Sum([]byte(nonce + ":" + password))
+	iv := md5.Sum([]byte(username + ":" + nonce))
+
+	block, err := aes.NewCipher(key[:])
+	if err != nil {
+		return
+	}
+
+	cbc := cipher.NewCBCDecrypter(block, iv[:]).(cbcMode)
+
+	c.decrypt = func(b []byte) []byte {
+		// restore IV
+		cbc.SetIV(iv[:])
+
+		// decrypt
+		cbc.CryptBlocks(b, b)
+
+		// unpad
+		padSize := int(b[len(b)-1])
+		return b[:len(b)-padSize]
+	}
+}
+
+func (c *Client) SetupStream() (err error) {
+	if c.session1 != "" {
+		return
+	}
+
+	// audio: default, disable, enable
+	c.session1, err = c.Request(c.conn1, []byte(`{"params":{"preview":{"audio":["default"],"channels":[0],"resolutions":["HD"]},"method":"get"},"seq":1,"type":"request"}`))
+	return
+}
+
+// Handle - first run will be in probe state
+func (c *Client) Handle() error {
+	mpReader := multipart.NewReader(c.conn1, "--device-stream-boundary--")
+	tsReader := mpegts.NewReader()
+
+	for {
+		p, err := mpReader.NextRawPart()
+		if err != nil {
+			return err
+		}
+
+		if ct := p.Header.Get("Content-Type"); ct != "video/mp2t" {
+			continue
+		}
+
+		cl := p.Header.Get("Content-Length")
+		size, err := strconv.Atoi(cl)
+		if err != nil {
+			return err
+		}
+
+		body := make([]byte, size)
+
+		b := body
+		for {
+			if n, err2 := p.Read(b); err2 == nil {
+				b = b[n:]
+			} else {
+				break
+			}
+		}
+
+		body = c.decrypt(body)
+		tsReader.SetBuffer(body)
+
+		for {
+			pkt := tsReader.GetPacket()
+			if pkt == nil {
+				break
+			}
+
+			if track := c.tracks[pkt.PayloadType]; track != nil {
+				_ = track.WriteRTP(pkt)
+			}
+		}
+	}
+}
+
+func (c *Client) Close() (err error) {
+	if c.conn1 != nil {
+		err = c.conn1.Close()
+	}
+	if c.conn2 != nil {
+		_ = c.conn2.Close()
+	}
+	return
+}
+
+func (c *Client) Request(conn net.Conn, body []byte) (string, error) {
+	// TODO: fixme (size)
+	buf := bytes.NewBuffer(nil)
+	buf.WriteString("----client-stream-boundary--\r\n")
+	buf.WriteString("Content-Type: application/json\r\n")
+	buf.WriteString("Content-Length: " + strconv.Itoa(len(body)) + "\r\n\r\n")
+	buf.Write(body)
+	buf.WriteString("\r\n")
+
+	if _, err := buf.WriteTo(conn); err != nil {
+		return "", err
+	}
+
+	mpReader := multipart.NewReader(conn, "--device-stream-boundary--")
+
+	for {
+		p, err := mpReader.NextRawPart()
+		if err != nil {
+			return "", err
+		}
+
+		var v struct {
+			Params struct {
+				SessionID string `json:"session_id"`
+			} `json:"params"`
+		}
+
+		if err = json.NewDecoder(p).Decode(&v); err != nil {
+			return "", err
+		}
+
+		return v.Params.SessionID, nil
+	}
+}

pkg/tapo/consumer.go

@@ -0,0 +1,18 @@
+package tapo
+
+import (
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/pion/rtp"
+)
+
+func (c *Client) AddTrack(media *streamer.Media, track *streamer.Track) *streamer.Track {
+	consCodec := media.MatchCodec(track.Codec)
+	consTrack := c.GetTrack(media, consCodec)
+	if consTrack == nil {
+		return nil
+	}
+
+	return track.Bind(func(packet *rtp.Packet) error {
+		return consTrack.WriteRTP(packet)
+	})
+}

pkg/tapo/producer.go

@@ -0,0 +1,88 @@
+package tapo
+
+import (
+	"github.com/AlexxIT/go2rtc/pkg/mpegts"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+)
+
+func (c *Client) GetMedias() []*streamer.Media {
+	// producer should have persistent medias
+	if c.medias == nil {
+		// don't know if all Tapo has this capabilities...
+		c.medias = []*streamer.Media{
+			{
+				Kind:      streamer.KindVideo,
+				Direction: streamer.DirectionSendonly,
+				Codecs: []*streamer.Codec{
+					{Name: streamer.CodecH264, ClockRate: 90000, PayloadType: streamer.PayloadTypeRAW},
+				},
+			},
+			{
+				Kind:      streamer.KindAudio,
+				Direction: streamer.DirectionSendonly,
+				Codecs: []*streamer.Codec{
+					{Name: streamer.CodecPCMA, ClockRate: 8000, PayloadType: 8},
+				},
+			},
+			{
+				Kind:      streamer.KindAudio,
+				Direction: streamer.DirectionRecvonly,
+				Codecs: []*streamer.Codec{
+					{Name: streamer.CodecPCMA, ClockRate: 8000, PayloadType: 8},
+				},
+			},
+		}
+	}
+
+	return c.medias
+}
+
+func (c *Client) GetTrack(media *streamer.Media, codec *streamer.Codec) (track *streamer.Track) {
+	for _, track := range c.tracks {
+		if track.Codec == codec {
+			return track
+		}
+	}
+
+	if c.tracks == nil {
+		c.tracks = map[byte]*streamer.Track{}
+	}
+
+	if media.Kind == streamer.KindVideo {
+		if err := c.SetupStream(); err != nil {
+			return nil
+		}
+
+		track = streamer.NewTrack2(media, codec)
+		c.tracks[mpegts.StreamTypeH264] = track
+	} else {
+		if media.Direction == streamer.DirectionSendonly {
+			if err := c.SetupStream(); err != nil {
+				return nil
+			}
+
+			track = streamer.NewTrack2(media, codec)
+			c.tracks[mpegts.StreamTypePCMATapo] = track
+		} else {
+			if err := c.SetupBackchannel(); err != nil {
+				return nil
+			}
+
+			if w := c.backchannelWriter(); w != nil {
+				track = streamer.NewTrack2(media, codec)
+				track.Bind(w)
+				c.tracks[0] = track
+			}
+		}
+	}
+
+	return
+}
+
+func (c *Client) Start() error {
+	return c.Handle()
+}
+
+func (c *Client) Stop() error {
+	return c.Close()
+}

pkg/tcp/auth.go

@@ -22,6 +22,7 @@ const (
 	AuthUnknown
 	AuthBasic
 	AuthDigest
+	AuthTPLink // https://drmnsamoliu.github.io/video.html
 )
 
 func NewAuth(user *url.Userinfo) *Auth {
@@ -79,6 +80,8 @@ func (a *Auth) Write(req *Request) {
 			`, uri="%s", response="%s"`, uri, response,
 		)
 		req.Header.Set("Authorization", header)
+	case AuthTPLink:
+		req.URL.Host = "127.0.0.1"
 	}
 }
 
@@ -100,6 +103,15 @@ func (a *Auth) Validate(req *Request) bool {
 	return header == a.header
 }
 
+func (a *Auth) ReadNone(res *Response) bool {
+	auth := res.Header.Get("WWW-Authenticate")
+	if strings.Contains(auth, "TP-LINK Streaming Media") {
+		a.Method = AuthTPLink
+		return true
+	}
+	return false
+}
+
 func Between(s, sub1, sub2 string) string {
 	i := strings.Index(s, sub1)
 	if i < 0 {

pkg/tcp/request.go

@@ -1,8 +1,10 @@
 package tcp
 
 import (
+	"context"
 	"errors"
 	"fmt"
+	"net"
 	"net/http"
 	"strings"
 	"time"
@@ -10,8 +12,22 @@ import (
 
 // Do - http.Client with support Digest Authorization
 func Do(req *http.Request) (*http.Response, error) {
-	// need to create new client each time to reset timeout
+	var conn net.Conn
+
 	client := http.Client{Timeout: time.Second * 5000}
+
+	// for multipart requests return conn as Body (for write support)
+	if ct := req.Header.Get("Content-Type"); strings.HasPrefix(ct, "multipart/mixed") {
+		var d net.Dialer
+		client.Transport = &http.Transport{
+			DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+				var err error
+				conn, err = d.DialContext(ctx, network, addr)
+				return conn, err
+			},
+		}
+	}
+
 	res, err := client.Do(req)
 	if err != nil {
 		return nil, err
@@ -64,5 +80,9 @@ func Do(req *http.Request) (*http.Response, error) {
 		}
 	}
 
+	if conn != nil {
+		res.Body = conn
+	}
+
 	return res, nil
 }

pkg/webrtc/conn.go

@@ -154,6 +154,10 @@ func (c *Conn) GetCompleteAnswer() (answer string, err error) {
 	return c.Conn.LocalDescription().SDP, nil
 }
 
+func (c *Conn) AddCandidate(candidate string) {
+	_ = c.Conn.AddICECandidate(webrtc.ICECandidateInit{Candidate: candidate})
+}
+
 func (c *Conn) remote() string {
 	if c.Conn == nil {
 		return ""

pkg/webrtc/consumer.go

@@ -106,12 +106,6 @@ func (c *Conn) AddTrack(media *streamer.Media, track *streamer.Track) *streamer.
 	panic("wrong direction")
 }
 
-//
-
-func (c *Conn) AddCandidate(candidate string) {
-	_ = c.Conn.AddICECandidate(webrtc.ICECandidateInit{Candidate: candidate})
-}
-
 func (c *Conn) MarshalJSON() ([]byte, error) {
 	info := &streamer.Info{
 		Type:       "WebRTC client",

www/links.html

@@ -41,10 +41,24 @@
 <body>
 <script src="main.js"></script>
 <div id="links"></div>
+<div id="audio">
+    <h2>Play audio</h2>
+    <input id="source" type="text" placeholder="source">
+    <a id="send" href="#">send</a> / cameras with two way audio support
+</div>
 <script>
     const params = new URLSearchParams(location.search);
     const src = params.get("src");
 
+    document.querySelector("#send")
+        .addEventListener("click", () => {
+            const source = document.querySelector("#source");
+            const url = new URL("api/streams", location.href);
+            url.searchParams.set("dst", src);
+            url.searchParams.set("src", source.value);
+            fetch(url, {method: "POST"});
+        });
+
     const links = document.querySelector("#links");
     links.innerHTML = `
         <h2>Any codec in source</h2>