pimhwang
/
mediamtx
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

print warning in case no key frames are being received (#1763)

pull/1756/head
Alessandro Ros 2 years ago committed by GitHub
parent
cdd907b016
commit
225220ddd5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
49 changed files with 505 additions and 358 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      internal/conf/conf.go
  2. 37
      internal/conf/logdestination.go
  3. 8
      internal/core/api.go
  4. 16
      internal/core/hls_muxer.go
  5. 8
      internal/core/hls_server.go
  6. 24
      internal/core/hls_source.go
  7. 8
      internal/core/http_logger.go
  8. 8
      internal/core/metrics.go
  9. 33
      internal/core/path.go
  10. 8
      internal/core/path_manager.go
  11. 8
      internal/core/pprof.go
  12. 9
      internal/core/rpicamera_source.go
  13. 49
      internal/core/rtmp_conn.go
  14. 10
      internal/core/rtmp_server.go
  15. 4
      internal/core/rtmp_source.go
  16. 22
      internal/core/rtsp_conn.go
  17. 10
      internal/core/rtsp_server.go
  18. 69
      internal/core/rtsp_session.go
  19. 9
      internal/core/rtsp_source.go
  20. 3
      internal/core/source.go
  21. 7
      internal/core/source_redirect.go
  22. 8
      internal/core/source_static.go
  23. 12
      internal/core/stream.go
  24. 13
      internal/core/stream_format.go
  25. 3
      internal/core/stream_media.go
  26. 24
      internal/core/udp_source.go
  27. 20
      internal/core/webrtc_conn.go
  28. 10
      internal/core/webrtc_server.go
  29. 3
      internal/formatprocessor/generic.go
  30. 2
      internal/formatprocessor/generic_test.go
  31. 48
      internal/formatprocessor/h264.go
  32. 6
      internal/formatprocessor/h264_test.go
  33. 49
      internal/formatprocessor/h265.go
  34. 6
      internal/formatprocessor/h265_test.go
  35. 7
      internal/formatprocessor/mpeg2audio.go
  36. 7
      internal/formatprocessor/mpeg4audio.go
  37. 7
      internal/formatprocessor/opus.go
  38. 29
      internal/formatprocessor/processor.go
  39. 7
      internal/formatprocessor/vp8.go
  40. 7
      internal/formatprocessor/vp9.go
  41. 20
      internal/logger/destination.go
  42. 34
      internal/logger/destination_file.go
  43. 25
      internal/logger/destination_stdout.go
  44. 34
      internal/logger/destination_syslog.go
  45. 12
      internal/logger/level.go
  46. 108
      internal/logger/logger.go
  47. 2
      internal/logger/syslog_unix.go
  48. 2
      internal/logger/syslog_win.go
  49. 6
      internal/logger/writer.go

2
internal/conf/conf.go
Unescape View File

@ -308,7 +308,7 @@ func (conf *Conf) CheckAndFillMissing() error {
conf.LogLevel = LogLevel(logger.Info) conf.LogLevel = LogLevel(logger.Info)
} }
if len(conf.LogDestinations) == 0 { if len(conf.LogDestinations) == 0 {
conf.LogDestinations = LogDestinations{logger.DestinationStdout: {}} conf.LogDestinations = LogDestinations{logger.DestinationStdout}
} }
if conf.LogFile == "" { if conf.LogFile == "" {
conf.LogFile = "mediamtx.log" conf.LogFile = "mediamtx.log"

37
internal/conf/logdestination.go
Unescape View File

@ -3,21 +3,20 @@ package conf
import ( import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"sort"
"strings" "strings"
"github.com/aler9/mediamtx/internal/logger" "github.com/aler9/mediamtx/internal/logger"
) )
// LogDestinations is the logDestionations parameter. // LogDestinations is the logDestionations parameter.
type LogDestinations map[logger.Destination]struct{} type LogDestinations []logger.Destination
// MarshalJSON implements json.Marshaler. // MarshalJSON implements json.Marshaler.
func (d LogDestinations) MarshalJSON() ([]byte, error) { func (d LogDestinations) MarshalJSON() ([]byte, error) {
out := make([]string, len(d)) out := make([]string, len(d))
i := 0 i := 0
for p := range d { for _, p := range d {
var v string var v string
switch p { switch p {
@ -38,11 +37,18 @@ func (d LogDestinations) MarshalJSON() ([]byte, error) {
i++ i++
} }
sort.Strings(out)
return json.Marshal(out) return json.Marshal(out)
} }
func (d *LogDestinations) contains(v logger.Destination) bool {
for _, item := range *d {
if item == v {
return true
}
}
return false
}
// UnmarshalJSON implements json.Unmarshaler. // UnmarshalJSON implements json.Unmarshaler.
func (d *LogDestinations) UnmarshalJSON(b []byte) error { func (d *LogDestinations) UnmarshalJSON(b []byte) error {
var in []string var in []string
@ -50,22 +56,27 @@ func (d *LogDestinations) UnmarshalJSON(b []byte) error {
return err return err
} }
*d = make(LogDestinations) for _, dest := range in {
var v logger.Destination
for _, proto := range in { switch dest {
switch proto {
case "stdout": case "stdout":
(*d)[logger.DestinationStdout] = struct{}{} v = logger.DestinationStdout
case "file": case "file":
(*d)[logger.DestinationFile] = struct{}{} v = logger.DestinationFile
case "syslog": case "syslog":
(*d)[logger.DestinationSyslog] = struct{}{} v = logger.DestinationSyslog
default: default:
return fmt.Errorf("invalid log destination: %s", proto) return fmt.Errorf("invalid log destination: %s", dest)
}
if d.contains(v) {
return fmt.Errorf("log destination set twice")
} }
*d = append(*d, v)
} }
return nil return nil

8
internal/core/api.go
Unescape View File

@ -98,7 +98,7 @@ type apiRTMPServer interface {
} }
type apiParent interface { type apiParent interface {
Log(logger.Level, string, ...interface{}) logger.Writer
apiConfigSet(conf *conf.Conf) apiConfigSet(conf *conf.Conf)
} }
@ -208,18 +208,18 @@ func newAPI(
go a.httpServer.Serve(ln) go a.httpServer.Serve(ln)
a.log(logger.Info, "listener opened on "+address) a.Log(logger.Info, "listener opened on "+address)
return a, nil return a, nil
} }
func (a *api) close() { func (a *api) close() {
a.log(logger.Info, "listener is closing") a.Log(logger.Info, "listener is closing")
a.httpServer.Shutdown(context.Background()) a.httpServer.Shutdown(context.Background())
a.ln.Close() // in case Shutdown() is called before Serve() a.ln.Close() // in case Shutdown() is called before Serve()
} }
func (a *api) log(level logger.Level, format string, args ...interface{}) { func (a *api) Log(level logger.Level, format string, args ...interface{}) {
a.parent.Log(level, "[API] "+format, args...) a.parent.Log(level, "[API] "+format, args...)
} }

16
internal/core/hls_muxer.go
Unescape View File

@ -60,7 +60,7 @@ type hlsMuxerPathManager interface {
} }
type hlsMuxerParent interface { type hlsMuxerParent interface {
log(logger.Level, string, ...interface{}) logger.Writer
muxerClose(*hlsMuxer) muxerClose(*hlsMuxer)
} }
@ -141,7 +141,7 @@ func newHLSMuxer(
chAPIHLSMuxersList: make(chan hlsServerAPIMuxersListSubReq), chAPIHLSMuxersList: make(chan hlsServerAPIMuxersListSubReq),
} }
m.log(logger.Info, "created %s", func() string { m.Log(logger.Info, "created %s", func() string {
if remoteAddr == "" { if remoteAddr == "" {
return "automatically" return "automatically"
} }
@ -158,8 +158,8 @@ func (m *hlsMuxer) close() {
m.ctxCancel() m.ctxCancel()
} }
func (m *hlsMuxer) log(level logger.Level, format string, args ...interface{}) { func (m *hlsMuxer) Log(level logger.Level, format string, args ...interface{}) {
m.parent.log(level, "[muxer %s] "+format, append([]interface{}{m.pathName}, args...)...) m.parent.Log(level, "[muxer %s] "+format, append([]interface{}{m.pathName}, args...)...)
} }
// PathName returns the path name. // PathName returns the path name.
@ -231,7 +231,7 @@ func (m *hlsMuxer) run() {
innerCtxCancel() innerCtxCancel()
if m.alwaysRemux { if m.alwaysRemux {
m.log(logger.Info, "ERR: %v", err) m.Log(logger.Info, "ERR: %v", err)
m.clearQueuedRequests() m.clearQueuedRequests()
isReady = false isReady = false
isRecreating = true isRecreating = true
@ -253,7 +253,7 @@ func (m *hlsMuxer) run() {
m.parent.muxerClose(m) m.parent.muxerClose(m)
m.log(logger.Info, "destroyed (%v)", err) m.Log(logger.Info, "destroyed (%v)", err)
} }
func (m *hlsMuxer) clearQueuedRequests() { func (m *hlsMuxer) clearQueuedRequests() {
@ -325,7 +325,7 @@ func (m *hlsMuxer) runInner(innerCtx context.Context, innerReady chan struct{})
innerReady <- struct{}{} innerReady <- struct{}{}
m.log(logger.Info, "is converting into HLS, %s", m.Log(logger.Info, "is converting into HLS, %s",
sourceMediaInfo(medias)) sourceMediaInfo(medias))
writerDone := make(chan error) writerDone := make(chan error)
@ -557,7 +557,7 @@ func (m *hlsMuxer) handleRequest(ctx *gin.Context) {
err := m.authenticate(ctx) err := m.authenticate(ctx)
if err != nil { if err != nil {
if terr, ok := err.(pathErrAuthCritical); ok { if terr, ok := err.(pathErrAuthCritical); ok {
m.log(logger.Info, "authentication error: %s", terr.message) m.Log(logger.Info, "authentication error: %s", terr.message)
} }
ctx.Header("WWW-Authenticate", `Basic realm="mediamtx"`) ctx.Header("WWW-Authenticate", `Basic realm="mediamtx"`)

8
internal/core/hls_server.go
Unescape View File

@ -50,7 +50,7 @@ type hlsServerAPIMuxersListSubReq struct {
} }
type hlsServerParent interface { type hlsServerParent interface {
Log(logger.Level, string, ...interface{}) logger.Writer
} }
type hlsServer struct { type hlsServer struct {
@ -162,7 +162,7 @@ func newHLSServer(
ErrorLog: log.New(&nilWriter{}, "", 0), ErrorLog: log.New(&nilWriter{}, "", 0),
} }
s.log(logger.Info, "listener opened on "+address) s.Log(logger.Info, "listener opened on "+address)
s.pathManager.hlsServerSet(s) s.pathManager.hlsServerSet(s)
@ -177,12 +177,12 @@ func newHLSServer(
} }
// Log is the main logging function. // Log is the main logging function.
func (s *hlsServer) log(level logger.Level, format string, args ...interface{}) { func (s *hlsServer) Log(level logger.Level, format string, args ...interface{}) {
s.parent.Log(level, "[HLS] "+format, append([]interface{}{}, args...)...) s.parent.Log(level, "[HLS] "+format, append([]interface{}{}, args...)...)
} }
func (s *hlsServer) close() { func (s *hlsServer) close() {
s.log(logger.Info, "listener is closing") s.Log(logger.Info, "listener is closing")
s.ctxCancel() s.ctxCancel()
s.wg.Wait() s.wg.Wait()
} }

24
internal/core/hls_source.go
Unescape View File

@ -21,7 +21,7 @@ import (
) )
type hlsSourceParent interface { type hlsSourceParent interface {
log(logger.Level, string, ...interface{}) logger.Writer
sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes
sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq) sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq)
} }
@ -39,7 +39,7 @@ func newHLSSource(
} }
func (s *hlsSource) Log(level logger.Level, format string, args ...interface{}) { func (s *hlsSource) Log(level logger.Level, format string, args ...interface{}) {
s.parent.log(level, "[hls source] "+format, args...) s.parent.Log(level, "[hls source] "+format, args...)
} }
// run implements sourceStaticImpl. // run implements sourceStaticImpl.
@ -103,14 +103,11 @@ func (s *hlsSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
} }
c.OnData(track, func(pts time.Duration, unit interface{}) { c.OnData(track, func(pts time.Duration, unit interface{}) {
err := stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH264{ stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH264{
PTS: pts, PTS: pts,
AU: unit.([][]byte), AU: unit.([][]byte),
NTP: time.Now(), NTP: time.Now(),
}) })
if err != nil {
s.Log(logger.Warn, "%v", err)
}
}) })
case *codecs.H265: case *codecs.H265:
@ -125,14 +122,11 @@ func (s *hlsSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
} }
c.OnData(track, func(pts time.Duration, unit interface{}) { c.OnData(track, func(pts time.Duration, unit interface{}) {
err := stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH265{ stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH265{
PTS: pts, PTS: pts,
AU: unit.([][]byte), AU: unit.([][]byte),
NTP: time.Now(), NTP: time.Now(),
}) })
if err != nil {
s.Log(logger.Warn, "%v", err)
}
}) })
case *codecs.MPEG4Audio: case *codecs.MPEG4Audio:
@ -148,14 +142,11 @@ func (s *hlsSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
} }
c.OnData(track, func(pts time.Duration, unit interface{}) { c.OnData(track, func(pts time.Duration, unit interface{}) {
err := stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitMPEG4Audio{ stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitMPEG4Audio{
PTS: pts, PTS: pts,
AUs: [][]byte{unit.([]byte)}, AUs: [][]byte{unit.([]byte)},
NTP: time.Now(), NTP: time.Now(),
}) })
if err != nil {
s.Log(logger.Warn, "%v", err)
}
}) })
case *codecs.Opus: case *codecs.Opus:
@ -168,14 +159,11 @@ func (s *hlsSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
} }
c.OnData(track, func(pts time.Duration, unit interface{}) { c.OnData(track, func(pts time.Duration, unit interface{}) {
err := stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitOpus{ stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitOpus{
PTS: pts, PTS: pts,
Frame: unit.([]byte), Frame: unit.([]byte),
NTP: time.Now(), NTP: time.Now(),
}) })
if err != nil {
s.Log(logger.Warn, "%v", err)
}
}) })
} }

