support publishing with WebRTC (#1659) (#1786)

README.md

@@ -16,6 +16,7 @@ Live streams can be published to the server with:
 |RTMP servers and cameras|RTMP, RTMPS, Enhanced RTMP|H264|MPEG-4 Audio (AAC), MPEG-2 Audio (MP3)|
 |HLS servers and cameras|Low-Latency HLS, MP4-based HLS, legacy HLS|H265, H264|Opus, MPEG-4 Audio (AAC)|
 |UDP/MPEG-TS streams|Unicast, broadcast, multicast|H265, H264|Opus, MPEG-4 Audio (AAC)|
+|WebRTC||AV1, VP9, VP8, H264|Opus, G722, G711|
 |Raspberry Pi Cameras||H264||
 
 And can be read from the server with:
@@ -86,6 +87,7 @@ In the next months, the repository name and the Docker image name will be change
   * [From OBS Studio](#from-obs-studio)
   * [From OpenCV](#from-opencv)
   * [From a UDP stream](#from-a-udp-stream)
+  * [From the browser](#from-the-browser)
 * [Read from the server](#read-from-the-server)
   * [From VLC and Ubuntu](#from-vlc-and-ubuntu)
 * [RTSP protocol](#rtsp-protocol)
@@ -800,6 +802,14 @@ paths:
 
 After starting the server, the stream can be reached on `rtsp://localhost:8554/udp`.
 
+### From the browser
+
+Open the page into the browser:
+
+```
+http://localhost:8889/mystream/publish
+```
+
 ## Read from the server
 
 ### From VLC and Ubuntu

apidocs/openapi.yaml

@@ -303,15 +303,7 @@ components:
         conf:
           $ref: '#/components/schemas/PathConf'
         source:
-          oneOf:
+          $ref: '#/components/schemas/PathSourceOrReader'
-          - $ref: '#/components/schemas/PathSourceRTSPSession'
-          - $ref: '#/components/schemas/PathSourceRTSPSSession'
-          - $ref: '#/components/schemas/PathSourceRTMPConn'
-          - $ref: '#/components/schemas/PathSourceRTMPSConn'
-          - $ref: '#/components/schemas/PathSourceRTSPSource'
-          - $ref: '#/components/schemas/PathSourceRTMPSource'
-          - $ref: '#/components/schemas/PathSourceHLSSource'
-          - $ref: '#/components/schemas/PathSourceRPICameraSource'
         sourceReady:
           type: boolean
         tracks:
@@ -324,127 +316,26 @@ components:
         readers:
           type: array
           items:
-            oneOf:
+            $ref: '#/components/schemas/PathSourceOrReader'
-            - $ref: '#/components/schemas/PathReaderHLSMuxer'
-            - $ref: '#/components/schemas/PathReaderRTMPConn'
-            - $ref: '#/components/schemas/PathReaderRTMPSConn'
-            - $ref: '#/components/schemas/PathReaderRTSPSession'
-            - $ref: '#/components/schemas/PathReaderRTSPSSession'
-            - $ref: '#/components/schemas/PathReaderWebRTCConn'
 
-    PathSourceRTSPSession:
+    PathSourceOrReader:
       type: object
       properties:
         type:
           type: string
-          enum: [rtspSession]
+          enum:
-        id:
+          - hlsMuxer
-          type: string
+          - hlsSource
-
+          - rpiCameraSource
-    PathSourceRTSPSSession:
+          - rtmpSession
-      type: object
+          - rtmpSource
-      properties:
+          - rtmpsSession
-        type:
+          - rtspSession
-          type: string
+          - rtspSource
-          enum: [rtspsSession]
+          - rtspsSession
-        id:
+          - redirect
-          type: string
+          - udpSource
-
+          - webRTCConn
-    PathSourceRTMPConn:
-      type: object
-      properties:
-        type:
-          type: string
-          enum: [rtmpConn]
-        id:
-          type: string
-
-    PathSourceRTMPSConn:
-      type: object
-      properties:
-        type:
-          type: string
-          enum: [rtmpsConn]
-        id:
-          type: string
-
-    PathSourceRTSPSource:
-      type: object
-      properties:
-        type:
-          type: string
-          enum: [rtspSource]
-
-    PathSourceRTMPSource:
-      type: object
-      properties:
-        type:
-          type: string
-          enum: [rtmpSource]
-
-    PathSourceHLSSource:
-      type: object
-      properties:
-        type:
-          type: string
-          enum: [hlsSource]
-
-    PathSourceRPICameraSource:
-      type: object
-      properties:
-        type:
-          type: string
-          enum: [rpiCameraSource]
-
-    PathReaderHLSMuxer:
-      type: object
-      properties:
-        type:
-          type: string
-          enum: [hlsMuxer]
-
-    PathReaderRTMPConn:
-      type: object
-      properties:
-        type:
-          type: string
-          enum: [rtmpConn]
-        id:
-          type: string
-
-    PathReaderRTMPSConn:
-      type: object
-      properties:
-        type:
-          type: string
-          enum: [rtmpsConn]
-        id:
-          type: string
-
-    PathReaderRTSPSession:
-      type: object
-      properties:
-        type:
-          type: string
-          enum: [rtspSession]
-        id:
-          type: string
-
-    PathReaderRTSPSSession:
-      type: object
-      properties:
-        type:
-          type: string
-          enum: [rtspsSession]
-        id:
-          type: string
-
-    PathReaderWebRTCConn:
-      type: object
-      properties:
-        type:
-          type: string
-          enum: [webRTCConn]
         id:
           type: string
 
@@ -560,6 +451,9 @@ components:
           type: string
         remoteCandidate:
           type: string
+        state:
+          type: string
+          enum: [read, publish]
         bytesReceived:
           type: integer
           format: int64

go.mod

@@ -19,7 +19,9 @@ require (
 	github.com/notedit/rtmp v0.0.2
 	github.com/pion/ice/v2 v2.3.2
 	github.com/pion/interceptor v0.1.16
+	github.com/pion/rtcp v1.2.10
 	github.com/pion/rtp v1.7.13
+	github.com/pion/sdp/v3 v3.0.6
 	github.com/pion/webrtc/v3 v3.2.1
 	github.com/stretchr/testify v1.8.2
 	golang.org/x/crypto v0.9.0
@@ -50,9 +52,7 @@ require (
 	github.com/pion/logging v0.2.2 // indirect
 	github.com/pion/mdns v0.0.7 // indirect
 	github.com/pion/randutil v0.1.0 // indirect
-	github.com/pion/rtcp v1.2.10 // indirect
 	github.com/pion/sctp v1.8.7 // indirect
-	github.com/pion/sdp/v3 v3.0.6 // indirect
 	github.com/pion/srtp/v2 v2.0.12 // indirect
 	github.com/pion/stun v0.4.0 // indirect
 	github.com/pion/transport/v2 v2.2.0 // indirect

internal/core/hls_muxer.go

@@ -275,9 +275,7 @@ func (m *hlsMuxer) runInner(innerCtx context.Context, innerReady chan struct{})
 
 	m.path = res.path
 
-	defer func() {
+	defer m.path.readerRemove(pathReaderRemoveReq{author: m})
-		m.path.readerRemove(pathReaderRemoveReq{author: m})
-	}()
 
 	m.ringBuffer, _ = ringbuffer.New(uint64(m.readBufferCount))
 
@@ -614,8 +612,9 @@ func (m *hlsMuxer) apiMuxersList(req hlsServerAPIMuxersListSubReq) {
 }
 
 // apiReaderDescribe implements reader.
-func (m *hlsMuxer) apiReaderDescribe() interface{} {
+func (m *hlsMuxer) apiReaderDescribe() pathAPISourceOrReader {
-	return struct {
+	return pathAPISourceOrReader{
-		Type string `json:"type"`
+		Type: "hlsMuxer",
-	}{"hlsMuxer"}
+		ID:   "",
+	}
 }

internal/core/hls_source.go

@@ -206,8 +206,9 @@ func (s *hlsSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
 }
 
 // apiSourceDescribe implements sourceStaticImpl.
-func (*hlsSource) apiSourceDescribe() interface{} {
+func (*hlsSource) apiSourceDescribe() pathAPISourceOrReader {
-	return struct {
+	return pathAPISourceOrReader{
-		Type string `json:"type"`
+		Type: "hlsSource",
-	}{"hlsSource"}
+		ID:   "",
+	}
 }

internal/core/path.go

@@ -90,6 +90,7 @@ type pathGetPathConfRes struct {
 
 type pathGetPathConfReq struct {
 	name        string
+	publish     bool
 	credentials authCredentials
 	res         chan pathGetPathConfRes
 }
@@ -130,6 +131,7 @@ type pathPublisherAnnounceRes struct {
 type pathPublisherAddReq struct {
 	author      publisher
 	pathName    string
+	skipAuth    bool
 	credentials authCredentials
 	res         chan pathPublisherAnnounceRes
 }
@@ -151,6 +153,11 @@ type pathPublisherStopReq struct {
 	res    chan struct{}
 }
 
+type pathAPISourceOrReader struct {
+	Type string `json:"type"`
+	ID   string `json:"id"`
+}
+
 type pathAPIPathsListItem struct {
 	ConfName      string         `json:"confName"`
 	Conf          *conf.PathConf `json:"conf"`

internal/core/path_manager.go

@@ -209,7 +209,8 @@ outer:
 				continue
 			}
 
-			err = authenticate(pm.externalAuthenticationURL, pm.authMethods, req.name, pathConf, false, req.credentials)
+			err = authenticate(pm.externalAuthenticationURL, pm.authMethods,
+				req.name, pathConf, req.publish, req.credentials)
 			if err != nil {
 				req.res <- pathGetPathConfRes{err: pathErrAuth{wrapped: err}}
 				continue
@@ -266,10 +267,12 @@ outer:
 				continue
 			}
 
-			err = authenticate(pm.externalAuthenticationURL, pm.authMethods, req.pathName, pathConf, true, req.credentials)
+			if !req.skipAuth {
-			if err != nil {
+				err = authenticate(pm.externalAuthenticationURL, pm.authMethods, req.pathName, pathConf, true, req.credentials)
-				req.res <- pathPublisherAnnounceRes{err: pathErrAuth{wrapped: err}}
+				if err != nil {
-				continue
+					req.res <- pathPublisherAnnounceRes{err: pathErrAuth{wrapped: err}}
+					continue
+				}
 			}
 
 			// create path if it doesn't exist

internal/core/reader.go

@@ -3,5 +3,5 @@ package core
 // reader is an entity that can read a stream.
 type reader interface {
 	close()
-	apiReaderDescribe() interface{}
+	apiReaderDescribe() pathAPISourceOrReader
 }

internal/core/rpicamera_source.go

@@ -128,8 +128,9 @@ func (s *rpiCameraSource) run(ctx context.Context, cnf *conf.PathConf, reloadCon
 }
 
 // apiSourceDescribe implements sourceStaticImpl.
-func (*rpiCameraSource) apiSourceDescribe() interface{} {
+func (*rpiCameraSource) apiSourceDescribe() pathAPISourceOrReader {
-	return struct {
+	return pathAPISourceOrReader{
-		Type string `json:"type"`
+		Type: "rpiCameraSource",
-	}{"rpiCameraSource"}
+		ID:   "",
+	}
 }

internal/core/rtmp_conn.go

@@ -377,11 +377,7 @@ func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error {
 		return res.err
 	}
 
-	path := res.path
+	defer res.path.readerRemove(pathReaderRemoveReq{author: c})
-
-	defer func() {
-		path.readerRemove(pathReaderRemoveReq{author: c})
-	}()
 
 	c.stateMutex.Lock()
 	c.state = rtmpConnStateRead
@@ -417,9 +413,9 @@ func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error {
 	defer res.stream.readerRemove(c)
 
 	c.Log(logger.Info, "is reading from path '%s', %s",
-		path.name, sourceMediaInfo(medias))
+		res.path.name, sourceMediaInfo(medias))
 
-	pathConf := path.safeConf()
+	pathConf := res.path.safeConf()
 
 	if pathConf.RunOnRead != "" {
 		c.Log(logger.Info, "runOnRead command started")
@@ -427,7 +423,7 @@ func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error {
 			c.externalCmdPool,
 			pathConf.RunOnRead,
 			pathConf.RunOnReadRestart,
-			path.externalCmdEnv(),
+			res.path.externalCmdEnv(),
 			func(co int) {
 				c.Log(logger.Info, "runOnRead command exited with code %d", co)
 			})
@@ -733,11 +729,7 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error {
 		return res.err
 	}
 
-	path := res.path
+	defer res.path.publisherRemove(pathPublisherRemoveReq{author: c})
-
-	defer func() {
-		path.publisherRemove(pathPublisherRemoveReq{author: c})
-	}()
 
 	c.stateMutex.Lock()
 	c.state = rtmpConnStatePublish
@@ -768,7 +760,7 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error {
 		medias = append(medias, audioMedia)
 	}
 
-	rres := path.publisherStart(pathPublisherStartReq{
+	rres := res.path.publisherStart(pathPublisherStartReq{
 		author:             c,
 		medias:             medias,
 		generateRTPPackets: true,
@@ -778,7 +770,7 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error {
 	}
 
 	c.Log(logger.Info, "is publishing to path '%s', %s",
-		path.name,
+		res.path.name,
 		sourceMediaInfo(medias))
 
 	// disable write deadline to allow outgoing acknowledges
@@ -819,21 +811,19 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error {
 }
 
 // apiReaderDescribe implements reader.
-func (c *rtmpConn) apiReaderDescribe() interface{} {
+func (c *rtmpConn) apiReaderDescribe() pathAPISourceOrReader {
-	return c.apiSourceDescribe()
+	return pathAPISourceOrReader{
+		Type: func() string {
+			if c.isTLS {
+				return "rtmpsConn"
+			}
+			return "rtmpConn"
+		}(),
+		ID: c.uuid.String(),
+	}
 }
 
 // apiSourceDescribe implements source.
-func (c *rtmpConn) apiSourceDescribe() interface{} {
+func (c *rtmpConn) apiSourceDescribe() pathAPISourceOrReader {
-	var typ string
+	return c.apiReaderDescribe()
-	if c.isTLS {
-		typ = "rtmpsConn"
-	} else {
-		typ = "rtmpConn"
-	}
-
-	return struct {
-		Type string `json:"type"`
-		ID   string `json:"id"`
-	}{typ, c.uuid.String()}
 }

internal/core/rtmp_source.go

@@ -148,9 +148,7 @@ func (s *rtmpSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf cha
 
 			s.Log(logger.Info, "ready: %s", sourceMediaInfo(medias))
 
-			defer func() {
+			defer s.parent.sourceStaticImplSetNotReady(pathSourceStaticSetNotReadyReq{})
-				s.parent.sourceStaticImplSetNotReady(pathSourceStaticSetNotReadyReq{})
-			}()
 
 			videoWriteFunc := getRTMPWriteFunc(videoMedia, videoFormat, res.stream)
 			audioWriteFunc := getRTMPWriteFunc(audioMedia, audioFormat, res.stream)
@@ -207,8 +205,9 @@ func (s *rtmpSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf cha
 }
 
 // apiSourceDescribe implements sourceStaticImpl.
-func (*rtmpSource) apiSourceDescribe() interface{} {
+func (*rtmpSource) apiSourceDescribe() pathAPISourceOrReader {
-	return struct {
+	return pathAPISourceOrReader{
-		Type string `json:"type"`
+		Type: "rtmpSource",
-	}{"rtmpSource"}
+		ID:   "",
+	}
 }

internal/core/rtsp_session.go

@@ -10,88 +10,15 @@ import (
 	"github.com/bluenviron/gortsplib/v3"
 	"github.com/bluenviron/gortsplib/v3/pkg/auth"
 	"github.com/bluenviron/gortsplib/v3/pkg/base"
-	"github.com/bluenviron/gortsplib/v3/pkg/formats"
-	"github.com/bluenviron/gortsplib/v3/pkg/media"
 	"github.com/bluenviron/gortsplib/v3/pkg/url"
 	"github.com/google/uuid"
 	"github.com/pion/rtp"
 
 	"github.com/aler9/mediamtx/internal/conf"
 	"github.com/aler9/mediamtx/internal/externalcmd"
-	"github.com/aler9/mediamtx/internal/formatprocessor"
 	"github.com/aler9/mediamtx/internal/logger"
 )
 
-type rtspWriteFunc func(*rtp.Packet)
-
-func getRTSPWriteFunc(medi *media.Media, forma formats.Format, stream *stream) rtspWriteFunc {
-	switch forma.(type) {
-	case *formats.H264:
-		return func(pkt *rtp.Packet) {
-			stream.writeUnit(medi, forma, &formatprocessor.UnitH264{
-				RTPPackets: []*rtp.Packet{pkt},
-				NTP:        time.Now(),
-			})
-		}
-
-	case *formats.H265:
-		return func(pkt *rtp.Packet) {
-			stream.writeUnit(medi, forma, &formatprocessor.UnitH265{
-				RTPPackets: []*rtp.Packet{pkt},
-				NTP:        time.Now(),
-			})
-		}
-
-	case *formats.VP8:
-		return func(pkt *rtp.Packet) {
-			stream.writeUnit(medi, forma, &formatprocessor.UnitVP8{
-				RTPPackets: []*rtp.Packet{pkt},
-				NTP:        time.Now(),
-			})
-		}
-
-	case *formats.VP9:
-		return func(pkt *rtp.Packet) {
-			stream.writeUnit(medi, forma, &formatprocessor.UnitVP9{
-				RTPPackets: []*rtp.Packet{pkt},
-				NTP:        time.Now(),
-			})
-		}
-
-	case *formats.MPEG2Audio:
-		return func(pkt *rtp.Packet) {
-			stream.writeUnit(medi, forma, &formatprocessor.UnitMPEG2Audio{
-				RTPPackets: []*rtp.Packet{pkt},
-				NTP:        time.Now(),
-			})
-		}
-
-	case *formats.MPEG4Audio:
-		return func(pkt *rtp.Packet) {
-			stream.writeUnit(medi, forma, &formatprocessor.UnitMPEG4Audio{
-				RTPPackets: []*rtp.Packet{pkt},
-				NTP:        time.Now(),
-			})
-		}
-
-	case *formats.Opus:
-		return func(pkt *rtp.Packet) {
-			stream.writeUnit(medi, forma, &formatprocessor.UnitOpus{
-				RTPPackets: []*rtp.Packet{pkt},
-				NTP:        time.Now(),
-			})
-		}
-
-	default:
-		return func(pkt *rtp.Packet) {
-			stream.writeUnit(medi, forma, &formatprocessor.UnitGeneric{
-				RTPPackets: []*rtp.Packet{pkt},
-				NTP:        time.Now(),
-			})
-		}
-	}
-}
-
 type rtspSessionPathManager interface {
 	publisherAdd(req pathPublisherAddReq) pathPublisherAnnounceRes
 	readerAdd(req pathReaderAddReq) pathReaderSetupPlayRes
@@ -387,10 +314,11 @@ func (s *rtspSession) onRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.R
 
 	for _, medi := range s.session.AnnouncedMedias() {
 		for _, forma := range medi.Formats {
-			writeFunc := getRTSPWriteFunc(medi, forma, s.stream)
+			cmedi := medi
+			cforma := forma
 
-			ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
+			ctx.Session.OnPacketRTP(cmedi, cforma, func(pkt *rtp.Packet) {
-				writeFunc(pkt)
+				res.stream.writeRTPPacket(cmedi, cforma, pkt, time.Now())
 			})
 		}
 	}
@@ -431,23 +359,21 @@ func (s *rtspSession) onPause(ctx *gortsplib.ServerHandlerOnPauseCtx) (*base.Res
 }
 
 // apiReaderDescribe implements reader.
-func (s *rtspSession) apiReaderDescribe() interface{} {
+func (s *rtspSession) apiReaderDescribe() pathAPISourceOrReader {
-	return s.apiSourceDescribe()
+	return pathAPISourceOrReader{
+		Type: func() string {
+			if s.isTLS {
+				return "rtspsSession"
+			}
+			return "rtspSession"
+		}(),
+		ID: s.uuid.String(),
+	}
 }
 
 // apiSourceDescribe implements source.
-func (s *rtspSession) apiSourceDescribe() interface{} {
+func (s *rtspSession) apiSourceDescribe() pathAPISourceOrReader {
-	var typ string
+	return s.apiReaderDescribe()
-	if s.isTLS {
-		typ = "rtspsSession"
-	} else {
-		typ = "rtspSession"
-	}
-
-	return struct {
-		Type string `json:"type"`
-		ID   string `json:"id"`
-	}{typ, s.uuid.String()}
 }
 
 // onPacketLost is called by rtspServer.

internal/core/rtsp_source.go

@@ -131,16 +131,15 @@ func (s *rtspSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf cha
 
 			s.Log(logger.Info, "ready: %s", sourceMediaInfo(medias))
 
-			defer func() {
+			defer s.parent.sourceStaticImplSetNotReady(pathSourceStaticSetNotReadyReq{})
-				s.parent.sourceStaticImplSetNotReady(pathSourceStaticSetNotReadyReq{})
-			}()
 
 			for _, medi := range medias {
 				for _, forma := range medi.Formats {
-					writeFunc := getRTSPWriteFunc(medi, forma, res.stream)
+					cmedi := medi
+					cforma := forma
 
-					c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
+					c.OnPacketRTP(cmedi, cforma, func(pkt *rtp.Packet) {
-						writeFunc(pkt)
+						res.stream.writeRTPPacket(cmedi, cforma, pkt, time.Now())
 					})
 				}
 			}
@@ -170,8 +169,9 @@ func (s *rtspSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf cha
 }
 
 // apiSourceDescribe implements sourceStaticImpl.
-func (*rtspSource) apiSourceDescribe() interface{} {
+func (*rtspSource) apiSourceDescribe() pathAPISourceOrReader {
-	return struct {
+	return pathAPISourceOrReader{
-		Type string `json:"type"`
+		Type: "rtspSource",
-	}{"rtspSource"}
+		ID:   "",
+	}
 }

internal/core/source.go

@@ -16,7 +16,7 @@ import (
 // - sourceRedirect
 type source interface {
 	logger.Writer
-	apiSourceDescribe() interface{}
+	apiSourceDescribe() pathAPISourceOrReader
 }
 
 func mediaDescription(media *media.Media) string {

internal/core/source_redirect.go

@@ -11,8 +11,9 @@ func (*sourceRedirect) Log(logger.Level, string, ...interface{}) {
 }
 
 // apiSourceDescribe implements source.
-func (*sourceRedirect) apiSourceDescribe() interface{} {
+func (*sourceRedirect) apiSourceDescribe() pathAPISourceOrReader {
-	return struct {
+	return pathAPISourceOrReader{
-		Type string `json:"type"`
+		Type: "redirect",
-	}{"redirect"}
+		ID:   "",
+	}
 }

internal/core/source_static.go

@@ -17,7 +17,7 @@ const (
 type sourceStaticImpl interface {
 	logger.Writer
 	run(context.Context, *conf.PathConf, chan *conf.PathConf) error
-	apiSourceDescribe() interface{}
+	apiSourceDescribe() pathAPISourceOrReader
 }
 
 type sourceStaticParent interface {
@@ -201,7 +201,7 @@ func (s *sourceStatic) reloadConf(newConf *conf.PathConf) {
 }
 
 // apiSourceDescribe implements source.
-func (s *sourceStatic) apiSourceDescribe() interface{} {
+func (s *sourceStatic) apiSourceDescribe() pathAPISourceOrReader {
 	return s.impl.apiSourceDescribe()
 }
 

internal/core/stream.go

@@ -1,9 +1,12 @@
 package core
 
 import (
+	"time"
+
 	"github.com/bluenviron/gortsplib/v3"
 	"github.com/bluenviron/gortsplib/v3/pkg/formats"
 	"github.com/bluenviron/gortsplib/v3/pkg/media"
+	"github.com/pion/rtp"
 
 	"github.com/aler9/mediamtx/internal/formatprocessor"
 )
@@ -67,3 +70,9 @@ func (s *stream) writeUnit(medi *media.Media, forma formats.Format, data formatp
 	sf := sm.formats[forma]
 	sf.writeUnit(s, medi, data)
 }
+
+func (s *stream) writeRTPPacket(medi *media.Media, forma formats.Format, pkt *rtp.Packet, ntp time.Time) {
+	sm := s.smedias[medi]
+	sf := sm.formats[forma]
+	sf.writeRTPPacket(s, medi, pkt, ntp)
+}

internal/core/stream_format.go

@@ -3,9 +3,11 @@ package core
 import (
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/bluenviron/gortsplib/v3/pkg/formats"
 	"github.com/bluenviron/gortsplib/v3/pkg/media"
+	"github.com/pion/rtp"
 
 	"github.com/aler9/mediamtx/internal/formatprocessor"
 	"github.com/aler9/mediamtx/internal/logger"
@@ -73,3 +75,7 @@ func (sf *streamFormat) writeUnit(s *stream, medi *media.Media, data formatproce
 		cb(data)
 	}
 }
+
+func (sf *streamFormat) writeRTPPacket(s *stream, medi *media.Media, pkt *rtp.Packet, ntp time.Time) {
+	sf.writeUnit(s, medi, sf.proc.UnitForRTPPacket(pkt, ntp))
+}

internal/core/udp_source.go

@@ -304,9 +304,7 @@ func (s *udpSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
 				return res.err
 			}
 
-			defer func() {
+			defer s.parent.sourceStaticImplSetNotReady(pathSourceStaticSetNotReadyReq{})
-				s.parent.sourceStaticImplSetNotReady(pathSourceStaticSetNotReadyReq{})
-			}()
 
 			s.Log(logger.Info, "ready: %s", sourceMediaInfo(medias))
 
@@ -360,8 +358,9 @@ func (s *udpSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
 }
 
 // apiSourceDescribe implements sourceStaticImpl.
-func (*udpSource) apiSourceDescribe() interface{} {
+func (*udpSource) apiSourceDescribe() pathAPISourceOrReader {
-	return struct {
+	return pathAPISourceOrReader{
-		Type string `json:"type"`
+		Type: "udpSource",
-	}{"udpSource"}
+		ID:   "",
+	}
 }

internal/core/webrtc_candidate_reader.go

@@ -0,0 +1,73 @@
+package core
+
+import (
+	"context"
+
+	"github.com/pion/webrtc/v3"
+
+	"github.com/aler9/mediamtx/internal/websocket"
+)
+
+type webRTCCandidateReader struct {
+	ws *websocket.ServerConn
+
+	ctx       context.Context
+	ctxCancel func()
+
+	stopGathering   chan struct{}
+	readError       chan error
+	remoteCandidate chan *webrtc.ICECandidateInit
+}
+
+func newWebRTCCandidateReader(ws *websocket.ServerConn) *webRTCCandidateReader {
+	ctx, ctxCancel := context.WithCancel(context.Background())
+
+	r := &webRTCCandidateReader{
+		ws:              ws,
+		ctx:             ctx,
+		ctxCancel:       ctxCancel,
+		stopGathering:   make(chan struct{}),
+		readError:       make(chan error),
+		remoteCandidate: make(chan *webrtc.ICECandidateInit),
+	}
+
+	go r.run()
+
+	return r
+}
+
+func (r *webRTCCandidateReader) close() {
+	r.ctxCancel()
+	// do not wait for ReadJSON() to return
+	// it is terminated by ws.Close() later
+}
+
+func (r *webRTCCandidateReader) run() {
+	for {
+		candidate, err := r.readCandidate()
+		if err != nil {
+			select {
+			case r.readError <- err:
+			case <-r.ctx.Done():
+			}
+			return
+		}
+
+		select {
+		case r.remoteCandidate <- candidate:
+		case <-r.stopGathering:
+		case <-r.ctx.Done():
+			return
+		}
+	}
+}
+
+func (r *webRTCCandidateReader) readCandidate() (*webrtc.ICECandidateInit, error) {
+	var candidate webrtc.ICECandidateInit
+	err := r.ws.ReadJSON(&candidate)
+	if err != nil {
+		return nil, err
+	}
+
+	return &candidate, err
+}

internal/core/webrtc_incoming_track.go

@@ -0,0 +1,133 @@
+package core
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/bluenviron/gortsplib/v3/pkg/formats"
+	"github.com/bluenviron/gortsplib/v3/pkg/media"
+	"github.com/pion/rtcp"
+	"github.com/pion/webrtc/v3"
+)
+
+const (
+	keyFrameInterval = 2 * time.Second
+)
+
+type webRTCIncomingTrack struct {
+	track     *webrtc.TrackRemote
+	receiver  *webrtc.RTPReceiver
+	writeRTCP func([]rtcp.Packet) error
+
+	mediaType media.Type
+	format    formats.Format
+	media     *media.Media
+}
+
+func newWebRTCIncomingTrack(
+	track *webrtc.TrackRemote,
+	receiver *webrtc.RTPReceiver,
+	writeRTCP func([]rtcp.Packet) error,
+) (*webRTCIncomingTrack, error) {
+	t := &webRTCIncomingTrack{
+		track:     track,
+		receiver:  receiver,
+		writeRTCP: writeRTCP,
+	}
+
+	switch track.Codec().MimeType {
+	case webrtc.MimeTypeAV1:
+		t.mediaType = media.TypeVideo
+		t.format = &formats.AV1{
+			PayloadTyp: uint8(track.PayloadType()),
+		}
+
+	case webrtc.MimeTypeVP9:
+		t.mediaType = media.TypeVideo
+		t.format = &formats.VP9{
+			PayloadTyp: uint8(track.PayloadType()),
+		}
+
+	case webrtc.MimeTypeVP8:
+		t.mediaType = media.TypeVideo
+		t.format = &formats.VP8{
+			PayloadTyp: uint8(track.PayloadType()),
+		}
+
+	case webrtc.MimeTypeH264:
+		t.mediaType = media.TypeVideo
+		t.format = &formats.H264{
+			PayloadTyp:        uint8(track.PayloadType()),
+			PacketizationMode: 1,
+		}
+
+	case webrtc.MimeTypeOpus:
+		t.mediaType = media.TypeAudio
+		t.format = &formats.Opus{
+			PayloadTyp: uint8(track.PayloadType()),
+		}
+
+	case webrtc.MimeTypeG722:
+		t.mediaType = media.TypeAudio
+		t.format = &formats.G722{}
+
+	case webrtc.MimeTypePCMU:
+		t.mediaType = media.TypeAudio
+		t.format = &formats.G711{MULaw: true}
+
+	case webrtc.MimeTypePCMA:
+		t.mediaType = media.TypeAudio
+		t.format = &formats.G711{MULaw: false}
+
+	default:
+		return nil, fmt.Errorf("unsupported codec: %v", track.Codec())
+	}
+
+	t.media = &media.Media{
+		Type:    t.mediaType,
+		Formats: []formats.Format{t.format},
+	}
+
+	return t, nil
+}
+
+func (t *webRTCIncomingTrack) start(stream *stream) {
+	go func() {
+		for {
+			pkt, _, err := t.track.ReadRTP()
+			if err != nil {
+				return
+			}
+
+			stream.writeRTPPacket(t.media, t.format, pkt, time.Now())
+		}
+	}()
+
+	// read incoming RTCP packets to make interceptors work
+	go func() {
+		buf := make([]byte, 1500)
+		for {
+			_, _, err := t.receiver.Read(buf)
+			if err != nil {
+				return
+			}
+		}
+	}()
+
+	if t.mediaType == media.TypeVideo {
+		go func() {
+			keyframeTicker := time.NewTicker(keyFrameInterval)
+
+			for range keyframeTicker.C {
+				err := t.writeRTCP([]rtcp.Packet{
+					&rtcp.PictureLossIndication{
+						MediaSSRC: uint32(t.track.SSRC()),
+					},
+				})
+				if err != nil {
+					return
+				}
+			}
+		}()
+	}
+}

internal/core/webrtc_outgoing_track.go

@@ -0,0 +1,333 @@
+package core
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/aler9/mediamtx/internal/formatprocessor"
+	"github.com/bluenviron/gortsplib/v3/pkg/formats"
+	"github.com/bluenviron/gortsplib/v3/pkg/formats/rtpav1"
+	"github.com/bluenviron/gortsplib/v3/pkg/formats/rtph264"
+	"github.com/bluenviron/gortsplib/v3/pkg/formats/rtpvp8"
+	"github.com/bluenviron/gortsplib/v3/pkg/formats/rtpvp9"
+	"github.com/bluenviron/gortsplib/v3/pkg/media"
+	"github.com/pion/webrtc/v3"
+)
+
+type webRTCOutgoingTrack struct {
+	sender *webrtc.RTPSender
+	media  *media.Media
+	format formats.Format
+	track  *webrtc.TrackLocalStaticRTP
+	cb     func(formatprocessor.Unit, context.Context, chan error)
+}
+
+func newWebRTCOutgoingTrackVideo(medias media.Medias) (*webRTCOutgoingTrack, error) {
+	var av1Format *formats.AV1
+	av1Media := medias.FindFormat(&av1Format)
+
+	if av1Format != nil {
+		webRTCTrak, err := webrtc.NewTrackLocalStaticRTP(
+			webrtc.RTPCodecCapability{
+				MimeType:  webrtc.MimeTypeAV1,
+				ClockRate: 90000,
+			},
+			"av1",
+			"rtspss",
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		encoder := &rtpav1.Encoder{
+			PayloadType:    105,
+			PayloadMaxSize: webrtcPayloadMaxSize,
+		}
+		encoder.Init()
+
+		return &webRTCOutgoingTrack{
+			media:  av1Media,
+			format: av1Format,
+			track:  webRTCTrak,
+			cb: func(unit formatprocessor.Unit, ctx context.Context, writeError chan error) {
+				tunit := unit.(*formatprocessor.UnitAV1)
+
+				if tunit.OBUs == nil {
+					return
+				}
+
+				packets, err := encoder.Encode(tunit.OBUs, tunit.PTS)
+				if err != nil {
+					return
+				}
+
+				for _, pkt := range packets {
+					webRTCTrak.WriteRTP(pkt)
+				}
+			},
+		}, nil
+	}
+
+	var vp9Format *formats.VP9
+	vp9Media := medias.FindFormat(&vp9Format)
+
+	if vp9Format != nil {
+		webRTCTrak, err := webrtc.NewTrackLocalStaticRTP(
+			webrtc.RTPCodecCapability{
+				MimeType:  webrtc.MimeTypeVP9,
+				ClockRate: uint32(vp9Format.ClockRate()),
+			},
+			"vp9",
+			"rtspss",
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		encoder := &rtpvp9.Encoder{
+			PayloadType:    96,
+			PayloadMaxSize: webrtcPayloadMaxSize,
+		}
+		encoder.Init()
+
+		return &webRTCOutgoingTrack{
+			media:  vp9Media,
+			format: vp9Format,
+			track:  webRTCTrak,
+			cb: func(unit formatprocessor.Unit, ctx context.Context, writeError chan error) {
+				tunit := unit.(*formatprocessor.UnitVP9)
+
+				if tunit.Frame == nil {
+					return
+				}
+
+				packets, err := encoder.Encode(tunit.Frame, tunit.PTS)
+				if err != nil {
+					return
+				}
+
+				for _, pkt := range packets {
+					webRTCTrak.WriteRTP(pkt)
+				}
+			},
+		}, nil
+	}
+
+	var vp8Format *formats.VP8
+	vp8Media := medias.FindFormat(&vp8Format)
+
+	if vp8Format != nil {
+		webRTCTrak, err := webrtc.NewTrackLocalStaticRTP(
+			webrtc.RTPCodecCapability{
+				MimeType:  webrtc.MimeTypeVP8,
+				ClockRate: uint32(vp8Format.ClockRate()),
+			},
+			"vp8",
+			"rtspss",
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		encoder := &rtpvp8.Encoder{
+			PayloadType:    96,
+			PayloadMaxSize: webrtcPayloadMaxSize,
+		}
+		encoder.Init()
+
+		return &webRTCOutgoingTrack{
+			media:  vp8Media,
+			format: vp8Format,
+			track:  webRTCTrak,
+			cb: func(unit formatprocessor.Unit, ctx context.Context, writeError chan error) {
+				tunit := unit.(*formatprocessor.UnitVP8)
+
+				if tunit.Frame == nil {
+					return
+				}
+
+				packets, err := encoder.Encode(tunit.Frame, tunit.PTS)
+				if err != nil {
+					return
+				}
+
+				for _, pkt := range packets {
+					webRTCTrak.WriteRTP(pkt)
+				}
+			},
+		}, nil
+	}
+
+	var h264Format *formats.H264
+	h264Media := medias.FindFormat(&h264Format)
+
+	if h264Format != nil {
+		webRTCTrak, err := webrtc.NewTrackLocalStaticRTP(
+			webrtc.RTPCodecCapability{
+				MimeType:  webrtc.MimeTypeH264,
+				ClockRate: uint32(h264Format.ClockRate()),
+			},
+			"h264",
+			"rtspss",
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		encoder := &rtph264.Encoder{
+			PayloadType:    96,
+			PayloadMaxSize: webrtcPayloadMaxSize,
+		}
+		encoder.Init()
+
+		var lastPTS time.Duration
+		firstNALUReceived := false
+
+		return &webRTCOutgoingTrack{
+			media:  h264Media,
+			format: h264Format,
+			track:  webRTCTrak,
+			cb: func(unit formatprocessor.Unit, ctx context.Context, writeError chan error) {
+				tunit := unit.(*formatprocessor.UnitH264)
+
+				if tunit.AU == nil {
+					return
+				}
+
+				if !firstNALUReceived {
+					firstNALUReceived = true
+					lastPTS = tunit.PTS
+				} else {
+					if tunit.PTS < lastPTS {
+						select {
+						case writeError <- fmt.Errorf("WebRTC doesn't support H264 streams with B-frames"):
+						case <-ctx.Done():
+						}
+						return
+					}
+					lastPTS = tunit.PTS
+				}
+
+				packets, err := encoder.Encode(tunit.AU, tunit.PTS)
+				if err != nil {
+					return
+				}
+
+				for _, pkt := range packets {
+					webRTCTrak.WriteRTP(pkt)
+				}
+			},
+		}, nil
+	}
+
+	return nil, nil
+}
+
+func newWebRTCOutgoingTrackAudio(medias media.Medias) (*webRTCOutgoingTrack, error) {
+	var opusFormat *formats.Opus
+	opusMedia := medias.FindFormat(&opusFormat)
+
+	if opusFormat != nil {
+		webRTCTrak, err := webrtc.NewTrackLocalStaticRTP(
+			webrtc.RTPCodecCapability{
+				MimeType:  webrtc.MimeTypeOpus,
+				ClockRate: uint32(opusFormat.ClockRate()),
+			},
+			"opus",
+			"rtspss",
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		return &webRTCOutgoingTrack{
+			media:  opusMedia,
+			format: opusFormat,
+			track:  webRTCTrak,
+			cb: func(unit formatprocessor.Unit, ctx context.Context, writeError chan error) {
+				for _, pkt := range unit.GetRTPPackets() {
+					webRTCTrak.WriteRTP(pkt)
+				}
+			},
+		}, nil
+	}
+
+	var g722Format *formats.G722
+	g722Media := medias.FindFormat(&g722Format)
+
+	if g722Format != nil {
+		webRTCTrak, err := webrtc.NewTrackLocalStaticRTP(
+			webrtc.RTPCodecCapability{
+				MimeType:  webrtc.MimeTypeG722,
+				ClockRate: uint32(g722Format.ClockRate()),
+			},
+			"g722",
+			"rtspss",
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		return &webRTCOutgoingTrack{
+			media:  g722Media,
+			format: g722Format,
+			track:  webRTCTrak,
+			cb: func(unit formatprocessor.Unit, ctx context.Context, writeError chan error) {
+				for _, pkt := range unit.GetRTPPackets() {
+					webRTCTrak.WriteRTP(pkt)
+				}
+			},
+		}, nil
+	}
+
+	var g711Format *formats.G711
+	g711Media := medias.FindFormat(&g711Format)
+
+	if g711Format != nil {
+		var mtyp string
+		if g711Format.MULaw {
+			mtyp = webrtc.MimeTypePCMU
+		} else {
+			mtyp = webrtc.MimeTypePCMA
+		}
+
+		webRTCTrak, err := webrtc.NewTrackLocalStaticRTP(
+			webrtc.RTPCodecCapability{
+				MimeType:  mtyp,
+				ClockRate: uint32(g711Format.ClockRate()),
+			},
+			"g711",
+			"rtspss",
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		return &webRTCOutgoingTrack{
+			media:  g711Media,
+			format: g711Format,
+			track:  webRTCTrak,
+			cb: func(unit formatprocessor.Unit, ctx context.Context, writeError chan error) {
+				for _, pkt := range unit.GetRTPPackets() {
+					webRTCTrak.WriteRTP(pkt)
+				}
+			},
+		}, nil
+	}
+
+	return nil, nil
+}
+
+func (t *webRTCOutgoingTrack) start() {
+	// read incoming RTCP packets to make interceptors work
+	go func() {
+		buf := make([]byte, 1500)
+		for {
+			_, _, err := t.sender.Read(buf)
+			if err != nil {
+				return
+			}
+		}
+	}()
+}

internal/core/webrtc_pc.go

@@ -0,0 +1,332 @@
+package core
+
+import (
+	"strconv"
+	"sync"
+
+	"github.com/pion/ice/v2"
+	"github.com/pion/interceptor"
+	"github.com/pion/webrtc/v3"
+
+	"github.com/aler9/mediamtx/internal/logger"
+)
+
+type peerConnection struct {
+	*webrtc.PeerConnection
+	stateChangeMutex   sync.Mutex
+	localCandidateRecv chan *webrtc.ICECandidateInit
+	connected          chan struct{}
+	disconnected       chan struct{}
+	closed             chan struct{}
+}
+
+func newPeerConnection(
+	videoCodec string,
+	audioCodec string,
+	iceServers []webrtc.ICEServer,
+	iceHostNAT1To1IPs []string,
+	iceUDPMux ice.UDPMux,
+	iceTCPMux ice.TCPMux,
+	log logger.Writer,
+) (*peerConnection, error) {
+	configuration := webrtc.Configuration{ICEServers: iceServers}
+	settingsEngine := webrtc.SettingEngine{}
+
+	if len(iceHostNAT1To1IPs) != 0 {
+		settingsEngine.SetNAT1To1IPs(iceHostNAT1To1IPs, webrtc.ICECandidateTypeHost)
+	}
+
+	if iceUDPMux != nil {
+		settingsEngine.SetICEUDPMux(iceUDPMux)
+	}
+
+	if iceTCPMux != nil {
+		settingsEngine.SetICETCPMux(iceTCPMux)
+		settingsEngine.SetNetworkTypes([]webrtc.NetworkType{webrtc.NetworkTypeTCP4})
+	}
+
+	mediaEngine := &webrtc.MediaEngine{}
+
+	if videoCodec != "" || audioCodec != "" {
+		switch videoCodec {
+		case "av1":
+			err := mediaEngine.RegisterCodec(
+				webrtc.RTPCodecParameters{
+					RTPCodecCapability: webrtc.RTPCodecCapability{
+						MimeType:  webrtc.MimeTypeAV1,
+						ClockRate: 90000,
+					},
+					PayloadType: 96,
+				},
+				webrtc.RTPCodecTypeVideo)
+			if err != nil {
+				return nil, err
+			}
+
+		case "vp9":
+			err := mediaEngine.RegisterCodec(
+				webrtc.RTPCodecParameters{
+					RTPCodecCapability: webrtc.RTPCodecCapability{
+						MimeType:    webrtc.MimeTypeVP9,
+						ClockRate:   90000,
+						SDPFmtpLine: "profile-id=0",
+					},
+					PayloadType: 96,
+				},
+				webrtc.RTPCodecTypeVideo)
+			if err != nil {
+				return nil, err
+			}
+
+			err = mediaEngine.RegisterCodec(
+				webrtc.RTPCodecParameters{
+					RTPCodecCapability: webrtc.RTPCodecCapability{
+						MimeType:    webrtc.MimeTypeVP9,
+						ClockRate:   90000,
+						SDPFmtpLine: "profile-id=1",
+					},
+					PayloadType: 96,
+				},
+				webrtc.RTPCodecTypeVideo)
+			if err != nil {
+				return nil, err
+			}
+
+		case "vp8":
+			err := mediaEngine.RegisterCodec(
+				webrtc.RTPCodecParameters{
+					RTPCodecCapability: webrtc.RTPCodecCapability{
+						MimeType:  webrtc.MimeTypeVP8,
+						ClockRate: 90000,
+					},
+					PayloadType: 96,
+				},
+				webrtc.RTPCodecTypeVideo)
+			if err != nil {
+				return nil, err
+			}
+
+		case "h264":
+			err := mediaEngine.RegisterCodec(
+				webrtc.RTPCodecParameters{
+					RTPCodecCapability: webrtc.RTPCodecCapability{
+						MimeType:    webrtc.MimeTypeH264,
+						ClockRate:   90000,
+						SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42001f",
+					},
+					PayloadType: 96,
+				},
+				webrtc.RTPCodecTypeVideo)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		switch audioCodec {
+		case "opus":
+			err := mediaEngine.RegisterCodec(
+				webrtc.RTPCodecParameters{
+					RTPCodecCapability: webrtc.RTPCodecCapability{
+						MimeType:    webrtc.MimeTypeOpus,
+						ClockRate:   48000,
+						Channels:    2,
+						SDPFmtpLine: "minptime=10;useinbandfec=1",
+					},
+					PayloadType: 111,
+				},
+				webrtc.RTPCodecTypeAudio)
+			if err != nil {
+				return nil, err
+			}
+
+		case "g722":
+			err := mediaEngine.RegisterCodec(
+				webrtc.RTPCodecParameters{
+					RTPCodecCapability: webrtc.RTPCodecCapability{
+						MimeType:  webrtc.MimeTypeG722,
+						ClockRate: 8000,
+					},
+					PayloadType: 9,
+				},
+				webrtc.RTPCodecTypeAudio)
+			if err != nil {
+				return nil, err
+			}
+
+		case "pcmu":
+			err := mediaEngine.RegisterCodec(
+				webrtc.RTPCodecParameters{
+					RTPCodecCapability: webrtc.RTPCodecCapability{
+						MimeType:  webrtc.MimeTypePCMU,
+						ClockRate: 8000,
+					},
+					PayloadType: 0,
+				},
+				webrtc.RTPCodecTypeAudio)
+			if err != nil {
+				return nil, err
+			}
+
+		case "pcma":
+			err := mediaEngine.RegisterCodec(
+				webrtc.RTPCodecParameters{
+					RTPCodecCapability: webrtc.RTPCodecCapability{
+						MimeType:  webrtc.MimeTypePCMA,
+						ClockRate: 8000,
+					},
+					PayloadType: 8,
+				},
+				webrtc.RTPCodecTypeAudio)
+			if err != nil {
+				return nil, err
+			}
+		}
+	} else {
+		// register all codecs
+		err := mediaEngine.RegisterDefaultCodecs()
+		if err != nil {
+			return nil, err
+		}
+		err = mediaEngine.RegisterCodec(
+			webrtc.RTPCodecParameters{
+				RTPCodecCapability: webrtc.RTPCodecCapability{
+					MimeType:  webrtc.MimeTypeAV1,
+					ClockRate: 90000,
+				},
+				PayloadType: 105,
+			},
+			webrtc.RTPCodecTypeVideo)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	interceptorRegistry := &interceptor.Registry{}
+	if err := webrtc.RegisterDefaultInterceptors(mediaEngine, interceptorRegistry); err != nil {
+		return nil, err
+	}
+
+	api := webrtc.NewAPI(
+		webrtc.WithSettingEngine(settingsEngine),
+		webrtc.WithMediaEngine(mediaEngine),
+		webrtc.WithInterceptorRegistry(interceptorRegistry))
+
+	pc, err := api.NewPeerConnection(configuration)
+	if err != nil {
+		return nil, err
+	}
+
+	co := &peerConnection{
+		PeerConnection:     pc,
+		localCandidateRecv: make(chan *webrtc.ICECandidateInit),
+		connected:          make(chan struct{}),
+		disconnected:       make(chan struct{}),
+		closed:             make(chan struct{}),
+	}
+
+	pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
+		co.stateChangeMutex.Lock()
+		defer co.stateChangeMutex.Unlock()
+
+		select {
+		case <-co.closed:
+			return
+		default:
+		}
+
+		log.Log(logger.Debug, "peer connection state: "+state.String())
+
+		switch state {
+		case webrtc.PeerConnectionStateConnected:
+			close(co.connected)
+
+		case webrtc.PeerConnectionStateDisconnected:
+			close(co.disconnected)
+
+		case webrtc.PeerConnectionStateClosed:
+			close(co.closed)
+		}
+	})
+
+	pc.OnICECandidate(func(i *webrtc.ICECandidate) {
+		if i != nil {
+			v := i.ToJSON()
+			select {
+			case co.localCandidateRecv <- &v:
+			case <-co.connected:
+			case <-co.closed:
+			}
+		}
+	})
+
+	return co, nil
+}
+
+func (co *peerConnection) close() {
+	co.PeerConnection.Close()
+	<-co.closed
+}
+
+func (co *peerConnection) localCandidate() string {
+	var cid string
+	for _, stats := range co.GetStats() {
+		if tstats, ok := stats.(webrtc.ICECandidatePairStats); ok && tstats.Nominated {
+			cid = tstats.LocalCandidateID
+			break
+		}
+	}
+
+	if cid != "" {
+		for _, stats := range co.GetStats() {
+			if tstats, ok := stats.(webrtc.ICECandidateStats); ok && tstats.ID == cid {
+				return tstats.CandidateType.String() + "/" + tstats.Protocol + "/" +
+					tstats.IP + "/" + strconv.FormatInt(int64(tstats.Port), 10)
+			}
+		}
+	}
+
+	return ""
+}
+
+func (co *peerConnection) remoteCandidate() string {
+	var cid string
+	for _, stats := range co.GetStats() {
+		if tstats, ok := stats.(webrtc.ICECandidatePairStats); ok && tstats.Nominated {
+			cid = tstats.RemoteCandidateID
+			break
+		}
+	}
+
+	if cid != "" {
+		for _, stats := range co.GetStats() {
+			if tstats, ok := stats.(webrtc.ICECandidateStats); ok && tstats.ID == cid {
+				return tstats.CandidateType.String() + "/" + tstats.Protocol + "/" +
+					tstats.IP + "/" + strconv.FormatInt(int64(tstats.Port), 10)
+			}
+		}
+	}
+
+	return ""
+}
+
+func (co *peerConnection) bytesReceived() uint64 {
+	for _, stats := range co.GetStats() {
+		if tstats, ok := stats.(webrtc.TransportStats); ok {
+			if tstats.ID == "iceTransport" {
+				return tstats.BytesReceived
+			}
+		}
+	}
+	return 0
+}
+
+func (co *peerConnection) bytesSent() uint64 {
+	for _, stats := range co.GetStats() {
+		if tstats, ok := stats.(webrtc.TransportStats); ok {
+			if tstats.ID == "iceTransport" {
+				return tstats.BytesSent
+			}
+		}
+	}
+	return 0
+}

internal/core/webrtc_publish_index.html

@@ -0,0 +1,374 @@
+<!DOCTYPE html>
+<html>
+<head>
+<meta charset="utf-8">
+<meta name="viewport" content="width=device-width">
+<style>
+html, body {
+	margin: 0;
+	padding: 0;
+	height: 100%;
+	overflow: hidden;
+}
+body {
+    display: flex;
+    flex-direction: column;
+}
+#video {
+	height: 100%;
+	background: black;
+    flex-grow: 1;
+    min-height: 0;
+}
+#controls {
+    height: 200px;
+    flex-shrink: 0;
+    display: flex;
+    align-items: center;
+    justify-content: center;
+}
+#device {
+    flex-direction: column;
+}
+#device > div {
+    margin: 10px 0;
+    display: flex;
+    gap: 20px;
+    justify-content: center;
+}
+select {
+    width: 200px;
+}
+</style>
+</head>
+<body>
+
+<video id="video" muted controls autoplay playsinline></video>
+<div id="controls">
+    <div id="initializing" style="display: block;">
+        initializing
+    </div>
+    <div id="device" style="display: none;">
+        <div id="device_line">
+            video device:
+            <select id="video_device">
+                <option value="none">none</option>
+            </select>
+
+            audio device:
+            <select id="audio_device">
+                <option value="none">none</option>
+            </select>
+        </div>
+        <div id="codec_line">
+            video codec:
+            <select id="video_codec">
+            </select>
+
+            audio codec:
+            <select id="audio_codec">
+            </select>
+        </div>
+        <div id="bitrate_line">
+            video bitrate (kbps):
+            <input id="video_bitrate" type="text" value="10000" />
+        </div>
+        <div id="submit_line">
+            <button id="publish_confirm">publish</button>
+        </div>
+    </div>
+    <div id="transmitting" style="display: none;">
+        publishing
+    </div>
+</div>
+
+<script>
+
+const INITIALIZING = 0;
+const DEVICE = 1;
+const TRANSMITTING = 2;
+
+let state = INITIALIZING;
+
+const setState = (newState) => {
+    state = newState;
+
+    switch (state) {
+    case DEVICE:
+        document.getElementById("initializing").style.display = 'none';
+        document.getElementById("device").style.display = 'flex';
+        document.getElementById("transmitting").style.display = 'none';
+        break;
+
+    case TRANSMITTING:
+        document.getElementById("initializing").style.display = 'none';
+        document.getElementById("device").style.display = 'none';
+        document.getElementById("transmitting").style.display = 'flex';
+        break;
+    }
+};
+
+const restartPause = 2000;
+
+class Transmitter {
+    constructor(stream) {
+        this.stream = stream;
+        this.terminated = false;
+		this.ws = null;
+		this.pc = null;
+		this.restartTimeout = null;
+		this.start();
+    }
+
+    start = () => {
+        console.log("connecting");
+
+        const videoCodec = document.getElementById('video_codec').value;
+        const audioCodec = document.getElementById('audio_codec').value;
+        const videoBitrate = document.getElementById('video_bitrate').value;
+
+        const u = window.location.href.replace(/^http/, "ws") + '/ws' +
+            '?video_codec=' + videoCodec +
+            '&audio_codec=' + audioCodec +
+            '&video_bitrate=' + videoBitrate;
+
+        this.ws = new WebSocket(u);
+
+        this.ws.onerror = () => {
+            console.log("ws error");
+            if (this.ws === null) {
+                return;
+            }
+            this.ws.close();
+            this.ws = null;
+        };
+
+        this.ws.onclose = () => {
+            console.log("ws closed");
+            this.ws = null;
+            this.scheduleRestart();
+        };
+
+        this.ws.onmessage = this.onIceServers;
+    };
+
+    scheduleRestart = () => {
+        if (this.terminated) {
+            return;
+        }
+
+        if (this.ws !== null) {
+            this.ws.close();
+            this.ws = null;
+        }
+
+        if (this.pc !== null) {
+            this.pc.close();
+            this.pc = null;
+        }
+
+        this.restartTimeout = window.setTimeout(() => {
+            this.restartTimeout = null;
+            this.start();
+        }, restartPause);
+    };
+
+    onIceServers = (msg) => {
+        if (this.ws === null) {
+            return;
+        }
+
+        this.pc = new RTCPeerConnection({
+            iceServers: JSON.parse(msg.data),
+        });
+
+        this.ws.onmessage = this.onOffer;
+    };
+
+    onOffer = (msg) => {
+        if (this.ws === null || this.pc === null) {
+            return;
+        }
+
+        this.stream.getTracks().forEach((track) => {
+            this.pc.addTrack(track, this.stream);
+        });
+
+        this.ws.onmessage = (msg) => {
+            if (this.pc === null) {
+                return;
+            }
+            this.pc.addIceCandidate(JSON.parse(msg.data));
+        };
+
+        this.pc.onicecandidate = (evt) => {
+            if (this.ws === null) {
+                return;
+            }
+
+            if (evt.candidate !== null) {
+                if (evt.candidate.candidate !== "") {
+                    this.ws.send(JSON.stringify(evt.candidate));
+                }
+            }
+        };
+
+        this.pc.oniceconnectionstatechange = () => {
+            if (this.pc === null) {
+                return;
+            }
+
+            console.log("peer connection state:", this.pc.iceConnectionState);
+
+            switch (this.pc.iceConnectionState) {
+            case "failed":
+            case "disconnected":
+                this.scheduleRestart();
+            }
+        };
+
+        this.pc.setRemoteDescription(new RTCSessionDescription(JSON.parse(msg.data)));
+
+        this.pc.createAnswer()
+            .then((desc) => {
+                if (this.ws === null || this.pc === null) {
+                    return;
+                }
+
+                this.pc.setLocalDescription(desc);
+                this.ws.send(JSON.stringify(desc));
+            });
+    };
+}
+
+const onTransmit = (stream) => {
+    setState(TRANSMITTING);
+    document.getElementById('video').srcObject = stream;
+    new Transmitter(stream);
+};
+
+const onPublish = () => {
+    const videoId = document.getElementById('video_device').value;
+    const audioId = document.getElementById('audio_device').value;
+
+    if (videoId !== 'screen') {
+        let video = false;
+        if (videoId !== 'none') {
+            video = {
+                deviceId: videoId,
+            };
+        }
+
+        let audio = false;
+
+        if (audioId !== 'none') {
+            audio = {
+                deviceId: audioId,
+            };
+        }
+
+        navigator.mediaDevices.getUserMedia({ video, audio })
+            .then(onTransmit);
+    } else {
+        navigator.mediaDevices.getDisplayMedia({
+            video: {
+                width: { ideal: 1920 },
+                height: { ideal: 1080 },
+                frameRate: { ideal: 30 },
+                cursor: "always",
+            },
+            audio: false,
+        })
+            .then(onTransmit);
+    }
+};
+
+const populateDevices = () => {
+    return navigator.mediaDevices.enumerateDevices()
+        .then((devices) => {
+            for (const device of devices) {
+                switch (device.kind) {
+                case 'videoinput':
+                    {
+                        const opt = document.createElement('option');
+                        opt.value = device.deviceId;
+                        opt.text = device.label;
+                        document.getElementById('video_device').appendChild(opt);
+                    }
+                    break;
+
+                case 'audioinput':
+                    {
+                        const opt = document.createElement('option');
+                        opt.value = device.deviceId;
+                        opt.text = device.label;
+                        document.getElementById('audio_device').appendChild(opt);
+                    }
+                    break;
+                }
+            }
+
+            // add screen
+            const opt = document.createElement('option');
+            opt.value = "screen";
+            opt.text = "screen";
+            document.getElementById('video_device').appendChild(opt);
+
+            // set default
+            document.getElementById('video_device').value = document.getElementById('video_device').children[1].value;
+            if (document.getElementById('audio_device').children.length > 1) {
+                document.getElementById('audio_device').value = document.getElementById('audio_device').children[1].value;
+            }
+        });
+};
+
+const populateCodecs = () => {
+    const pc = new RTCPeerConnection({});
+    pc.addTransceiver("video", { direction: 'sendonly' });
+    pc.addTransceiver("audio", { direction: 'sendonly' });
+
+    return pc.createOffer()
+        .then((desc) => {
+            const sdp = desc.sdp.toLowerCase();
+
+            for (const codec of ['av1/90000', 'vp9/90000', 'vp8/90000', 'h264/90000']) {
+                if (sdp.includes(codec)) {
+                    const opt = document.createElement('option');
+                    opt.value = codec.split('/')[0];
+                    opt.text = codec.split('/')[0].toUpperCase();
+                    document.getElementById('video_codec').appendChild(opt);
+                }
+            }
+
+            for (const codec of ['opus/48000', 'g722/8000', 'pcmu/8000', 'pcma/8000']) {
+                if (sdp.includes(codec)) {
+                    const opt = document.createElement('option');
+                    opt.value = codec.split('/')[0];
+                    opt.text = codec.split('/')[0].toUpperCase();
+                    document.getElementById('audio_codec').appendChild(opt);
+                }
+            }
+
+            pc.close();
+        });
+};
+
+const initialize = () => {
+    navigator.mediaDevices.getUserMedia({ video: true, audio: true })
+        .then(() => Promise.all([
+            populateDevices(),
+            populateCodecs(),
+        ]))
+        .then(() => {
+            setState(DEVICE);
+        });
+};
+
+document.getElementById("publish_confirm").addEventListener('click', onPublish);
+
+initialize();
+
+</script>
+
+</body>
+</html>

internal/core/webrtc_read_index.html

@@ -62,10 +62,8 @@ class Receiver {
             return;
         }
 
-        const iceServers = JSON.parse(msg.data);
-
         this.pc = new RTCPeerConnection({
-            iceServers,
+            iceServers: JSON.parse(msg.data),
         });
 
         this.ws.onmessage = (msg) => this.onRemoteDescription(msg);

internal/core/webrtc_server.go

@@ -8,7 +8,6 @@ import (
 	"log"
 	"net"
 	"net/http"
-	gopath "path"
 	"strings"
 	"sync"
 	"time"
@@ -22,8 +21,11 @@ import (
 	"github.com/aler9/mediamtx/internal/websocket"
 )
 
-//go:embed webrtc_index.html
+//go:embed webrtc_publish_index.html
-var webrtcIndex []byte
+var webrtcPublishIndex []byte
+
+//go:embed webrtc_read_index.html
+var webrtcReadIndex []byte
 
 type webRTCServerAPIConnsListItem struct {
 	Created                   time.Time `json:"created"`
@@ -31,6 +33,7 @@ type webRTCServerAPIConnsListItem struct {
 	PeerConnectionEstablished bool      `json:"peerConnectionEstablished"`
 	LocalCandidate            string    `json:"localCandidate"`
 	RemoteCandidate           string    `json:"remoteCandidate"`
+	State                     string    `json:"state"`
 	BytesReceived             uint64    `json:"bytesReceived"`
 	BytesSent                 uint64    `json:"bytesSent"`
 }
@@ -58,9 +61,13 @@ type webRTCServerAPIConnsKickReq struct {
 }
 
 type webRTCConnNewReq struct {
-	pathName string
+	pathName     string
-	wsconn   *websocket.ServerConn
+	publish      bool
-	res      chan *webRTCConn
+	wsconn       *websocket.ServerConn
+	res          chan *webRTCConn
+	videoCodec   string
+	audioCodec   string
+	videoBitrate string
 }
 
 type webRTCServerParent interface {
@@ -242,7 +249,11 @@ outer:
 				s.ctx,
 				s.readBufferCount,
 				req.pathName,
+				req.publish,
 				req.wsconn,
+				req.videoCodec,
+				req.audioCodec,
+				req.videoBitrate,
 				s.iceServers,
 				&wg,
 				s.pathManager,
@@ -263,14 +274,35 @@ outer:
 			}
 
 			for c := range s.conns {
+				peerConnectionEstablished := false
+				localCandidate := ""
+				remoteCandidate := ""
+				bytesReceived := uint64(0)
+				bytesSent := uint64(0)
+
+				pc := c.safePC()
+				if pc != nil {
+					peerConnectionEstablished = true
+					localCandidate = pc.localCandidate()
+					remoteCandidate = pc.remoteCandidate()
+					bytesReceived = pc.bytesReceived()
+					bytesSent = pc.bytesSent()
+				}
+
 				data.Items[c.uuid.String()] = webRTCServerAPIConnsListItem{
 					Created:                   c.created,
 					RemoteAddr:                c.remoteAddr().String(),
-					PeerConnectionEstablished: c.peerConnectionEstablished(),
+					PeerConnectionEstablished: peerConnectionEstablished,
-					LocalCandidate:            c.localCandidate(),
+					LocalCandidate:            localCandidate,
-					RemoteCandidate:           c.remoteCandidate(),
+					RemoteCandidate:           remoteCandidate,
-					BytesReceived:             c.bytesReceived(),
+					State: func() string {
-					BytesSent:                 c.bytesSent(),
+						if c.publish {
+							return "publish"
+						}
+						return "read"
+					}(),
+					BytesReceived: bytesReceived,
+					BytesSent:     bytesSent,
 				}
 			}
 
@@ -302,8 +334,8 @@ outer:
 
 	s.httpServer.Shutdown(context.Background())
 	s.ln.Close() // in case Shutdown() is called before Serve()
-	s.requestPool.close()
 
+	s.requestPool.close()
 	wg.Wait()
 
 	if s.udpMuxLn != nil {
@@ -335,29 +367,51 @@ func (s *webRTCServer) onRequest(ctx *gin.Context) {
 	// remove leading prefix
 	pa := ctx.Request.URL.Path[1:]
 
-	switch pa {
+	var dir string
-	case "", "favicon.ico":
+	var fname string
-		return
+	var publish bool
-	}
 
-	dir, fname := func() (string, string) {
+	switch {
-		if strings.HasSuffix(pa, "/ws") {
+	case strings.HasSuffix(pa, "/publish/ws"):
-			return gopath.Dir(pa), gopath.Base(pa)
+		dir = pa[:len(pa)-len("/publish/ws")]
+		fname = "publish/ws"
+		publish = true
+
+	case strings.HasSuffix(pa, "/publish"):
+		dir = pa[:len(pa)-len("/publish")]
+		fname = "publish"
+		publish = true
+
+	case strings.HasSuffix(pa, "/ws"):
+		dir = pa[:len(pa)-len("/ws")]
+		fname = "ws"
+		publish = false
+
+	case pa == "favicon.ico":
+		return
+
+	default:
+		dir = pa
+		fname = ""
+		publish = false
+
+		if !strings.HasSuffix(dir, "/") {
+			ctx.Writer.Header().Set("Location", "/"+dir+"/")
+			ctx.Writer.WriteHeader(http.StatusMovedPermanently)
+			return
 		}
-		return pa, ""
-	}()
-
-	if fname == "" && !strings.HasSuffix(dir, "/") {
-		ctx.Writer.Header().Set("Location", "/"+dir+"/")
-		ctx.Writer.WriteHeader(http.StatusMovedPermanently)
-		return
 	}
 
 	dir = strings.TrimSuffix(dir, "/")
+	if dir == "" {
+		return
+	}
+
 	user, pass, hasCredentials := ctx.Request.BasicAuth()
 
 	res := s.pathManager.getPathConf(pathGetPathConfReq{
-		name: dir,
+		name:    dir,
+		publish: publish,
 		credentials: authCredentials{
 			query: ctx.Request.URL.RawQuery,
 			ip:    net.ParseIP(ctx.ClientIP()),
@@ -387,10 +441,14 @@ func (s *webRTCServer) onRequest(ctx *gin.Context) {
 	case "":
 		ctx.Writer.Header().Set("Content-Type", "text/html")
 		ctx.Writer.WriteHeader(http.StatusOK)
-		ctx.Writer.Write(webrtcIndex)
+		ctx.Writer.Write(webrtcReadIndex)
-		return
 
-	case "ws":
+	case "publish":
+		ctx.Writer.Header().Set("Content-Type", "text/html")
+		ctx.Writer.WriteHeader(http.StatusOK)
+		ctx.Writer.Write(webrtcPublishIndex)
+
+	case "ws", "publish/ws":
 		wsconn, err := websocket.NewServerConn(ctx.Writer, ctx.Request)
 		if err != nil {
 			return
@@ -398,8 +456,12 @@ func (s *webRTCServer) onRequest(ctx *gin.Context) {
 		defer wsconn.Close()
 
 		c := s.newConn(webRTCConnNewReq{
-			pathName: dir,
+			pathName:     dir,
-			wsconn:   wsconn,
+			publish:      (fname == "publish/ws"),
+			wsconn:       wsconn,
+			videoCodec:   ctx.Query("video_codec"),
+			audioCodec:   ctx.Query("audio_codec"),
+			videoBitrate: ctx.Query("video_bitrate"),
 		})
 		if c == nil {
 			return

internal/core/webrtc_server_test.go

@@ -42,7 +42,7 @@ func newWebRTCTestClient(addr string) (*webRTCTestClient, error) {
 		return nil, err
 	}
 
-	pc, err := newPeerConnection(webrtc.Configuration{
+	pc, err := webrtc.NewPeerConnection(webrtc.Configuration{
 		ICEServers: iceServers,
 	})
 	if err != nil {

internal/formatprocessor/av1.go

@@ -96,7 +96,7 @@ func (t *formatProcessorAV1) Process(unit Unit, hasNonRTSPReaders bool) error {
 		}
 
 		// decode from RTP
-		if hasNonRTSPReaders {
+		if hasNonRTSPReaders || t.decoder != nil {
 			if t.decoder == nil {
 				t.decoder = t.format.CreateDecoder()
 				t.lastKeyFrameReceived = time.Now()
@@ -131,3 +131,10 @@ func (t *formatProcessorAV1) Process(unit Unit, hasNonRTSPReaders bool) error {
 
 	return nil
 }
+
+func (t *formatProcessorAV1) UnitForRTPPacket(pkt *rtp.Packet, ntp time.Time) Unit {
+	return &UnitAV1{
+		RTPPackets: []*rtp.Packet{pkt},
+		NTP:        ntp,
+	}
+}

internal/formatprocessor/generic.go

@@ -61,3 +61,10 @@ func (t *formatProcessorGeneric) Process(unit Unit, hasNonRTSPReaders bool) erro
 
 	return nil
 }
+
+func (t *formatProcessorGeneric) UnitForRTPPacket(pkt *rtp.Packet, ntp time.Time) Unit {
+	return &UnitGeneric{
+		RTPPackets: []*rtp.Packet{pkt},
+		NTP:        ntp,
+	}
+}

internal/formatprocessor/h264.go

@@ -275,7 +275,7 @@ func (t *formatProcessorH264) Process(unit Unit, hasNonRTSPReaders bool) error {
 		}
 
 		// decode from RTP
-		if hasNonRTSPReaders || t.encoder != nil {
+		if hasNonRTSPReaders || t.decoder != nil || t.encoder != nil {
 			if t.decoder == nil {
 				t.decoder = t.format.CreateDecoder()
 				t.lastKeyFrameReceived = time.Now()
@@ -320,3 +320,10 @@ func (t *formatProcessorH264) Process(unit Unit, hasNonRTSPReaders bool) error {
 
 	return nil
 }
+
+func (t *formatProcessorH264) UnitForRTPPacket(pkt *rtp.Packet, ntp time.Time) Unit {
+	return &UnitH264{
+		RTPPackets: []*rtp.Packet{pkt},
+		NTP:        ntp,
+	}
+}

internal/formatprocessor/h265.go

@@ -296,7 +296,7 @@ func (t *formatProcessorH265) Process(unit Unit, hasNonRTSPReaders bool) error {
 		}
 
 		// decode from RTP
-		if hasNonRTSPReaders || t.encoder != nil {
+		if hasNonRTSPReaders || t.decoder != nil || t.encoder != nil {
 			if t.decoder == nil {
 				t.decoder = t.format.CreateDecoder()
 				t.lastKeyFrameReceived = time.Now()
@@ -341,3 +341,10 @@ func (t *formatProcessorH265) Process(unit Unit, hasNonRTSPReaders bool) error {
 
 	return nil
 }
+
+func (t *formatProcessorH265) UnitForRTPPacket(pkt *rtp.Packet, ntp time.Time) Unit {
+	return &UnitH265{
+		RTPPackets: []*rtp.Packet{pkt},
+		NTP:        ntp,
+	}
+}

internal/formatprocessor/mpeg2audio.go

@@ -73,7 +73,7 @@ func (t *formatProcessorMPEG2Audio) Process(unit Unit, hasNonRTSPReaders bool) e
 		}
 
 		// decode from RTP
-		if hasNonRTSPReaders {
+		if hasNonRTSPReaders || t.decoder != nil {
 			if t.decoder == nil {
 				t.decoder = t.format.CreateDecoder()
 			}
@@ -103,3 +103,10 @@ func (t *formatProcessorMPEG2Audio) Process(unit Unit, hasNonRTSPReaders bool) e
 
 	return nil
 }
+
+func (t *formatProcessorMPEG2Audio) UnitForRTPPacket(pkt *rtp.Packet, ntp time.Time) Unit {
+	return &UnitMPEG2Audio{
+		RTPPackets: []*rtp.Packet{pkt},
+		NTP:        ntp,
+	}
+}

internal/formatprocessor/mpeg4audio.go

@@ -78,7 +78,7 @@ func (t *formatProcessorMPEG4Audio) Process(unit Unit, hasNonRTSPReaders bool) e
 		}
 
 		// decode from RTP
-		if hasNonRTSPReaders {
+		if hasNonRTSPReaders || t.decoder != nil {
 			if t.decoder == nil {
 				t.decoder = t.format.CreateDecoder()
 			}
@@ -108,3 +108,10 @@ func (t *formatProcessorMPEG4Audio) Process(unit Unit, hasNonRTSPReaders bool) e
 
 	return nil
 }
+
+func (t *formatProcessorMPEG4Audio) UnitForRTPPacket(pkt *rtp.Packet, ntp time.Time) Unit {
+	return &UnitMPEG4Audio{
+		RTPPackets: []*rtp.Packet{pkt},
+		NTP:        ntp,
+	}
+}

internal/formatprocessor/opus.go

@@ -75,7 +75,7 @@ func (t *formatProcessorOpus) Process(unit Unit, hasNonRTSPReaders bool) error {
 		}
 
 		// decode from RTP
-		if hasNonRTSPReaders {
+		if hasNonRTSPReaders || t.decoder != nil {
 			if t.decoder == nil {
 				t.decoder = t.format.CreateDecoder()
 			}
@@ -102,3 +102,10 @@ func (t *formatProcessorOpus) Process(unit Unit, hasNonRTSPReaders bool) error {
 
 	return nil
 }
+
+func (t *formatProcessorOpus) UnitForRTPPacket(pkt *rtp.Packet, ntp time.Time) Unit {
+	return &UnitOpus{
+		RTPPackets: []*rtp.Packet{pkt},
+		NTP:        ntp,
+	}
+}

internal/formatprocessor/processor.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/bluenviron/gortsplib/v3/pkg/formats"
+	"github.com/pion/rtp"
 
 	"github.com/aler9/mediamtx/internal/logger"
 )
@@ -17,6 +18,9 @@ const (
 type Processor interface {
 	// cleans and normalizes a data unit.
 	Process(Unit, bool) error
+
+	// returns an unit for the given RTP packet.
+	UnitForRTPPacket(pkt *rtp.Packet, ntp time.Time) Unit
 }
 
 // New allocates a Processor.

internal/formatprocessor/unit.go

@@ -8,6 +8,9 @@ import (
 
 // Unit is the elementary data unit routed across the server.
 type Unit interface {
+	// returns RTP packets contained into the unit.
 	GetRTPPackets() []*rtp.Packet
+
+	// returns the NTP timestamp of the unit.
 	GetNTP() time.Time
 }

internal/formatprocessor/vp8.go

@@ -74,7 +74,7 @@ func (t *formatProcessorVP8) Process(unit Unit, hasNonRTSPReaders bool) error {
 		}
 
 		// decode from RTP
-		if hasNonRTSPReaders {
+		if hasNonRTSPReaders || t.decoder != nil {
 			if t.decoder == nil {
 				t.decoder = t.format.CreateDecoder()
 			}
@@ -104,3 +104,10 @@ func (t *formatProcessorVP8) Process(unit Unit, hasNonRTSPReaders bool) error {
 
 	return nil
 }
+
+func (t *formatProcessorVP8) UnitForRTPPacket(pkt *rtp.Packet, ntp time.Time) Unit {
+	return &UnitVP8{
+		RTPPackets: []*rtp.Packet{pkt},
+		NTP:        ntp,
+	}
+}

internal/formatprocessor/vp9.go

@@ -74,7 +74,7 @@ func (t *formatProcessorVP9) Process(unit Unit, hasNonRTSPReaders bool) error {
 		}
 
 		// decode from RTP
-		if hasNonRTSPReaders {
+		if hasNonRTSPReaders || t.decoder != nil {
 			if t.decoder == nil {
 				t.decoder = t.format.CreateDecoder()
 			}
@@ -104,3 +104,10 @@ func (t *formatProcessorVP9) Process(unit Unit, hasNonRTSPReaders bool) error {
 
 	return nil
 }
+
+func (t *formatProcessorVP9) UnitForRTPPacket(pkt *rtp.Packet, ntp time.Time) Unit {
+	return &UnitVP9{
+		RTPPackets: []*rtp.Packet{pkt},
+		NTP:        ntp,
+	}
+}