8
internal/core/http_logger.go
Unescape View File

@ -38,21 +38,21 @@ func (w *httpLoggerWriter) dump() string {
} }
type httpLoggerParent interface { type httpLoggerParent interface {
log(logger.Level, string, ...interface{}) logger.Writer
} }
func httpLoggerMiddleware(p httpLoggerParent) func(*gin.Context) { func httpLoggerMiddleware(p httpLoggerParent) func(*gin.Context) {
return func(ctx *gin.Context) { return func(ctx *gin.Context) {
p.log(logger.Debug, "[conn %v] %s %s", ctx.Request.RemoteAddr, ctx.Request.Method, ctx.Request.URL.Path) p.Log(logger.Debug, "[conn %v] %s %s", ctx.Request.RemoteAddr, ctx.Request.Method, ctx.Request.URL.Path)
byts, _ := httputil.DumpRequest(ctx.Request, true) byts, _ := httputil.DumpRequest(ctx.Request, true)
p.log(logger.Debug, "[conn %v] [c->s] %s", ctx.Request.RemoteAddr, string(byts)) p.Log(logger.Debug, "[conn %v] [c->s] %s", ctx.Request.RemoteAddr, string(byts))
logw := &httpLoggerWriter{ResponseWriter: ctx.Writer} logw := &httpLoggerWriter{ResponseWriter: ctx.Writer}
ctx.Writer = logw ctx.Writer = logw
ctx.Next() ctx.Next()
p.log(logger.Debug, "[conn %v] [s->c] %s", ctx.Request.RemoteAddr, logw.dump()) p.Log(logger.Debug, "[conn %v] [s->c] %s", ctx.Request.RemoteAddr, logw.dump())
} }
} }

8
internal/core/metrics.go
Unescape View File

@ -21,7 +21,7 @@ func metric(key string, tags string, value int64) string {
} }
type metricsParent interface { type metricsParent interface {
Log(logger.Level, string, ...interface{}) logger.Writer
} }
type metrics struct { type metrics struct {
@ -66,7 +66,7 @@ func newMetrics(
ErrorLog: log.New(&nilWriter{}, "", 0), ErrorLog: log.New(&nilWriter{}, "", 0),
} }
m.log(logger.Info, "listener opened on "+address) m.Log(logger.Info, "listener opened on "+address)
go m.httpServer.Serve(m.ln) go m.httpServer.Serve(m.ln)
@ -74,12 +74,12 @@ func newMetrics(
} }
func (m *metrics) close() { func (m *metrics) close() {
m.log(logger.Info, "listener is closing") m.Log(logger.Info, "listener is closing")
m.httpServer.Shutdown(context.Background()) m.httpServer.Shutdown(context.Background())
m.ln.Close() // in case Shutdown() is called before Serve() m.ln.Close() // in case Shutdown() is called before Serve()
} }
func (m *metrics) log(level logger.Level, format string, args ...interface{}) { func (m *metrics) Log(level logger.Level, format string, args ...interface{}) {
m.parent.Log(level, "[metrics] "+format, args...) m.parent.Log(level, "[metrics] "+format, args...)
} }

33
internal/core/path.go
Unescape View File

@ -61,7 +61,7 @@ func (pathErrAuthCritical) Error() string {
} }
type pathParent interface { type pathParent interface {
log(logger.Level, string, ...interface{}) logger.Writer
pathSourceReady(*path) pathSourceReady(*path)
pathSourceNotReady(*path) pathSourceNotReady(*path)
onPathClose(*path) onPathClose(*path)
@ -287,7 +287,7 @@ func newPath(
done: make(chan struct{}), done: make(chan struct{}),
} }
pa.log(logger.Debug, "created") pa.Log(logger.Debug, "created")
pa.wg.Add(1) pa.wg.Add(1)
go pa.run() go pa.run()
@ -304,8 +304,8 @@ func (pa *path) wait() {
} }
// Log is the main logging function. // Log is the main logging function.
func (pa *path) log(level logger.Level, format string, args ...interface{}) { func (pa *path) Log(level logger.Level, format string, args ...interface{}) {
pa.parent.log(level, "[path "+pa.name+"] "+format, args...) pa.parent.Log(level, "[path "+pa.name+"] "+format, args...)
} }
func (pa *path) safeConf() *conf.PathConf { func (pa *path) safeConf() *conf.PathConf {
@ -335,14 +335,14 @@ func (pa *path) run() {
var onInitCmd *externalcmd.Cmd var onInitCmd *externalcmd.Cmd
if pa.conf.RunOnInit != "" { if pa.conf.RunOnInit != "" {
pa.log(logger.Info, "runOnInit command started") pa.Log(logger.Info, "runOnInit command started")
onInitCmd = externalcmd.NewCmd( onInitCmd = externalcmd.NewCmd(
pa.externalCmdPool, pa.externalCmdPool,
pa.conf.RunOnInit, pa.conf.RunOnInit,
pa.conf.RunOnInitRestart, pa.conf.RunOnInitRestart,
pa.externalCmdEnv(), pa.externalCmdEnv(),
func(co int) { func(co int) {
pa.log(logger.Info, "runOnInit command exited with code %d", co) pa.Log(logger.Info, "runOnInit command exited with code %d", co)
}) })
} }
@ -507,7 +507,7 @@ func (pa *path) run() {
if onInitCmd != nil { if onInitCmd != nil {
onInitCmd.Close() onInitCmd.Close()
pa.log(logger.Info, "runOnInit command stopped") pa.Log(logger.Info, "runOnInit command stopped")
} }
for _, req := range pa.describeRequestsOnHold { for _, req := range pa.describeRequestsOnHold {
@ -532,10 +532,10 @@ func (pa *path) run() {
if pa.onDemandCmd != nil { if pa.onDemandCmd != nil {
pa.onDemandCmd.Close() pa.onDemandCmd.Close()
pa.log(logger.Info, "runOnDemand command stopped") pa.Log(logger.Info, "runOnDemand command stopped")
} }
pa.log(logger.Debug, "destroyed (%v)", err) pa.Log(logger.Debug, "destroyed (%v)", err)
} }
func (pa *path) shouldClose() bool { func (pa *path) shouldClose() bool {
@ -590,14 +590,14 @@ func (pa *path) onDemandStaticSourceStop() {
} }
func (pa *path) onDemandPublisherStart() { func (pa *path) onDemandPublisherStart() {
pa.log(logger.Info, "runOnDemand command started") pa.Log(logger.Info, "runOnDemand command started")
pa.onDemandCmd = externalcmd.NewCmd( pa.onDemandCmd = externalcmd.NewCmd(
pa.externalCmdPool, pa.externalCmdPool,
pa.conf.RunOnDemand, pa.conf.RunOnDemand,
pa.conf.RunOnDemandRestart, pa.conf.RunOnDemandRestart,
pa.externalCmdEnv(), pa.externalCmdEnv(),
func(co int) { func(co int) {
pa.log(logger.Info, "runOnDemand command exited with code %d", co) pa.Log(logger.Info, "runOnDemand command exited with code %d", co)
}) })
pa.onDemandPublisherReadyTimer.Stop() pa.onDemandPublisherReadyTimer.Stop()
@ -630,7 +630,7 @@ func (pa *path) onDemandPublisherStop() {
if pa.onDemandCmd != nil { if pa.onDemandCmd != nil {
pa.onDemandCmd.Close() pa.onDemandCmd.Close()
pa.onDemandCmd = nil pa.onDemandCmd = nil
pa.log(logger.Info, "runOnDemand command stopped") pa.Log(logger.Info, "runOnDemand command stopped")
} }
} }
@ -640,6 +640,7 @@ func (pa *path) sourceSetReady(medias media.Medias, allocateEncoder bool) error
medias, medias,
allocateEncoder, allocateEncoder,
pa.bytesReceived, pa.bytesReceived,
pa.source,
) )
if err != nil { if err != nil {
return err return err
@ -648,14 +649,14 @@ func (pa *path) sourceSetReady(medias media.Medias, allocateEncoder bool) error
pa.stream = stream pa.stream = stream
if pa.conf.RunOnReady != "" { if pa.conf.RunOnReady != "" {
pa.log(logger.Info, "runOnReady command started") pa.Log(logger.Info, "runOnReady command started")
pa.onReadyCmd = externalcmd.NewCmd( pa.onReadyCmd = externalcmd.NewCmd(
pa.externalCmdPool, pa.externalCmdPool,
pa.conf.RunOnReady, pa.conf.RunOnReady,
pa.conf.RunOnReadyRestart, pa.conf.RunOnReadyRestart,
pa.externalCmdEnv(), pa.externalCmdEnv(),
func(co int) { func(co int) {
pa.log(logger.Info, "runOnReady command exited with code %d", co) pa.Log(logger.Info, "runOnReady command exited with code %d", co)
}) })
} }
@ -675,7 +676,7 @@ func (pa *path) sourceSetNotReady() {
if pa.onReadyCmd != nil { if pa.onReadyCmd != nil {
pa.onReadyCmd.Close() pa.onReadyCmd.Close()
pa.onReadyCmd = nil pa.onReadyCmd = nil
pa.log(logger.Info, "runOnReady command stopped") pa.Log(logger.Info, "runOnReady command stopped")
} }
if pa.stream != nil { if pa.stream != nil {
@ -772,7 +773,7 @@ func (pa *path) handlePublisherAdd(req pathPublisherAddReq) {
return return
} }
pa.log(logger.Info, "closing existing publisher") pa.Log(logger.Info, "closing existing publisher")
pa.source.(publisher).close() pa.source.(publisher).close()
pa.doPublisherRemove() pa.doPublisherRemove()
} }

8
internal/core/path_manager.go
Unescape View File

@ -35,7 +35,7 @@ type pathManagerHLSServer interface {
} }
type pathManagerParent interface { type pathManagerParent interface {
Log(logger.Level, string, ...interface{}) logger.Writer
} }
type pathManager struct { type pathManager struct {
@ -117,7 +117,7 @@ func newPathManager(
pm.metrics.pathManagerSet(pm) pm.metrics.pathManagerSet(pm)
} }
pm.log(logger.Debug, "path manager created") pm.Log(logger.Debug, "path manager created")
pm.wg.Add(1) pm.wg.Add(1)
go pm.run() go pm.run()
@ -126,13 +126,13 @@ func newPathManager(
} }
func (pm *pathManager) close() { func (pm *pathManager) close() {
pm.log(logger.Debug, "path manager is shutting down") pm.Log(logger.Debug, "path manager is shutting down")
pm.ctxCancel() pm.ctxCancel()
pm.wg.Wait() pm.wg.Wait()
} }
// Log is the main logging function. // Log is the main logging function.
func (pm *pathManager) log(level logger.Level, format string, args ...interface{}) { func (pm *pathManager) Log(level logger.Level, format string, args ...interface{}) {
pm.parent.Log(level, format, args...) pm.parent.Log(level, format, args...)
} }

8
internal/core/pprof.go
Unescape View File

@ -15,7 +15,7 @@ import (
) )
type pprofParent interface { type pprofParent interface {
Log(logger.Level, string, ...interface{}) logger.Writer
} }
type pprof struct { type pprof struct {
@ -46,7 +46,7 @@ func newPPROF(
ErrorLog: log.New(&nilWriter{}, "", 0), ErrorLog: log.New(&nilWriter{}, "", 0),
} }
pp.log(logger.Info, "listener opened on "+address) pp.Log(logger.Info, "listener opened on "+address)
go pp.httpServer.Serve(pp.ln) go pp.httpServer.Serve(pp.ln)
@ -54,11 +54,11 @@ func newPPROF(
} }
func (pp *pprof) close() { func (pp *pprof) close() {
pp.log(logger.Info, "listener is closing") pp.Log(logger.Info, "listener is closing")
pp.httpServer.Shutdown(context.Background()) pp.httpServer.Shutdown(context.Background())
pp.ln.Close() // in case Shutdown() is called before Serve() pp.ln.Close() // in case Shutdown() is called before Serve()
} }
func (pp *pprof) log(level logger.Level, format string, args ...interface{}) { func (pp *pprof) Log(level logger.Level, format string, args ...interface{}) {
pp.parent.Log(level, "[pprof] "+format, args...) pp.parent.Log(level, "[pprof] "+format, args...)
} }

9
internal/core/rpicamera_source.go
Unescape View File

@ -50,7 +50,7 @@ func paramsFromConf(cnf *conf.PathConf) rpicamera.Params {
} }
type rpiCameraSourceParent interface { type rpiCameraSourceParent interface {
log(logger.Level, string, ...interface{}) logger.Writer
sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes
sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq) sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq)
} }
@ -68,7 +68,7 @@ func newRPICameraSource(
} }
func (s *rpiCameraSource) Log(level logger.Level, format string, args ...interface{}) { func (s *rpiCameraSource) Log(level logger.Level, format string, args ...interface{}) {
s.parent.log(level, "[rpicamera source] "+format, args...) s.parent.Log(level, "[rpicamera source] "+format, args...)
} }
// run implements sourceStaticImpl. // run implements sourceStaticImpl.
@ -97,14 +97,11 @@ func (s *rpiCameraSource) run(ctx context.Context, cnf *conf.PathConf, reloadCon
stream = res.stream stream = res.stream
} }
err := stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH264{ stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH264{
PTS: dts, PTS: dts,
AU: au, AU: au,
NTP: time.Now(), NTP: time.Now(),
}) })
if err != nil {
s.Log(logger.Warn, "%v", err)
}
} }
cam, err := rpicamera.New(paramsFromConf(cnf), onData) cam, err := rpicamera.New(paramsFromConf(cnf), onData)

49
internal/core/rtmp_conn.go
Unescape View File

@ -61,7 +61,7 @@ func getRTMPWriteFunc(medi *media.Media, format formats.Format, stream *stream)
conf.PPS, conf.PPS,
} }
return stream.writeUnit(medi, format, &formatprocessor.UnitH264{ stream.writeUnit(medi, format, &formatprocessor.UnitH264{
PTS: tmsg.DTS + tmsg.PTSDelta, PTS: tmsg.DTS + tmsg.PTSDelta,
AU: au, AU: au,
NTP: time.Now(), NTP: time.Now(),
@ -73,7 +73,7 @@ func getRTMPWriteFunc(medi *media.Media, format formats.Format, stream *stream)
return fmt.Errorf("unable to decode AVCC: %v", err) return fmt.Errorf("unable to decode AVCC: %v", err)
} }
return stream.writeUnit(medi, format, &formatprocessor.UnitH264{ stream.writeUnit(medi, format, &formatprocessor.UnitH264{
PTS: tmsg.DTS + tmsg.PTSDelta, PTS: tmsg.DTS + tmsg.PTSDelta,
AU: au, AU: au,
NTP: time.Now(), NTP: time.Now(),
@ -92,22 +92,26 @@ func getRTMPWriteFunc(medi *media.Media, format formats.Format, stream *stream)
return fmt.Errorf("unable to decode AVCC: %v", err) return fmt.Errorf("unable to decode AVCC: %v", err)
} }
return stream.writeUnit(medi, format, &formatprocessor.UnitH265{ stream.writeUnit(medi, format, &formatprocessor.UnitH265{
PTS: tmsg.DTS + tmsg.PTSDelta, PTS: tmsg.DTS + tmsg.PTSDelta,
AU: au, AU: au,
NTP: time.Now(), NTP: time.Now(),
}) })
return nil
} }
case *formats.MPEG2Audio: case *formats.MPEG2Audio:
return func(msg interface{}) error { return func(msg interface{}) error {
tmsg := msg.(*message.MsgAudio) tmsg := msg.(*message.MsgAudio)
return stream.writeUnit(medi, format, &formatprocessor.UnitMPEG2Audio{ stream.writeUnit(medi, format, &formatprocessor.UnitMPEG2Audio{
PTS: tmsg.DTS, PTS: tmsg.DTS,
Frames: [][]byte{tmsg.Payload}, Frames: [][]byte{tmsg.Payload},
NTP: time.Now(), NTP: time.Now(),
}) })
return nil
} }
case *formats.MPEG4Audio: case *formats.MPEG4Audio:
@ -115,7 +119,7 @@ func getRTMPWriteFunc(medi *media.Media, format formats.Format, stream *stream)
tmsg := msg.(*message.MsgAudio) tmsg := msg.(*message.MsgAudio)
if tmsg.AACType == message.MsgAudioAACTypeAU { if tmsg.AACType == message.MsgAudioAACTypeAU {
return stream.writeUnit(medi, format, &formatprocessor.UnitMPEG4Audio{ stream.writeUnit(medi, format, &formatprocessor.UnitMPEG4Audio{
PTS: tmsg.DTS, PTS: tmsg.DTS,
AUs: [][]byte{tmsg.Payload}, AUs: [][]byte{tmsg.Payload},
NTP: time.Now(), NTP: time.Now(),
@ -124,10 +128,9 @@ func getRTMPWriteFunc(medi *media.Media, format formats.Format, stream *stream)
return nil return nil
} }
default:
return nil
} }
return nil
} }
type rtmpConnState int type rtmpConnState int
@ -144,7 +147,7 @@ type rtmpConnPathManager interface {
} }
type rtmpConnParent interface { type rtmpConnParent interface {
log(logger.Level, string, ...interface{}) logger.Writer
connClose(*rtmpConn) connClose(*rtmpConn)
} }
@ -211,7 +214,7 @@ func newRTMPConn(
created: time.Now(), created: time.Now(),
} }
c.log(logger.Info, "opened") c.Log(logger.Info, "opened")
c.wg.Add(1) c.wg.Add(1)
go c.run() go c.run()
@ -227,8 +230,8 @@ func (c *rtmpConn) remoteAddr() net.Addr {
return c.nconn.RemoteAddr() return c.nconn.RemoteAddr()
} }
func (c *rtmpConn) log(level logger.Level, format string, args ...interface{}) { func (c *rtmpConn) Log(level logger.Level, format string, args ...interface{}) {
c.parent.log(level, "[conn %v] "+format, append([]interface{}{c.nconn.RemoteAddr()}, args...)...) c.parent.Log(level, "[conn %v] "+format, append([]interface{}{c.nconn.RemoteAddr()}, args...)...)
} }
func (c *rtmpConn) ip() net.IP { func (c *rtmpConn) ip() net.IP {
@ -245,7 +248,7 @@ func (c *rtmpConn) run() {
defer c.wg.Done() defer c.wg.Done()
if c.runOnConnect != "" { if c.runOnConnect != "" {
c.log(logger.Info, "runOnConnect command started") c.Log(logger.Info, "runOnConnect command started")
_, port, _ := net.SplitHostPort(c.rtspAddress) _, port, _ := net.SplitHostPort(c.rtspAddress)
onConnectCmd := externalcmd.NewCmd( onConnectCmd := externalcmd.NewCmd(
c.externalCmdPool, c.externalCmdPool,
@ -256,12 +259,12 @@ func (c *rtmpConn) run() {
"RTSP_PORT": port, "RTSP_PORT": port,
}, },
func(co int) { func(co int) {
c.log(logger.Info, "runOnConnect command exited with code %d", co) c.Log(logger.Info, "runOnConnect command exited with code %d", co)
}) })
defer func() { defer func() {
onConnectCmd.Close() onConnectCmd.Close()
c.log(logger.Info, "runOnConnect command stopped") c.Log(logger.Info, "runOnConnect command stopped")
}() }()
} }
@ -286,7 +289,7 @@ func (c *rtmpConn) run() {
c.parent.connClose(c) c.parent.connClose(c)
c.log(logger.Info, "closed (%v)", err) c.Log(logger.Info, "closed (%v)", err)
} }
func (c *rtmpConn) runInner(ctx context.Context) error { func (c *rtmpConn) runInner(ctx context.Context) error {
@ -371,24 +374,24 @@ func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error {
defer res.stream.readerRemove(c) defer res.stream.readerRemove(c)
c.log(logger.Info, "is reading from path '%s', %s", c.Log(logger.Info, "is reading from path '%s', %s",
path.name, sourceMediaInfo(medias)) path.name, sourceMediaInfo(medias))
pathConf := path.safeConf() pathConf := path.safeConf()
if pathConf.RunOnRead != "" { if pathConf.RunOnRead != "" {
c.log(logger.Info, "runOnRead command started") c.Log(logger.Info, "runOnRead command started")
onReadCmd := externalcmd.NewCmd( onReadCmd := externalcmd.NewCmd(
c.externalCmdPool, c.externalCmdPool,
pathConf.RunOnRead, pathConf.RunOnRead,
pathConf.RunOnReadRestart, pathConf.RunOnReadRestart,
path.externalCmdEnv(), path.externalCmdEnv(),
func(co int) { func(co int) {
c.log(logger.Info, "runOnRead command exited with code %d", co) c.Log(logger.Info, "runOnRead command exited with code %d", co)
}) })
defer func() { defer func() {
onReadCmd.Close() onReadCmd.Close()
c.log(logger.Info, "runOnRead command stopped") c.Log(logger.Info, "runOnRead command stopped")
}() }()
} }
@ -731,7 +734,7 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error {
return rres.err return rres.err
} }
c.log(logger.Info, "is publishing to path '%s', %s", c.Log(logger.Info, "is publishing to path '%s', %s",
path.name, path.name,
sourceMediaInfo(medias)) sourceMediaInfo(medias))
@ -756,7 +759,7 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error {
err := videoWriteFunc(tmsg) err := videoWriteFunc(tmsg)
if err != nil { if err != nil {
c.log(logger.Warn, "%v", err) c.Log(logger.Warn, "%v", err)
} }
case *message.MsgAudio: case *message.MsgAudio:
@ -766,7 +769,7 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error {
err := audioWriteFunc(tmsg) err := audioWriteFunc(tmsg)
if err != nil { if err != nil {
c.log(logger.Warn, "%v", err) c.Log(logger.Warn, "%v", err)
} }
} }
} }

10
internal/core/rtmp_server.go
Unescape View File

@ -44,7 +44,7 @@ type rtmpServerAPIConnsKickReq struct {
} }
type rtmpServerParent interface { type rtmpServerParent interface {
Log(logger.Level, string, ...interface{}) logger.Writer
} }
type rtmpServer struct { type rtmpServer struct {
@ -132,7 +132,7 @@ func newRTMPServer(
chAPIConnsKick: make(chan rtmpServerAPIConnsKickReq), chAPIConnsKick: make(chan rtmpServerAPIConnsKickReq),
} }
s.log(logger.Info, "listener opened on %s", address) s.Log(logger.Info, "listener opened on %s", address)
if s.metrics != nil { if s.metrics != nil {
s.metrics.rtmpServerSet(s) s.metrics.rtmpServerSet(s)
@ -144,7 +144,7 @@ func newRTMPServer(
return s, nil return s, nil
} }
func (s *rtmpServer) log(level logger.Level, format string, args ...interface{}) { func (s *rtmpServer) Log(level logger.Level, format string, args ...interface{}) {
label := func() string { label := func() string {
if s.isTLS { if s.isTLS {
return "RTMPS" return "RTMPS"
@ -155,7 +155,7 @@ func (s *rtmpServer) log(level logger.Level, format string, args ...interface{})
} }
func (s *rtmpServer) close() { func (s *rtmpServer) close() {
s.log(logger.Info, "listener is closing") s.Log(logger.Info, "listener is closing")
s.ctxCancel() s.ctxCancel()
s.wg.Wait() s.wg.Wait()
} }
@ -193,7 +193,7 @@ outer:
for { for {
select { select {
case err := <-acceptErr: case err := <-acceptErr:
s.log(logger.Error, "%s", err) s.Log(logger.Error, "%s", err)
break outer break outer
case nconn := <-connNew: case nconn := <-connNew:

4
internal/core/rtmp_source.go
Unescape View File

@ -21,7 +21,7 @@ import (
) )
type rtmpSourceParent interface { type rtmpSourceParent interface {
log(logger.Level, string, ...interface{}) logger.Writer
sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes
sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq) sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq)
} }
@ -45,7 +45,7 @@ func newRTMPSource(
} }
func (s *rtmpSource) Log(level logger.Level, format string, args ...interface{}) { func (s *rtmpSource) Log(level logger.Level, format string, args ...interface{}) {
s.parent.log(level, "[rtmp source] "+format, args...) s.parent.Log(level, "[rtmp source] "+format, args...)
} }
// run implements sourceStaticImpl. // run implements sourceStaticImpl.

22
internal/core/rtsp_conn.go
Unescape View File

@ -23,7 +23,7 @@ const (
) )
type rtspConnParent interface { type rtspConnParent interface {
log(logger.Level, string, ...interface{}) logger.Writer
} }
type rtspConn struct { type rtspConn struct {
@ -74,10 +74,10 @@ func newRTSPConn(
created: time.Now(), created: time.Now(),
} }
c.log(logger.Info, "opened") c.Log(logger.Info, "opened")
if c.runOnConnect != "" { if c.runOnConnect != "" {
c.log(logger.Info, "runOnConnect command started") c.Log(logger.Info, "runOnConnect command started")
_, port, _ := net.SplitHostPort(c.rtspAddress) _, port, _ := net.SplitHostPort(c.rtspAddress)
c.onConnectCmd = externalcmd.NewCmd( c.onConnectCmd = externalcmd.NewCmd(
c.externalCmdPool, c.externalCmdPool,
@ -88,15 +88,15 @@ func newRTSPConn(
"RTSP_PORT": port, "RTSP_PORT": port,
}, },
func(co int) { func(co int) {
c.log(logger.Info, "runOnInit command exited with code %d", co) c.Log(logger.Info, "runOnInit command exited with code %d", co)
}) })
} }
return c return c
} }
func (c *rtspConn) log(level logger.Level, format string, args ...interface{}) { func (c *rtspConn) Log(level logger.Level, format string, args ...interface{}) {
c.parent.log(level, "[conn %v] "+format, append([]interface{}{c.conn.NetConn().RemoteAddr()}, args...)...) c.parent.Log(level, "[conn %v] "+format, append([]interface{}{c.conn.NetConn().RemoteAddr()}, args...)...)
} }
// Conn returns the RTSP connection. // Conn returns the RTSP connection.
@ -235,22 +235,22 @@ func (c *rtspConn) authenticate(
// onClose is called by rtspServer. // onClose is called by rtspServer.
func (c *rtspConn) onClose(err error) { func (c *rtspConn) onClose(err error) {
c.log(logger.Info, "closed (%v)", err) c.Log(logger.Info, "closed (%v)", err)
if c.onConnectCmd != nil { if c.onConnectCmd != nil {
c.onConnectCmd.Close() c.onConnectCmd.Close()
c.log(logger.Info, "runOnConnect command stopped") c.Log(logger.Info, "runOnConnect command stopped")
} }
} }
// onRequest is called by rtspServer. // onRequest is called by rtspServer.
func (c *rtspConn) onRequest(req *base.Request) { func (c *rtspConn) onRequest(req *base.Request) {
c.log(logger.Debug, "[c->s] %v", req) c.Log(logger.Debug, "[c->s] %v", req)
} }
// OnResponse is called by rtspServer. // OnResponse is called by rtspServer.
func (c *rtspConn) OnResponse(res *base.Response) { func (c *rtspConn) OnResponse(res *base.Response) {
c.log(logger.Debug, "[s->c] %v", res) c.Log(logger.Debug, "[s->c] %v", res)
} }
// onDescribe is called by rtspServer. // onDescribe is called by rtspServer.
@ -278,7 +278,7 @@ func (c *rtspConn) onDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx,
if res.err != nil { if res.err != nil {
switch terr := res.err.(type) { switch terr := res.err.(type) {
case pathErrAuthNotCritical: case pathErrAuthNotCritical:
c.log(logger.Debug, "non-critical authentication error: %s", terr.message) c.Log(logger.Debug, "non-critical authentication error: %s", terr.message)
return terr.response, nil, nil return terr.response, nil, nil
case pathErrAuthCritical: case pathErrAuthCritical:

10
internal/core/rtsp_server.go
Unescape View File

@ -56,7 +56,7 @@ type rtspServerAPISessionsKickRes struct {
} }
type rtspServerParent interface { type rtspServerParent interface {
Log(logger.Level, string, ...interface{}) logger.Writer
} }
func printAddresses(srv *gortsplib.Server) string { func printAddresses(srv *gortsplib.Server) string {
@ -180,7 +180,7 @@ func newRTSPServer(
return nil, err return nil, err
} }
s.log(logger.Info, "listener opened on %s", printAddresses(s.srv)) s.Log(logger.Info, "listener opened on %s", printAddresses(s.srv))
if metrics != nil { if metrics != nil {
if !isTLS { if !isTLS {
@ -196,7 +196,7 @@ func newRTSPServer(
return s, nil return s, nil
} }
func (s *rtspServer) log(level logger.Level, format string, args ...interface{}) { func (s *rtspServer) Log(level logger.Level, format string, args ...interface{}) {
label := func() string { label := func() string {
if s.isTLS { if s.isTLS {
return "RTSPS" return "RTSPS"
@ -207,7 +207,7 @@ func (s *rtspServer) log(level logger.Level, format string, args ...interface{})
} }
func (s *rtspServer) close() { func (s *rtspServer) close() {
s.log(logger.Info, "listener is closing") s.Log(logger.Info, "listener is closing")
s.ctxCancel() s.ctxCancel()
s.wg.Wait() s.wg.Wait()
} }
@ -223,7 +223,7 @@ func (s *rtspServer) run() {
outer: outer:
select { select {
case err := <-serverErr: case err := <-serverErr:
s.log(logger.Error, "%s", err) s.Log(logger.Error, "%s", err)
break outer break outer
case <-s.ctx.Done(): case <-s.ctx.Done():

69
internal/core/rtsp_session.go
Unescape View File

@ -26,69 +26,69 @@ const (
pauseAfterAuthError = 2 * time.Second pauseAfterAuthError = 2 * time.Second
) )
type rtspWriteFunc func(*rtp.Packet) error type rtspWriteFunc func(*rtp.Packet)
func getRTSPWriteFunc(medi *media.Media, forma formats.Format, stream *stream) rtspWriteFunc { func getRTSPWriteFunc(medi *media.Media, forma formats.Format, stream *stream) rtspWriteFunc {
switch forma.(type) { switch forma.(type) {
case *formats.H264: case *formats.H264:
return func(pkt *rtp.Packet) error { return func(pkt *rtp.Packet) {
return stream.writeUnit(medi, forma, &formatprocessor.UnitH264{ stream.writeUnit(medi, forma, &formatprocessor.UnitH264{
RTPPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
NTP: time.Now(), NTP: time.Now(),
}) })
} }
case *formats.H265: case *formats.H265:
return func(pkt *rtp.Packet) error { return func(pkt *rtp.Packet) {
return stream.writeUnit(medi, forma, &formatprocessor.UnitH265{ stream.writeUnit(medi, forma, &formatprocessor.UnitH265{
RTPPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
NTP: time.Now(), NTP: time.Now(),
}) })
} }
case *formats.VP8: case *formats.VP8:
return func(pkt *rtp.Packet) error { return func(pkt *rtp.Packet) {
return stream.writeUnit(medi, forma, &formatprocessor.UnitVP8{ stream.writeUnit(medi, forma, &formatprocessor.UnitVP8{
RTPPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
NTP: time.Now(), NTP: time.Now(),
}) })
} }
case *formats.VP9: case *formats.VP9:
return func(pkt *rtp.Packet) error { return func(pkt *rtp.Packet) {
return stream.writeUnit(medi, forma, &formatprocessor.UnitVP9{ stream.writeUnit(medi, forma, &formatprocessor.UnitVP9{
RTPPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
NTP: time.Now(), NTP: time.Now(),
}) })
} }
case *formats.MPEG2Audio: case *formats.MPEG2Audio:
return func(pkt *rtp.Packet) error { return func(pkt *rtp.Packet) {
return stream.writeUnit(medi, forma, &formatprocessor.UnitMPEG2Audio{ stream.writeUnit(medi, forma, &formatprocessor.UnitMPEG2Audio{
RTPPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
NTP: time.Now(), NTP: time.Now(),
}) })
} }
case *formats.MPEG4Audio: case *formats.MPEG4Audio:
return func(pkt *rtp.Packet) error { return func(pkt *rtp.Packet) {
return stream.writeUnit(medi, forma, &formatprocessor.UnitMPEG4Audio{ stream.writeUnit(medi, forma, &formatprocessor.UnitMPEG4Audio{
RTPPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
NTP: time.Now(), NTP: time.Now(),
}) })
} }
case *formats.Opus: case *formats.Opus:
return func(pkt *rtp.Packet) error { return func(pkt *rtp.Packet) {
return stream.writeUnit(medi, forma, &formatprocessor.UnitOpus{ stream.writeUnit(medi, forma, &formatprocessor.UnitOpus{
RTPPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
NTP: time.Now(), NTP: time.Now(),
}) })
} }
default: default:
return func(pkt *rtp.Packet) error { return func(pkt *rtp.Packet) {
return stream.writeUnit(medi, forma, &formatprocessor.UnitGeneric{ stream.writeUnit(medi, forma, &formatprocessor.UnitGeneric{
RTPPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
NTP: time.Now(), NTP: time.Now(),
}) })
@ -102,7 +102,7 @@ type rtspSessionPathManager interface {
} }
type rtspSessionParent interface { type rtspSessionParent interface {
log(logger.Level, string, ...interface{}) logger.Writer
} }
type rtspSession struct { type rtspSession struct {
@ -144,7 +144,7 @@ func newRTSPSession(
created: time.Now(), created: time.Now(),
} }
s.log(logger.Info, "created by %v", s.author.NetConn().RemoteAddr()) s.Log(logger.Info, "created by %v", s.author.NetConn().RemoteAddr())
return s return s
} }
@ -164,9 +164,9 @@ func (s *rtspSession) remoteAddr() net.Addr {
return s.author.NetConn().RemoteAddr() return s.author.NetConn().RemoteAddr()
} }
func (s *rtspSession) log(level logger.Level, format string, args ...interface{}) { func (s *rtspSession) Log(level logger.Level, format string, args ...interface{}) {
id := hex.EncodeToString(s.uuid[:4]) id := hex.EncodeToString(s.uuid[:4])
s.parent.log(level, "[session %s] "+format, append([]interface{}{id}, args...)...) s.parent.Log(level, "[session %s] "+format, append([]interface{}{id}, args...)...)
} }
// onClose is called by rtspServer. // onClose is called by rtspServer.
@ -175,7 +175,7 @@ func (s *rtspSession) onClose(err error) {
if s.onReadCmd != nil { if s.onReadCmd != nil {
s.onReadCmd.Close() s.onReadCmd.Close()
s.onReadCmd = nil s.onReadCmd = nil
s.log(logger.Info, "runOnRead command stopped") s.Log(logger.Info, "runOnRead command stopped")
} }
} }
@ -190,7 +190,7 @@ func (s *rtspSession) onClose(err error) {
s.path = nil s.path = nil
s.stream = nil s.stream = nil
s.log(logger.Info, "destroyed (%v)", err) s.Log(logger.Info, "destroyed (%v)", err)
} }
// onAnnounce is called by rtspServer. // onAnnounce is called by rtspServer.
@ -217,7 +217,7 @@ func (s *rtspSession) onAnnounce(c *rtspConn, ctx *gortsplib.ServerHandlerOnAnno
if res.err != nil { if res.err != nil {
switch terr := res.err.(type) { switch terr := res.err.(type) {
case pathErrAuthNotCritical: case pathErrAuthNotCritical:
s.log(logger.Debug, "non-critical authentication error: %s", terr.message) s.Log(logger.Debug, "non-critical authentication error: %s", terr.message)
return terr.response, nil return terr.response, nil
case pathErrAuthCritical: case pathErrAuthCritical:
@ -296,7 +296,7 @@ func (s *rtspSession) onSetup(c *rtspConn, ctx *gortsplib.ServerHandlerOnSetupCt
if res.err != nil { if res.err != nil {
switch terr := res.err.(type) { switch terr := res.err.(type) {
case pathErrAuthNotCritical: case pathErrAuthNotCritical:
s.log(logger.Debug, "non-critical authentication error: %s", terr.message) s.Log(logger.Debug, "non-critical authentication error: %s", terr.message)
return terr.response, nil, nil return terr.response, nil, nil
case pathErrAuthCritical: case pathErrAuthCritical:
@ -340,7 +340,7 @@ func (s *rtspSession) onPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Respo
h := make(base.Header) h := make(base.Header)
if s.session.State() == gortsplib.ServerSessionStatePrePlay { if s.session.State() == gortsplib.ServerSessionStatePrePlay {
s.log(logger.Info, "is reading from path '%s', with %s, %s", s.Log(logger.Info, "is reading from path '%s', with %s, %s",
s.path.name, s.path.name,
s.session.SetuppedTransport(), s.session.SetuppedTransport(),
sourceMediaInfo(s.session.SetuppedMedias())) sourceMediaInfo(s.session.SetuppedMedias()))
@ -348,14 +348,14 @@ func (s *rtspSession) onPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Respo
pathConf := s.path.safeConf() pathConf := s.path.safeConf()
if pathConf.RunOnRead != "" { if pathConf.RunOnRead != "" {
s.log(logger.Info, "runOnRead command started") s.Log(logger.Info, "runOnRead command started")
s.onReadCmd = externalcmd.NewCmd( s.onReadCmd = externalcmd.NewCmd(
s.externalCmdPool, s.externalCmdPool,
pathConf.RunOnRead, pathConf.RunOnRead,
pathConf.RunOnReadRestart, pathConf.RunOnReadRestart,
s.path.externalCmdEnv(), s.path.externalCmdEnv(),
func(co int) { func(co int) {
s.log(logger.Info, "runOnRead command exited with code %d", co) s.Log(logger.Info, "runOnRead command exited with code %d", co)
}) })
} }
@ -383,7 +383,7 @@ func (s *rtspSession) onRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.R
}, res.err }, res.err
} }
s.log(logger.Info, "is publishing to path '%s', with %s, %s", s.Log(logger.Info, "is publishing to path '%s', with %s, %s",
s.path.name, s.path.name,
s.session.SetuppedTransport(), s.session.SetuppedTransport(),
sourceMediaInfo(s.session.AnnouncedMedias())) sourceMediaInfo(s.session.AnnouncedMedias()))
@ -395,10 +395,7 @@ func (s *rtspSession) onRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.R
writeFunc := getRTSPWriteFunc(medi, forma, s.stream) writeFunc := getRTSPWriteFunc(medi, forma, s.stream)
ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := writeFunc(pkt) writeFunc(pkt)
if err != nil {
s.log(logger.Warn, "%v", err)
}
}) })
} }
} }
@ -417,7 +414,7 @@ func (s *rtspSession) onPause(ctx *gortsplib.ServerHandlerOnPauseCtx) (*base.Res
switch s.session.State() { switch s.session.State() {
case gortsplib.ServerSessionStatePlay: case gortsplib.ServerSessionStatePlay:
if s.onReadCmd != nil { if s.onReadCmd != nil {
s.log(logger.Info, "runOnRead command stopped") s.Log(logger.Info, "runOnRead command stopped")
s.onReadCmd.Close() s.onReadCmd.Close()
} }
@ -470,10 +467,10 @@ func (s *rtspSession) apiSourceDescribe() interface{} {
// onPacketLost is called by rtspServer. // onPacketLost is called by rtspServer.
func (s *rtspSession) onPacketLost(ctx *gortsplib.ServerHandlerOnPacketLostCtx) { func (s *rtspSession) onPacketLost(ctx *gortsplib.ServerHandlerOnPacketLostCtx) {
s.log(logger.Warn, ctx.Error.Error()) s.Log(logger.Warn, ctx.Error.Error())
} }
// onDecodeError is called by rtspServer. // onDecodeError is called by rtspServer.
func (s *rtspSession) onDecodeError(ctx *gortsplib.ServerHandlerOnDecodeErrorCtx) { func (s *rtspSession) onDecodeError(ctx *gortsplib.ServerHandlerOnDecodeErrorCtx) {
s.log(logger.Warn, ctx.Error.Error()) s.Log(logger.Warn, ctx.Error.Error())
} }

9
internal/core/rtsp_source.go
Unescape View File

@ -19,7 +19,7 @@ import (
) )
type rtspSourceParent interface { type rtspSourceParent interface {
log(logger.Level, string, ...interface{}) logger.Writer
sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes
sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq) sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq)
} }
@ -46,7 +46,7 @@ func newRTSPSource(
} }
func (s *rtspSource) Log(level logger.Level, format string, args ...interface{}) { func (s *rtspSource) Log(level logger.Level, format string, args ...interface{}) {
s.parent.log(level, "[rtsp source] "+format, args...) s.parent.Log(level, "[rtsp source] "+format, args...)
} }
// run implements sourceStaticImpl. // run implements sourceStaticImpl.
@ -140,10 +140,7 @@ func (s *rtspSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf cha
writeFunc := getRTSPWriteFunc(medi, forma, res.stream) writeFunc := getRTSPWriteFunc(medi, forma, res.stream)
c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := writeFunc(pkt) writeFunc(pkt)
if err != nil {
s.Log(logger.Warn, "%v", err)
}
}) })
} }
} }

3
internal/core/source.go
Unescape View File

@ -5,6 +5,8 @@ import (
"strings" "strings"
"github.com/bluenviron/gortsplib/v3/pkg/media" "github.com/bluenviron/gortsplib/v3/pkg/media"
"github.com/aler9/mediamtx/internal/logger"
) )
// source is an entity that can provide a stream. // source is an entity that can provide a stream.
@ -13,6 +15,7 @@ import (
// - sourceStatic // - sourceStatic
// - sourceRedirect // - sourceRedirect
type source interface { type source interface {
logger.Writer
apiSourceDescribe() interface{} apiSourceDescribe() interface{}
} }

7
internal/core/source_redirect.go
Unescape View File

@ -1,8 +1,15 @@
package core package core
import (
"github.com/aler9/mediamtx/internal/logger"
)
// sourceRedirect is a source that redirects to another one. // sourceRedirect is a source that redirects to another one.
type sourceRedirect struct{} type sourceRedirect struct{}
func (*sourceRedirect) Log(logger.Level, string, ...interface{}) {
}
// apiSourceDescribe implements source. // apiSourceDescribe implements source.
func (*sourceRedirect) apiSourceDescribe() interface{} { func (*sourceRedirect) apiSourceDescribe() interface{} {
return struct { return struct {

8
internal/core/source_static.go
Unescape View File

@ -15,13 +15,13 @@ const (
) )
type sourceStaticImpl interface { type sourceStaticImpl interface {
Log(logger.Level, string, ...interface{}) logger.Writer
run(context.Context, *conf.PathConf, chan *conf.PathConf) error run(context.Context, *conf.PathConf, chan *conf.PathConf) error
apiSourceDescribe() interface{} apiSourceDescribe() interface{}
} }
type sourceStaticParent interface { type sourceStaticParent interface {
log(logger.Level, string, ...interface{}) logger.Writer
sourceStaticSetReady(context.Context, pathSourceStaticSetReadyReq) sourceStaticSetReady(context.Context, pathSourceStaticSetReadyReq)
sourceStaticSetNotReady(context.Context, pathSourceStaticSetNotReadyReq) sourceStaticSetNotReady(context.Context, pathSourceStaticSetNotReadyReq)
} }
@ -128,8 +128,8 @@ func (s *sourceStatic) stop() {
<-s.done <-s.done
} }
func (s *sourceStatic) log(level logger.Level, format string, args ...interface{}) { func (s *sourceStatic) Log(level logger.Level, format string, args ...interface{}) {
s.parent.log(level, format, args...) s.parent.Log(level, format, args...)
} }
func (s *sourceStatic) run() { func (s *sourceStatic) run() {

12
internal/core/stream.go
Unescape View File

@ -10,8 +10,9 @@ import (
type stream struct { type stream struct {
bytesReceived *uint64 bytesReceived *uint64
rtspStream *gortsplib.ServerStream
smedias map[*media.Media]*streamMedia rtspStream *gortsplib.ServerStream
smedias map[*media.Media]*streamMedia
} }
func newStream( func newStream(
@ -19,6 +20,7 @@ func newStream(
medias media.Medias, medias media.Medias,
generateRTPPackets bool, generateRTPPackets bool,
bytesReceived *uint64, bytesReceived *uint64,
source source,
) (*stream, error) { ) (*stream, error) {
s := &stream{ s := &stream{
bytesReceived: bytesReceived, bytesReceived: bytesReceived,
@ -29,7 +31,7 @@ func newStream(
for _, media := range s.rtspStream.Medias() { for _, media := range s.rtspStream.Medias() {
var err error var err error
s.smedias[media], err = newStreamMedia(udpMaxPayloadSize, media, generateRTPPackets) s.smedias[media], err = newStreamMedia(udpMaxPayloadSize, media, generateRTPPackets, source)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -60,8 +62,8 @@ func (s *stream) readerRemove(r reader) {
} }
} }
func (s *stream) writeUnit(medi *media.Media, forma formats.Format, data formatprocessor.Unit) error { func (s *stream) writeUnit(medi *media.Media, forma formats.Format, data formatprocessor.Unit) {
sm := s.smedias[medi] sm := s.smedias[medi]
sf := sm.formats[forma] sf := sm.formats[forma]
return sf.writeUnit(s, medi, data) sf.writeUnit(s, medi, data)
} }

13
internal/core/stream_format.go
Unescape View File

@ -8,9 +8,11 @@ import (
"github.com/bluenviron/gortsplib/v3/pkg/media" "github.com/bluenviron/gortsplib/v3/pkg/media"
"github.com/aler9/mediamtx/internal/formatprocessor" "github.com/aler9/mediamtx/internal/formatprocessor"
"github.com/aler9/mediamtx/internal/logger"
) )
type streamFormat struct { type streamFormat struct {
source source
proc formatprocessor.Processor proc formatprocessor.Processor
mutex sync.RWMutex mutex sync.RWMutex
nonRTSPReaders map[reader]func(formatprocessor.Unit) nonRTSPReaders map[reader]func(formatprocessor.Unit)
@ -20,13 +22,15 @@ func newStreamFormat(
udpMaxPayloadSize int, udpMaxPayloadSize int,
forma formats.Format, forma formats.Format,
generateRTPPackets bool, generateRTPPackets bool,
source source,
) (*streamFormat, error) { ) (*streamFormat, error) {
proc, err := formatprocessor.New(udpMaxPayloadSize, forma, generateRTPPackets) proc, err := formatprocessor.New(udpMaxPayloadSize, forma, generateRTPPackets, source)
if err != nil { if err != nil {
return nil, err return nil, err
} }
sf := &streamFormat{ sf := &streamFormat{
source: source,
proc: proc, proc: proc,
nonRTSPReaders: make(map[reader]func(formatprocessor.Unit)), nonRTSPReaders: make(map[reader]func(formatprocessor.Unit)),
} }
@ -46,7 +50,7 @@ func (sf *streamFormat) readerRemove(r reader) {
delete(sf.nonRTSPReaders, r) delete(sf.nonRTSPReaders, r)
} }
func (sf *streamFormat) writeUnit(s *stream, medi *media.Media, data formatprocessor.Unit) error { func (sf *streamFormat) writeUnit(s *stream, medi *media.Media, data formatprocessor.Unit) {
sf.mutex.RLock() sf.mutex.RLock()
defer sf.mutex.RUnlock() defer sf.mutex.RUnlock()
@ -54,7 +58,8 @@ func (sf *streamFormat) writeUnit(s *stream, medi *media.Media, data formatproce
err := sf.proc.Process(data, hasNonRTSPReaders) err := sf.proc.Process(data, hasNonRTSPReaders)
if err != nil { if err != nil {
return err sf.source.Log(logger.Warn, err.Error())
return
} }
// forward RTP packets to RTSP readers // forward RTP packets to RTSP readers
@ -67,6 +72,4 @@ func (sf *streamFormat) writeUnit(s *stream, medi *media.Media, data formatproce
for _, cb := range sf.nonRTSPReaders { for _, cb := range sf.nonRTSPReaders {
cb(data) cb(data)
} }
return nil
} }

3
internal/core/stream_media.go
Unescape View File

@ -12,6 +12,7 @@ type streamMedia struct {
func newStreamMedia(udpMaxPayloadSize int, func newStreamMedia(udpMaxPayloadSize int,
medi *media.Media, medi *media.Media,
generateRTPPackets bool, generateRTPPackets bool,
source source,
) (*streamMedia, error) { ) (*streamMedia, error) {
sm := &streamMedia{ sm := &streamMedia{
formats: make(map[formats.Format]*streamFormat), formats: make(map[formats.Format]*streamFormat),
@ -19,7 +20,7 @@ func newStreamMedia(udpMaxPayloadSize int,
for _, forma := range medi.Formats { for _, forma := range medi.Formats {
var err error var err error
sm.formats[forma], err = newStreamFormat(udpMaxPayloadSize, forma, generateRTPPackets) sm.formats[forma], err = newStreamFormat(udpMaxPayloadSize, forma, generateRTPPackets, source)
if err != nil { if err != nil {
return nil, err return nil, err
} }

24
internal/core/udp_source.go
Unescape View File

@ -97,7 +97,7 @@ func (r *packetConnReader) Read(p []byte) (int, error) {
} }
type udpSourceParent interface { type udpSourceParent interface {
log(logger.Level, string, ...interface{}) logger.Writer
sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes sourceStaticImplSetReady(req pathSourceStaticSetReadyReq) pathSourceStaticSetReadyRes
sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq) sourceStaticImplSetNotReady(req pathSourceStaticSetNotReadyReq)
} }
@ -118,7 +118,7 @@ func newUDPSource(
} }
func (s *udpSource) Log(level logger.Level, format string, args ...interface{}) { func (s *udpSource) Log(level logger.Level, format string, args ...interface{}) {
s.parent.log(level, "[udp source] "+format, args...) s.parent.Log(level, "[udp source] "+format, args...)
} }
// run implements sourceStaticImpl. // run implements sourceStaticImpl.
@ -196,14 +196,11 @@ func (s *udpSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
return return
} }
err = stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH264{ stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH264{
PTS: pts, PTS: pts,
AU: au, AU: au,
NTP: time.Now(), NTP: time.Now(),
}) })
if err != nil {
s.Log(logger.Warn, "%v", err)
}
} }
case *mpegts.CodecH265: case *mpegts.CodecH265:
@ -221,14 +218,11 @@ func (s *udpSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
return return
} }
err = stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH265{ stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH265{
PTS: pts, PTS: pts,
AU: au, AU: au,
NTP: time.Now(), NTP: time.Now(),
}) })
if err != nil {
s.Log(logger.Warn, "%v", err)
}
} }
case *mpegts.CodecMPEG4Audio: case *mpegts.CodecMPEG4Audio:
@ -256,14 +250,11 @@ func (s *udpSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
aus[i] = pkt.AU aus[i] = pkt.AU
} }
err = stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitMPEG4Audio{ stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitMPEG4Audio{
PTS: pts, PTS: pts,
AUs: aus, AUs: aus,
NTP: time.Now(), NTP: time.Now(),
}) })
if err != nil {
s.Log(logger.Warn, "%v", err)
}
} }
case *mpegts.CodecOpus: case *mpegts.CodecOpus:
@ -287,14 +278,11 @@ func (s *udpSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
} }
pos += n pos += n
err = stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitOpus{ stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitOpus{
PTS: pts, PTS: pts,
Frame: au.Frame, Frame: au.Frame,
NTP: time.Now(), NTP: time.Now(),
}) })
if err != nil {
s.Log(logger.Warn, "%v", err)
}
if len(data[pos:]) == 0 { if len(data[pos:]) == 0 {
break break

20
internal/core/webrtc_conn.go
Unescape View File

@ -84,7 +84,7 @@ type webRTCConnPathManager interface {
} }
type webRTCConnParent interface { type webRTCConnParent interface {
log(logger.Level, string, ...interface{}) logger.Writer
connClose(*webRTCConn) connClose(*webRTCConn)
} }
@ -143,7 +143,7 @@ func newWebRTCConn(
closed: make(chan struct{}), closed: make(chan struct{}),
} }
c.log(logger.Info, "opened") c.Log(logger.Info, "opened")
wg.Add(1) wg.Add(1)
go c.run() go c.run()
@ -252,8 +252,8 @@ func (c *webRTCConn) bytesSent() uint64 {
return 0 return 0
} }
func (c *webRTCConn) log(level logger.Level, format string, args ...interface{}) { func (c *webRTCConn) Log(level logger.Level, format string, args ...interface{}) {
c.parent.log(level, "[conn %v] "+format, append([]interface{}{c.wsconn.RemoteAddr()}, args...)...) c.parent.Log(level, "[conn %v] "+format, append([]interface{}{c.wsconn.RemoteAddr()}, args...)...)
} }
func (c *webRTCConn) run() { func (c *webRTCConn) run() {
@ -281,7 +281,7 @@ func (c *webRTCConn) run() {
c.parent.connClose(c) c.parent.connClose(c)
c.log(logger.Info, "closed (%v)", err) c.Log(logger.Info, "closed (%v)", err)
} }
func (c *webRTCConn) runInner(ctx context.Context) error { func (c *webRTCConn) runInner(ctx context.Context) error {
@ -377,7 +377,7 @@ func (c *webRTCConn) runInner(ctx context.Context) error {
default: default:
} }
c.log(logger.Debug, "peer connection state: "+state.String()) c.Log(logger.Debug, "peer connection state: "+state.String())
switch state { switch state {
case webrtc.PeerConnectionStateConnected: case webrtc.PeerConnectionStateConnected:
@ -475,14 +475,14 @@ outer:
for { for {
select { select {
case candidate := <-localCandidate: case candidate := <-localCandidate:
c.log(logger.Debug, "local candidate: %+v", candidate.Candidate) c.Log(logger.Debug, "local candidate: %+v", candidate.Candidate)
err := c.wsconn.WriteJSON(candidate) err := c.wsconn.WriteJSON(candidate)
if err != nil { if err != nil {
return err return err
} }
case candidate := <-remoteCandidate: case candidate := <-remoteCandidate:
c.log(logger.Debug, "remote candidate: %+v", candidate.Candidate) c.Log(logger.Debug, "remote candidate: %+v", candidate.Candidate)
err := pc.AddICECandidate(*candidate) err := pc.AddICECandidate(*candidate)
if err != nil { if err != nil {
return err return err
@ -512,7 +512,7 @@ outer:
c.curPC = pc c.curPC = pc
c.mutex.Unlock() c.mutex.Unlock()
c.log(logger.Info, "peer connection established, local candidate: %v, remote candidate: %v", c.Log(logger.Info, "peer connection established, local candidate: %v, remote candidate: %v",
c.localCandidate(), c.remoteCandidate()) c.localCandidate(), c.remoteCandidate())
ringBuffer, _ := ringbuffer.New(uint64(c.readBufferCount)) ringBuffer, _ := ringbuffer.New(uint64(c.readBufferCount))
@ -530,7 +530,7 @@ outer:
} }
defer res.stream.readerRemove(c) defer res.stream.readerRemove(c)
c.log(logger.Info, "is reading from path '%s', %s", c.Log(logger.Info, "is reading from path '%s', %s",
path.name, sourceMediaInfo(gatherMedias(tracks))) path.name, sourceMediaInfo(gatherMedias(tracks)))
go func() { go func() {

10
internal/core/webrtc_server.go
Unescape View File

@ -64,7 +64,7 @@ type webRTCConnNewReq struct {
} }
type webRTCServerParent interface { type webRTCServerParent interface {
Log(logger.Level, string, ...interface{}) logger.Writer
} }
type webRTCServer struct { type webRTCServer struct {
@ -204,7 +204,7 @@ func newWebRTCServer(
if tcpMuxLn != nil { if tcpMuxLn != nil {
str += ", " + iceTCPMuxAddress + " (ICE/TCP)" str += ", " + iceTCPMuxAddress + " (ICE/TCP)"
} }
s.log(logger.Info, str) s.Log(logger.Info, str)
if s.metrics != nil { if s.metrics != nil {
s.metrics.webRTCServerSet(s) s.metrics.webRTCServerSet(s)
@ -216,12 +216,12 @@ func newWebRTCServer(
} }
// Log is the main logging function. // Log is the main logging function.
func (s *webRTCServer) log(level logger.Level, format string, args ...interface{}) { func (s *webRTCServer) Log(level logger.Level, format string, args ...interface{}) {
s.parent.Log(level, "[WebRTC] "+format, append([]interface{}{}, args...)...) s.parent.Log(level, "[WebRTC] "+format, append([]interface{}{}, args...)...)
} }
func (s *webRTCServer) close() { func (s *webRTCServer) close() {
s.log(logger.Info, "listener is closing") s.Log(logger.Info, "listener is closing")
s.ctxCancel() s.ctxCancel()
<-s.done <-s.done
} }
@ -369,7 +369,7 @@ func (s *webRTCServer) onRequest(ctx *gin.Context) {
err := s.authenticate(res.path, ctx) err := s.authenticate(res.path, ctx)
if err != nil { if err != nil {
if terr, ok := err.(pathErrAuthCritical); ok { if terr, ok := err.(pathErrAuthCritical); ok {
s.log(logger.Info, "authentication error: %s", terr.message) s.Log(logger.Info, "authentication error: %s", terr.message)
ctx.Writer.Header().Set("WWW-Authenticate", `Basic realm="mediamtx"`) ctx.Writer.Header().Set("WWW-Authenticate", `Basic realm="mediamtx"`)
ctx.Writer.WriteHeader(http.StatusUnauthorized) ctx.Writer.WriteHeader(http.StatusUnauthorized)
return return

3
internal/formatprocessor/generic.go
Unescape View File

@ -6,6 +6,8 @@ import (
"github.com/bluenviron/gortsplib/v3/pkg/formats" "github.com/bluenviron/gortsplib/v3/pkg/formats"
"github.com/pion/rtp" "github.com/pion/rtp"
"github.com/aler9/mediamtx/internal/logger"
) )
// UnitGeneric is a generic data unit. // UnitGeneric is a generic data unit.
@ -32,6 +34,7 @@ func newGeneric(
udpMaxPayloadSize int, udpMaxPayloadSize int,
forma formats.Format, forma formats.Format,
generateRTPPackets bool, generateRTPPackets bool,
log logger.Writer,
) (*formatProcessorGeneric, error) { ) (*formatProcessorGeneric, error) {
if generateRTPPackets { if generateRTPPackets {
return nil, fmt.Errorf("we don't know how to generate RTP packets of format %+v", forma) return nil, fmt.Errorf("we don't know how to generate RTP packets of format %+v", forma)

2
internal/formatprocessor/generic_test.go
Unescape View File

@ -15,7 +15,7 @@ func TestGenericRemovePadding(t *testing.T) {
} }
forma.Init() forma.Init()
p, err := New(1472, forma, false) p, err := New(1472, forma, false, nil)
require.NoError(t, err) require.NoError(t, err)
pkt := &rtp.Packet{ pkt := &rtp.Packet{

48
internal/formatprocessor/h264.go
Unescape View File

@ -8,6 +8,8 @@ import (
"github.com/bluenviron/gortsplib/v3/pkg/formats/rtph264" "github.com/bluenviron/gortsplib/v3/pkg/formats/rtph264"
"github.com/bluenviron/mediacommon/pkg/codecs/h264" "github.com/bluenviron/mediacommon/pkg/codecs/h264"
"github.com/pion/rtp" "github.com/pion/rtp"
"github.com/aler9/mediamtx/internal/logger"
) )
// extract SPS and PPS without decoding RTP packets // extract SPS and PPS without decoding RTP packets
@ -88,28 +90,33 @@ func (d *UnitH264) GetNTP() time.Time {
type formatProcessorH264 struct { type formatProcessorH264 struct {
udpMaxPayloadSize int udpMaxPayloadSize int
format *formats.H264 format *formats.H264
log logger.Writer
encoder *rtph264.Encoder encoder *rtph264.Encoder
decoder *rtph264.Decoder decoder *rtph264.Decoder
lastKeyFrameReceived time.Time
} }
func newH264( func newH264(
udpMaxPayloadSize int, udpMaxPayloadSize int,
forma *formats.H264, forma *formats.H264,
allocateEncoder bool, generateRTPPackets bool,
log logger.Writer,
) (*formatProcessorH264, error) { ) (*formatProcessorH264, error) {
t := &formatProcessorH264{ t := &formatProcessorH264{
udpMaxPayloadSize: udpMaxPayloadSize, udpMaxPayloadSize: udpMaxPayloadSize,
format: forma, format: forma,
log: log,
} }
if allocateEncoder { if generateRTPPackets {
t.encoder = &rtph264.Encoder{ t.encoder = &rtph264.Encoder{
PayloadMaxSize: udpMaxPayloadSize - 12, PayloadMaxSize: udpMaxPayloadSize - 12,
PayloadType: forma.PayloadTyp, PayloadType: forma.PayloadTyp,
PacketizationMode: forma.PacketizationMode, PacketizationMode: forma.PacketizationMode,
} }
t.encoder.Init() t.encoder.Init()
t.lastKeyFrameReceived = time.Now()
} }
return t, nil return t, nil
@ -166,24 +173,37 @@ func (t *formatProcessorH264) updateTrackParametersFromNALUs(nalus [][]byte) {
} }
} }
func (t *formatProcessorH264) checkKeyFrameInterval(isKeyFrame bool) {
if isKeyFrame {
t.lastKeyFrameReceived = time.Now()
} else {
now := time.Now()
if now.Sub(t.lastKeyFrameReceived) >= maxKeyFrameInterval {
t.lastKeyFrameReceived = now
t.log.Log(logger.Warn, "no key frames received in %v, stream can't be decoded")
}
}
}
func (t *formatProcessorH264) remuxAccessUnit(nalus [][]byte) [][]byte { func (t *formatProcessorH264) remuxAccessUnit(nalus [][]byte) [][]byte {
addParameters := false isKeyFrame := false
n := 0 n := 0
for _, nalu := range nalus { for _, nalu := range nalus {
typ := h264.NALUType(nalu[0] & 0x1F) typ := h264.NALUType(nalu[0] & 0x1F)
switch typ { switch typ {
case h264.NALUTypeSPS, h264.NALUTypePPS: // remove parameters case h264.NALUTypeSPS, h264.NALUTypePPS: // parameters: remove
continue continue
case h264.NALUTypeAccessUnitDelimiter: // remove AUDs case h264.NALUTypeAccessUnitDelimiter: // AUD: remove
continue continue
case h264.NALUTypeIDR: // prepend parameters if there's at least an IDR case h264.NALUTypeIDR: // key frame
if !addParameters { if !isKeyFrame {
addParameters = true isKeyFrame = true
// prepend parameters
if t.format.SPS != nil && t.format.PPS != nil { if t.format.SPS != nil && t.format.PPS != nil {
n += 2 n += 2
} }
@ -192,6 +212,8 @@ func (t *formatProcessorH264) remuxAccessUnit(nalus [][]byte) [][]byte {
n++ n++
} }
t.checkKeyFrameInterval(isKeyFrame)
if n == 0 { if n == 0 {
return nil return nil
} }
@ -199,7 +221,7 @@ func (t *formatProcessorH264) remuxAccessUnit(nalus [][]byte) [][]byte {
filteredNALUs := make([][]byte, n) filteredNALUs := make([][]byte, n)
i := 0 i := 0
if addParameters && t.format.SPS != nil && t.format.PPS != nil { if isKeyFrame && t.format.SPS != nil && t.format.PPS != nil {
filteredNALUs[0] = t.format.SPS filteredNALUs[0] = t.format.SPS
filteredNALUs[1] = t.format.PPS filteredNALUs[1] = t.format.PPS
i = 2 i = 2
@ -256,6 +278,7 @@ func (t *formatProcessorH264) Process(unit Unit, hasNonRTSPReaders bool) error {
if hasNonRTSPReaders || t.encoder != nil { if hasNonRTSPReaders || t.encoder != nil {
if t.decoder == nil { if t.decoder == nil {
t.decoder = t.format.CreateDecoder() t.decoder = t.format.CreateDecoder()
t.lastKeyFrameReceived = time.Now()
} }
if t.encoder != nil { if t.encoder != nil {
@ -271,9 +294,8 @@ func (t *formatProcessorH264) Process(unit Unit, hasNonRTSPReaders bool) error {
return err return err
} }
tunit.AU = au tunit.AU = t.remuxAccessUnit(au)
tunit.PTS = pts tunit.PTS = pts
tunit.AU = t.remuxAccessUnit(tunit.AU)
} }
// route packet as is // route packet as is

6
internal/formatprocessor/h264_test.go
Unescape View File

@ -16,7 +16,7 @@ func TestH264DynamicParams(t *testing.T) {
PacketizationMode: 1, PacketizationMode: 1,
} }
p, err := New(1472, forma, false) p, err := New(1472, forma, false, nil)
require.NoError(t, err) require.NoError(t, err)
enc := forma.CreateEncoder() enc := forma.CreateEncoder()
@ -61,7 +61,7 @@ func TestH264OversizedPackets(t *testing.T) {
PacketizationMode: 1, PacketizationMode: 1,
} }
p, err := New(1472, forma, false) p, err := New(1472, forma, false, nil)
require.NoError(t, err) require.NoError(t, err)
var out []*rtp.Packet var out []*rtp.Packet
@ -158,7 +158,7 @@ func TestH264EmptyPacket(t *testing.T) {
PacketizationMode: 1, PacketizationMode: 1,
} }
p, err := New(1472, forma, true) p, err := New(1472, forma, true, nil)
require.NoError(t, err) require.NoError(t, err)
unit := &UnitH264{ unit := &UnitH264{

49
internal/formatprocessor/h265.go
Unescape View File

@ -8,6 +8,8 @@ import (
"github.com/bluenviron/gortsplib/v3/pkg/formats/rtph265" "github.com/bluenviron/gortsplib/v3/pkg/formats/rtph265"
"github.com/bluenviron/mediacommon/pkg/codecs/h265" "github.com/bluenviron/mediacommon/pkg/codecs/h265"
"github.com/pion/rtp" "github.com/pion/rtp"
"github.com/aler9/mediamtx/internal/logger"
) )
// extract VPS, SPS and PPS without decoding RTP packets // extract VPS, SPS and PPS without decoding RTP packets
@ -95,27 +97,32 @@ func (d *UnitH265) GetNTP() time.Time {
type formatProcessorH265 struct { type formatProcessorH265 struct {
udpMaxPayloadSize int udpMaxPayloadSize int
format *formats.H265 format *formats.H265
log logger.Writer
encoder *rtph265.Encoder encoder *rtph265.Encoder
decoder *rtph265.Decoder decoder *rtph265.Decoder
lastKeyFrameReceived time.Time
} }
func newH265( func newH265(
udpMaxPayloadSize int, udpMaxPayloadSize int,
forma *formats.H265, forma *formats.H265,
allocateEncoder bool, generateRTPPackets bool,
log logger.Writer,
) (*formatProcessorH265, error) { ) (*formatProcessorH265, error) {
t := &formatProcessorH265{ t := &formatProcessorH265{
udpMaxPayloadSize: udpMaxPayloadSize, udpMaxPayloadSize: udpMaxPayloadSize,
format: forma, format: forma,
log: log,
} }
if allocateEncoder { if generateRTPPackets {
t.encoder = &rtph265.Encoder{ t.encoder = &rtph265.Encoder{
PayloadMaxSize: t.udpMaxPayloadSize - 12, PayloadMaxSize: t.udpMaxPayloadSize - 12,
PayloadType: forma.PayloadTyp, PayloadType: forma.PayloadTyp,
} }
t.encoder.Init() t.encoder.Init()
t.lastKeyFrameReceived = time.Now()
} }
return t, nil return t, nil
@ -186,25 +193,37 @@ func (t *formatProcessorH265) updateTrackParametersFromNALUs(nalus [][]byte) {
} }
} }
func (t *formatProcessorH265) checkKeyFrameInterval(isKeyFrame bool) {
if isKeyFrame {
t.lastKeyFrameReceived = time.Now()
} else {
now := time.Now()
if now.Sub(t.lastKeyFrameReceived) >= maxKeyFrameInterval {
t.lastKeyFrameReceived = now
t.log.Log(logger.Warn, "no key frames received in %v, stream can't be decoded")
}
}
}
func (t *formatProcessorH265) remuxAccessUnit(nalus [][]byte) [][]byte { func (t *formatProcessorH265) remuxAccessUnit(nalus [][]byte) [][]byte {
addParameters := false isKeyFrame := false
n := 0 n := 0
for _, nalu := range nalus { for _, nalu := range nalus {
typ := h265.NALUType((nalu[0] >> 1) & 0b111111) typ := h265.NALUType((nalu[0] >> 1) & 0b111111)
switch typ { switch typ {
case h265.NALUType_VPS_NUT, h265.NALUType_SPS_NUT, h265.NALUType_PPS_NUT: // remove parameters case h265.NALUType_VPS_NUT, h265.NALUType_SPS_NUT, h265.NALUType_PPS_NUT: // parameters: remove
continue continue
case h265.NALUType_AUD_NUT: // remove AUDs case h265.NALUType_AUD_NUT: // AUD: remove
continue continue
// prepend parameters if there's at least a random access unit case h265.NALUType_IDR_W_RADL, h265.NALUType_IDR_N_LP, h265.NALUType_CRA_NUT: // key frame
case h265.NALUType_IDR_W_RADL, h265.NALUType_IDR_N_LP, h265.NALUType_CRA_NUT: if !isKeyFrame {
if !addParameters { isKeyFrame = true
addParameters = true
// prepend parameters
if t.format.VPS != nil && t.format.SPS != nil && t.format.PPS != nil { if t.format.VPS != nil && t.format.SPS != nil && t.format.PPS != nil {
n += 3 n += 3
} }
@ -213,6 +232,8 @@ func (t *formatProcessorH265) remuxAccessUnit(nalus [][]byte) [][]byte {
n++ n++
} }
t.checkKeyFrameInterval(isKeyFrame)
if n == 0 { if n == 0 {
return nil return nil
} }
@ -220,7 +241,7 @@ func (t *formatProcessorH265) remuxAccessUnit(nalus [][]byte) [][]byte {
filteredNALUs := make([][]byte, n) filteredNALUs := make([][]byte, n)
i := 0 i := 0
if addParameters && t.format.VPS != nil && t.format.SPS != nil && t.format.PPS != nil { if isKeyFrame && t.format.VPS != nil && t.format.SPS != nil && t.format.PPS != nil {
filteredNALUs[0] = t.format.VPS filteredNALUs[0] = t.format.VPS
filteredNALUs[1] = t.format.SPS filteredNALUs[1] = t.format.SPS
filteredNALUs[2] = t.format.PPS filteredNALUs[2] = t.format.PPS
@ -278,6 +299,7 @@ func (t *formatProcessorH265) Process(unit Unit, hasNonRTSPReaders bool) error {
if hasNonRTSPReaders || t.encoder != nil { if hasNonRTSPReaders || t.encoder != nil {
if t.decoder == nil { if t.decoder == nil {
t.decoder = t.format.CreateDecoder() t.decoder = t.format.CreateDecoder()
t.lastKeyFrameReceived = time.Now()
} }
if t.encoder != nil { if t.encoder != nil {
@ -293,9 +315,8 @@ func (t *formatProcessorH265) Process(unit Unit, hasNonRTSPReaders bool) error {
return err return err
} }
tunit.AU = au tunit.AU = t.remuxAccessUnit(au)
tunit.PTS = pts tunit.PTS = pts
tunit.AU = t.remuxAccessUnit(tunit.AU)
} }
// route packet as is // route packet as is

6
internal/formatprocessor/h265_test.go
Unescape View File

@ -15,7 +15,7 @@ func TestH265DynamicParams(t *testing.T) {
PayloadTyp: 96, PayloadTyp: 96,
} }
p, err := New(1472, forma, false) p, err := New(1472, forma, false, nil)
require.NoError(t, err) require.NoError(t, err)
enc := forma.CreateEncoder() enc := forma.CreateEncoder()
@ -66,7 +66,7 @@ func TestH265OversizedPackets(t *testing.T) {
PPS: []byte{byte(h265.NALUType_PPS_NUT) << 1, 16, 17, 18}, PPS: []byte{byte(h265.NALUType_PPS_NUT) << 1, 16, 17, 18},
} }
p, err := New(1472, forma, false) p, err := New(1472, forma, false, nil)
require.NoError(t, err) require.NoError(t, err)
var out []*rtp.Packet var out []*rtp.Packet
@ -150,7 +150,7 @@ func TestH265EmptyPacket(t *testing.T) {
PayloadTyp: 96, PayloadTyp: 96,
} }
p, err := New(1472, forma, true) p, err := New(1472, forma, true, nil)
require.NoError(t, err) require.NoError(t, err)
unit := &UnitH265{ unit := &UnitH265{

7
internal/formatprocessor/mpeg2audio.go
Unescape View File

@ -7,6 +7,8 @@ import (
"github.com/bluenviron/gortsplib/v3/pkg/formats" "github.com/bluenviron/gortsplib/v3/pkg/formats"
"github.com/bluenviron/gortsplib/v3/pkg/formats/rtpmpeg2audio" "github.com/bluenviron/gortsplib/v3/pkg/formats/rtpmpeg2audio"
"github.com/pion/rtp" "github.com/pion/rtp"
"github.com/aler9/mediamtx/internal/logger"
) )
// UnitMPEG2Audio is a MPEG-2 Audio data unit. // UnitMPEG2Audio is a MPEG-2 Audio data unit.
@ -37,14 +39,15 @@ type formatProcessorMPEG2Audio struct {
func newMPEG2Audio( func newMPEG2Audio(
udpMaxPayloadSize int, udpMaxPayloadSize int,
forma *formats.MPEG2Audio, forma *formats.MPEG2Audio,
allocateEncoder bool, generateRTPPackets bool,
log logger.Writer,
) (*formatProcessorMPEG2Audio, error) { ) (*formatProcessorMPEG2Audio, error) {
t := &formatProcessorMPEG2Audio{ t := &formatProcessorMPEG2Audio{
udpMaxPayloadSize: udpMaxPayloadSize, udpMaxPayloadSize: udpMaxPayloadSize,
format: forma, format: forma,
} }
if allocateEncoder { if generateRTPPackets {
t.encoder = &rtpmpeg2audio.Encoder{ t.encoder = &rtpmpeg2audio.Encoder{
PayloadMaxSize: t.udpMaxPayloadSize - 12, PayloadMaxSize: t.udpMaxPayloadSize - 12,
} }

7
internal/formatprocessor/mpeg4audio.go
Unescape View File

@ -7,6 +7,8 @@ import (
"github.com/bluenviron/gortsplib/v3/pkg/formats" "github.com/bluenviron/gortsplib/v3/pkg/formats"
"github.com/bluenviron/gortsplib/v3/pkg/formats/rtpmpeg4audio" "github.com/bluenviron/gortsplib/v3/pkg/formats/rtpmpeg4audio"
"github.com/pion/rtp" "github.com/pion/rtp"
"github.com/aler9/mediamtx/internal/logger"
) )
// UnitMPEG4Audio is a MPEG-4 Audio data unit. // UnitMPEG4Audio is a MPEG-4 Audio data unit.
@ -37,14 +39,15 @@ type formatProcessorMPEG4Audio struct {
func newMPEG4Audio( func newMPEG4Audio(
udpMaxPayloadSize int, udpMaxPayloadSize int,
forma *formats.MPEG4Audio, forma *formats.MPEG4Audio,
allocateEncoder bool, generateRTPPackets bool,
log logger.Writer,
) (*formatProcessorMPEG4Audio, error) { ) (*formatProcessorMPEG4Audio, error) {
t := &formatProcessorMPEG4Audio{ t := &formatProcessorMPEG4Audio{
udpMaxPayloadSize: udpMaxPayloadSize, udpMaxPayloadSize: udpMaxPayloadSize,
format: forma, format: forma,
} }
if allocateEncoder { if generateRTPPackets {
t.encoder = &rtpmpeg4audio.Encoder{ t.encoder = &rtpmpeg4audio.Encoder{
PayloadMaxSize: t.udpMaxPayloadSize - 12, PayloadMaxSize: t.udpMaxPayloadSize - 12,
PayloadType: forma.PayloadTyp, PayloadType: forma.PayloadTyp,

7
internal/formatprocessor/opus.go
Unescape View File

@ -7,6 +7,8 @@ import (
"github.com/bluenviron/gortsplib/v3/pkg/formats" "github.com/bluenviron/gortsplib/v3/pkg/formats"
"github.com/bluenviron/gortsplib/v3/pkg/formats/rtpsimpleaudio" "github.com/bluenviron/gortsplib/v3/pkg/formats/rtpsimpleaudio"
"github.com/pion/rtp" "github.com/pion/rtp"
"github.com/aler9/mediamtx/internal/logger"
) )
// UnitOpus is a Opus data unit. // UnitOpus is a Opus data unit.
@ -37,14 +39,15 @@ type formatProcessorOpus struct {
func newOpus( func newOpus(
udpMaxPayloadSize int, udpMaxPayloadSize int,
forma *formats.Opus, forma *formats.Opus,
allocateEncoder bool, generateRTPPackets bool,
log logger.Writer,
) (*formatProcessorOpus, error) { ) (*formatProcessorOpus, error) {
t := &formatProcessorOpus{ t := &formatProcessorOpus{
udpMaxPayloadSize: udpMaxPayloadSize, udpMaxPayloadSize: udpMaxPayloadSize,
format: forma, format: forma,
} }
if allocateEncoder { if generateRTPPackets {
t.encoder = &rtpsimpleaudio.Encoder{ t.encoder = &rtpsimpleaudio.Encoder{
PayloadMaxSize: t.udpMaxPayloadSize - 12, PayloadMaxSize: t.udpMaxPayloadSize - 12,
PayloadType: forma.PayloadTyp, PayloadType: forma.PayloadTyp,

29
internal/formatprocessor/processor.go
Unescape View File

@ -2,12 +2,20 @@
package formatprocessor package formatprocessor
import ( import (
"time"
"github.com/bluenviron/gortsplib/v3/pkg/formats" "github.com/bluenviron/gortsplib/v3/pkg/formats"
"github.com/aler9/mediamtx/internal/logger"
)
const (
maxKeyFrameInterval = 10 * time.Second
) )
// Processor allows to cleanup and normalize streams. // Processor cleans and normalizes streams.
type Processor interface { type Processor interface {
// clears and normalizes a data unit. // cleans and normalizes a data unit.
Process(Unit, bool) error Process(Unit, bool) error
} }
@ -16,30 +24,31 @@ func New(
udpMaxPayloadSize int, udpMaxPayloadSize int,
forma formats.Format, forma formats.Format,
generateRTPPackets bool, generateRTPPackets bool,
log logger.Writer,
) (Processor, error) { ) (Processor, error) {
switch forma := forma.(type) { switch forma := forma.(type) {
case *formats.H264: case *formats.H264:
return newH264(udpMaxPayloadSize, forma, generateRTPPackets) return newH264(udpMaxPayloadSize, forma, generateRTPPackets, log)
case *formats.H265: case *formats.H265:
return newH265(udpMaxPayloadSize, forma, generateRTPPackets) return newH265(udpMaxPayloadSize, forma, generateRTPPackets, log)
case *formats.VP8: case *formats.VP8:
return newVP8(udpMaxPayloadSize, forma, generateRTPPackets) return newVP8(udpMaxPayloadSize, forma, generateRTPPackets, log)
case *formats.VP9: case *formats.VP9:
return newVP9(udpMaxPayloadSize, forma, generateRTPPackets) return newVP9(udpMaxPayloadSize, forma, generateRTPPackets, log)
case *formats.MPEG2Audio: case *formats.MPEG2Audio:
return newMPEG2Audio(udpMaxPayloadSize, forma, generateRTPPackets) return newMPEG2Audio(udpMaxPayloadSize, forma, generateRTPPackets, log)
case *formats.MPEG4Audio: case *formats.MPEG4Audio:
return newMPEG4Audio(udpMaxPayloadSize, forma, generateRTPPackets) return newMPEG4Audio(udpMaxPayloadSize, forma, generateRTPPackets, log)
case *formats.Opus: case *formats.Opus:
return newOpus(udpMaxPayloadSize, forma, generateRTPPackets) return newOpus(udpMaxPayloadSize, forma, generateRTPPackets, log)
default: default:
return newGeneric(udpMaxPayloadSize, forma, generateRTPPackets) return newGeneric(udpMaxPayloadSize, forma, generateRTPPackets, log)
} }
} }

7
internal/formatprocessor/vp8.go
Unescape View File

@ -7,6 +7,8 @@ import (
"github.com/bluenviron/gortsplib/v3/pkg/formats" "github.com/bluenviron/gortsplib/v3/pkg/formats"
"github.com/bluenviron/gortsplib/v3/pkg/formats/rtpvp8" "github.com/bluenviron/gortsplib/v3/pkg/formats/rtpvp8"
"github.com/pion/rtp" "github.com/pion/rtp"
"github.com/aler9/mediamtx/internal/logger"
) )
// UnitVP8 is a VP8 data unit. // UnitVP8 is a VP8 data unit.
@ -37,14 +39,15 @@ type formatProcessorVP8 struct {
func newVP8( func newVP8(
udpMaxPayloadSize int, udpMaxPayloadSize int,
forma *formats.VP8, forma *formats.VP8,
allocateEncoder bool, generateRTPPackets bool,
log logger.Writer,
) (*formatProcessorVP8, error) { ) (*formatProcessorVP8, error) {
t := &formatProcessorVP8{ t := &formatProcessorVP8{
udpMaxPayloadSize: udpMaxPayloadSize, udpMaxPayloadSize: udpMaxPayloadSize,
format: forma, format: forma,
} }
if allocateEncoder { if generateRTPPackets {
t.encoder = &rtpvp8.Encoder{ t.encoder = &rtpvp8.Encoder{
PayloadMaxSize: t.udpMaxPayloadSize - 12, PayloadMaxSize: t.udpMaxPayloadSize - 12,
PayloadType: forma.PayloadTyp, PayloadType: forma.PayloadTyp,

7
internal/formatprocessor/vp9.go
Unescape View File

@ -7,6 +7,8 @@ import (
"github.com/bluenviron/gortsplib/v3/pkg/formats" "github.com/bluenviron/gortsplib/v3/pkg/formats"
"github.com/bluenviron/gortsplib/v3/pkg/formats/rtpvp9" "github.com/bluenviron/gortsplib/v3/pkg/formats/rtpvp9"
"github.com/pion/rtp" "github.com/pion/rtp"
"github.com/aler9/mediamtx/internal/logger"
) )
// UnitVP9 is a VP9 data unit. // UnitVP9 is a VP9 data unit.
@ -37,14 +39,15 @@ type formatProcessorVP9 struct {
func newVP9( func newVP9(
udpMaxPayloadSize int, udpMaxPayloadSize int,
forma *formats.VP9, forma *formats.VP9,
allocateEncoder bool, generateRTPPackets bool,
log logger.Writer,
) (*formatProcessorVP9, error) { ) (*formatProcessorVP9, error) {
t := &formatProcessorVP9{ t := &formatProcessorVP9{
udpMaxPayloadSize: udpMaxPayloadSize, udpMaxPayloadSize: udpMaxPayloadSize,
format: forma, format: forma,
} }
if allocateEncoder { if generateRTPPackets {
t.encoder = &rtpvp9.Encoder{ t.encoder = &rtpvp9.Encoder{
PayloadMaxSize: t.udpMaxPayloadSize - 12, PayloadMaxSize: t.udpMaxPayloadSize - 12,
PayloadType: forma.PayloadTyp, PayloadType: forma.PayloadTyp,

20
internal/logger/destination.go
Unescape View File

@ -0,0 +1,20 @@
package logger
// Destination is a log destination.
type Destination int
const (
// DestinationStdout writes logs to the standard output.
DestinationStdout Destination = iota
// DestinationFile writes logs to a file.
DestinationFile
// DestinationSyslog writes logs to the system logger.
DestinationSyslog
)
type destination interface {
log(Level, string, ...interface{})
close()
}

34
internal/logger/destination_file.go
Unescape View File

@ -0,0 +1,34 @@
package logger
import (
"bytes"
"os"
)
type destinationFile struct {
file *os.File
buf bytes.Buffer
}
func newDestinationFile(filePath string) (destination, error) {
f, err := os.OpenFile(filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
if err != nil {
return nil, err
}
return &destinationFile{
file: f,
}, nil
}
func (d *destinationFile) log(level Level, format string, args ...interface{}) {
d.buf.Reset()
writeTime(&d.buf, false)
writeLevel(&d.buf, level, false)
writeContent(&d.buf, format, args)
d.file.Write(d.buf.Bytes())
}
func (d *destinationFile) close() {
d.file.Close()
}

25
internal/logger/destination_stdout.go
Unescape View File

@ -0,0 +1,25 @@
package logger
import (
"bytes"
"os"
)
type destinationStdout struct {
buf bytes.Buffer
}
func newDestionationStdout() destination {
return &destinationStdout{}
}
func (d *destinationStdout) log(level Level, format string, args ...interface{}) {
d.buf.Reset()
writeTime(&d.buf, true)
writeLevel(&d.buf, level, true)
writeContent(&d.buf, format, args)
os.Stdout.Write(d.buf.Bytes())
}
func (d *destinationStdout) close() {
}

34
internal/logger/destination_syslog.go
Unescape View File

@ -0,0 +1,34 @@
package logger
import (
"bytes"
"io"
)
type destinationSysLog struct {
syslog io.WriteCloser
buf bytes.Buffer
}
func newDestinationSyslog() (destination, error) {
syslog, err := newSysLog("mediamtx")
if err != nil {
return nil, err
}
return &destinationSysLog{
syslog: syslog,
}, nil
}
func (d *destinationSysLog) log(level Level, format string, args ...interface{}) {
d.buf.Reset()
writeTime(&d.buf, false)
writeLevel(&d.buf, level, false)
writeContent(&d.buf, format, args)
d.syslog.Write(d.buf.Bytes())
}
func (d *destinationSysLog) close() {
d.syslog.Close()
}

12
internal/logger/level.go
Unescape View File

@ -0,0 +1,12 @@
package logger
// Level is a log level.
type Level int
// Log levels.
const (
Debug Level = iota + 1
Info
Warn
Error
)

108
internal/logger/logger.go
Unescape View File

@ -4,74 +4,46 @@ package logger
import ( import (
"bytes" "bytes"
"fmt" "fmt"
"io"
"os"
"sync" "sync"
"time" "time"
"github.com/gookit/color" "github.com/gookit/color"
) )
// Level is a log level.
type Level int
// Log levels.
const (
Debug Level = iota + 1
Info
Warn
Error
)
// Destination is a log destination.
type Destination int
const (
// DestinationStdout writes logs to the standard output.
DestinationStdout Destination = iota
// DestinationFile writes logs to a file.
DestinationFile
// DestinationSyslog writes logs to the system logger.
DestinationSyslog
)
// Logger is a log handler. // Logger is a log handler.
type Logger struct { type Logger struct {
level Level level Level
destinations map[Destination]struct{}
destinations []destination
mutex sync.Mutex mutex sync.Mutex
file *os.File
syslog io.WriteCloser
stdoutBuffer bytes.Buffer
fileBuffer bytes.Buffer
syslogBuffer bytes.Buffer
} }
// New allocates a log handler. // New allocates a log handler.
func New(level Level, destinations map[Destination]struct{}, filePath string) (*Logger, error) { func New(level Level, destinations []Destination, filePath string) (*Logger, error) {
lh := &Logger{ lh := &Logger{
level: level, level: level,
destinations: destinations,
} }
if _, ok := destinations[DestinationFile]; ok { for _, destType := range destinations {
var err error switch destType {
lh.file, err = os.OpenFile(filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) case DestinationStdout:
if err != nil { lh.destinations = append(lh.destinations, newDestionationStdout())
lh.Close()
return nil, err case DestinationFile:
} dest, err := newDestinationFile(filePath)
} if err != nil {
lh.Close()
if _, ok := destinations[DestinationSyslog]; ok { return nil, err
var err error }
lh.syslog, err = newSyslog("mediamtx") lh.destinations = append(lh.destinations, dest)
if err != nil {
lh.Close() case DestinationSyslog:
return nil, err dest, err := newDestinationSyslog()
if err != nil {
lh.Close()
return nil, err
}
lh.destinations = append(lh.destinations, dest)
} }
} }
@ -80,12 +52,8 @@ func New(level Level, destinations map[Destination]struct{}, filePath string) (*
// Close closes a log handler. // Close closes a log handler.
func (lh *Logger) Close() { func (lh *Logger) Close() {
if lh.file != nil { for _, dest := range lh.destinations {
lh.file.Close() dest.close()
}
if lh.syslog != nil {
lh.syslog.Close()
} }
} }
@ -182,27 +150,7 @@ func (lh *Logger) Log(level Level, format string, args ...interface{}) {
lh.mutex.Lock() lh.mutex.Lock()
defer lh.mutex.Unlock() defer lh.mutex.Unlock()
if _, ok := lh.destinations[DestinationStdout]; ok { for _, dest := range lh.destinations {
lh.stdoutBuffer.Reset() dest.log(level, format, args...)
writeTime(&lh.stdoutBuffer, true)
writeLevel(&lh.stdoutBuffer, level, true)
writeContent(&lh.stdoutBuffer, format, args)
os.Stdout.Write(lh.stdoutBuffer.Bytes())
}
if _, ok := lh.destinations[DestinationFile]; ok {
lh.fileBuffer.Reset()
writeTime(&lh.fileBuffer, false)
writeLevel(&lh.fileBuffer, level, false)
writeContent(&lh.fileBuffer, format, args)
lh.file.Write(lh.fileBuffer.Bytes())
}
if _, ok := lh.destinations[DestinationSyslog]; ok {
lh.syslogBuffer.Reset()
writeTime(&lh.syslogBuffer, false)
writeLevel(&lh.syslogBuffer, level, false)
writeContent(&lh.syslogBuffer, format, args)
lh.syslog.Write(lh.syslogBuffer.Bytes())
} }
} }

2
internal/logger/syslog_unix.go
Unescape View File

@ -12,7 +12,7 @@ type syslog struct {
inner *native.Writer inner *native.Writer
} }
func newSyslog(prefix string) (io.WriteCloser, error) { func newSysLog(prefix string) (io.WriteCloser, error) {
inner, err := native.New(native.LOG_INFO|native.LOG_DAEMON, prefix) inner, err := native.New(native.LOG_INFO|native.LOG_DAEMON, prefix)
if err != nil { if err != nil {
return nil, err return nil, err

2
internal/logger/syslog_win.go
Unescape View File

@ -8,6 +8,6 @@ import (
"io" "io"
) )
func newSyslog(prefix string) (io.WriteCloser, error) { func newSysLog(prefix string) (io.WriteCloser, error) {
return nil, fmt.Errorf("not implemented on windows") return nil, fmt.Errorf("not implemented on windows")
} }

6
internal/logger/writer.go
Unescape View File

@ -0,0 +1,6 @@
package logger
// Writer is an object that provides a log method.
type Writer interface {
Log(Level, string, ...interface{})
}
Write Preview
Loading…
Cancel
Save