Merged in give-encoder-log (pull request #405)

container/mts: doing logging in MTS encoder

Approved-by: Trek Hopton <trek.hopton@gmail.com>

container/mts/encoder.go

@@ -47,6 +47,14 @@ const (
 	audioStreamID = 0xc0 // ADPCM audio stream ID.
 )
 
+// These three constants are used to select between the three different
+// methods of when the PSI is sent.
+const (
+	pktBased = iota
+	timeBased
+	nalBased
+)
+
 // Constants used to communicate which media codec will be packetized.
 const (
 	EncodeH264 = iota
@@ -142,6 +150,14 @@ var (
 	pmtTable []byte
 )
 
+type logger interface {
+	Debug(string, ...interface{})
+	Info(string, ...interface{})
+	Warning(string, ...interface{})
+	Error(string, ...interface{})
+	Fatal(string, ...interface{})
+}
+
 // Encoder encapsulates properties of an MPEG-TS generator.
 type Encoder struct {
 	dst io.WriteCloser
@@ -163,11 +179,14 @@ type Encoder struct {
 	startTime    time.Time
 	mediaPid     uint16
 	streamID     byte
+
+	// log is a function that will be used through the encoder code for logging.
+	log logger
 }
 
 // NewEncoder returns an Encoder with the specified media type and rate eg. if a video stream
 // calls write for every frame, the rate will be the frame rate of the video.
-func NewEncoder(dst io.WriteCloser, rate float64, mediaType int, options ...func(*Encoder) error) (*Encoder, error) {
+func NewEncoder(dst io.WriteCloser, rate float64, mediaType int, log logger, options ...func(*Encoder) error) (*Encoder, error) {
 	var mPID uint16
 	var sID byte
 	psiM := timeBased
@@ -176,18 +195,22 @@ func NewEncoder(dst io.WriteCloser, rate float64, mediaType int, options ...func
 		mPID = AudioPid
 		sID = audioStreamID
 		psiM = pktBased
+		log.Debug("configured for audio packetisation")
 	case EncodeH265:
 		mPID = VideoPid
 		sID = H265ID
 		psiM = nalBased
+		log.Debug("configured for h.265 packetisation")
 	case EncodeH264:
 		mPID = VideoPid
 		sID = H264ID
 		psiM = nalBased
+		log.Debug("configured for h.264 packetisation")
 	case EncodeMJPEG:
 		mPID = VideoPid
 		sID = MJPEGID
 		psiM = timeBased
+		log.Debug("configured for MJPEG packetisation")
 	}
 
 	pmt := BasePMT
@@ -204,22 +227,14 @@ func NewEncoder(dst io.WriteCloser, rate float64, mediaType int, options ...func
 
 	e := &Encoder{
 		dst:         dst,
-
 		writePeriod: time.Duration(float64(time.Second) / rate),
 		ptsOffset:   ptsOffset,
-
 		psiMethod:   psiM,
-
 		pktCount:    8,
-
 		mediaPid:    mPID,
 		streamID:    sID,
-
+		continuity:  map[uint16]byte{PatPid: 0, PmtPid: 0, mPID: 0},
-		continuity: map[uint16]byte{
+		log:         log,
-			PatPid: 0,
-			PmtPid: 0,
-			mPID:   0,
-		},
 	}
 
 	for _, option := range options {
@@ -228,17 +243,11 @@ func NewEncoder(dst io.WriteCloser, rate float64, mediaType int, options ...func
 			return nil, fmt.Errorf("option failed with error: %w", err)
 		}
 	}
+	log.Debug("encoder options applied")
+
 	return e, nil
 }
 
-// These three constants are used to select between the three different
-// methods of when the PSI is sent.
-const (
-	pktBased = iota
-	timeBased
-	nalBased
-)
-
 // PacketBasedPSI is an option that can be passed to NewEncoder to select
 // packet based PSI writing, i.e. PSI are written to the destination every
 // sendCount packets.
@@ -247,6 +256,7 @@ func PacketBasedPSI(sendCount int) func(*Encoder) error {
 		e.psiMethod = pktBased
 		e.psiSendCount = sendCount
 		e.pktCount = e.psiSendCount
+		e.log.Debug("configured for packet based PSI insertion", "count", sendCount)
 		return nil
 	}
 }
@@ -260,6 +270,7 @@ func TimeBasedPSI(dur time.Duration) func(*Encoder) error {
 		e.psiTime = 0
 		e.psiSetTime = dur
 		e.startTime = time.Now()
+		e.log.Debug("configured for time based PSI insertion")
 		return nil
 	}
 }
@@ -269,6 +280,7 @@ func TimeBasedPSI(dur time.Duration) func(*Encoder) error {
 func (e *Encoder) Write(data []byte) (int, error) {
 	switch e.psiMethod {
 	case pktBased:
+		e.log.Debug("checking packet no. conditions for PSI write", "count", e.pktCount, "PSI count", e.psiSendCount)
 		if e.pktCount >= e.psiSendCount {
 			e.pktCount = 0
 			err := e.writePSI()
@@ -281,7 +293,7 @@ func (e *Encoder) Write(data []byte) (int, error) {
 		if err != nil {
 			return 0, fmt.Errorf("could not get type from NAL unit, failed with error: %w", err)
 		}
-
+		e.log.Debug("checking conditions for PSI write", "AU type", nalType, "needed type", h264dec.NALTypeSPS)
 		if nalType == h264dec.NALTypeSPS {
 			err := e.writePSI()
 			if err != nil {
@@ -289,24 +301,26 @@ func (e *Encoder) Write(data []byte) (int, error) {
 			}
 		}
 	case timeBased:
-		if time.Now().Sub(e.startTime) >= e.psiTime {
+		dur := time.Now().Sub(e.startTime)
+		e.log.Debug("checking time conditions for PSI write")
+		if dur >= e.psiTime {
 			e.psiTime = e.psiSetTime
 			e.startTime = time.Now()
 			err := e.writePSI()
 			if err != nil {
 				return 0, err
 			}
-
 		}
 	default:
-		panic("Undefined PSI method")
+		panic("undefined PSI method")
 	}
 
 	// Prepare PES data.
+	pts := e.pts()
 	pesPkt := pes.Packet{
 		StreamID:     e.streamID,
 		PDI:          hasPTS,
-		PTS:          e.pts(),
+		PTS:          pts,
 		Data:         data,
 		HeaderLength: 5,
 	}
@@ -329,7 +343,9 @@ func (e *Encoder) Write(data []byte) (int, error) {
 		if pusi {
 			// If the packet has a Payload Unit Start Indicator
 			// flag set then we need to write a PCR.
-			pkt.PCR = e.pcr()
+			pcr := e.pcr()
+			e.log.Debug("new access unit", "PCR", pcr, "PTS", pts)
+			pkt.PCR = pcr
 			pusi = false
 		}
 		_, err := e.dst.Write(pkt.Bytes(e.tsSpace[:PacketSize]))
@@ -360,7 +376,8 @@ func (e *Encoder) writePSI() error {
 		return err
 	}
 	e.pktCount++
-	pmtTable, err = updateMeta(pmtTable)
+
+	pmtTable, err = updateMeta(pmtTable, e.log)
 
 	if err != nil {
 		return err
@@ -379,6 +396,8 @@ func (e *Encoder) writePSI() error {
 		return err
 	}
 	e.pktCount++
+
+	e.log.Debug("PSI written", "PAT CC", patPkt.CC, "PMT CC", pmtPkt.CC)
 	return nil
 }
 
@@ -407,15 +426,18 @@ func (e *Encoder) ccFor(pid uint16) byte {
 
 // updateMeta adds/updates a metaData descriptor in the given psi bytes using data
 // contained in the global Meta struct.
-func updateMeta(b []byte) ([]byte, error) {
+func updateMeta(b []byte, log logger) ([]byte, error) {
 	p := psi.PSIBytes(b)
 	if RealTime.IsSet() {
-		Meta.Add("ts", strconv.Itoa(int(RealTime.Get().Unix())))
+		t := strconv.Itoa(int(RealTime.Get().Unix()))
+		Meta.Add("ts", t)
+		log.Debug("latest time added to meta", "time", t)
 	}
 	err := p.AddDescriptor(psi.MetadataTag, Meta.Encode())
 	return []byte(p), err
 }
 
 func (e *Encoder) Close() error {
+	e.log.Debug("closing encoder")
 	return e.dst.Close()
 }

container/mts/encoder_test.go

@@ -54,6 +54,32 @@ func (d *destination) Write(p []byte) (int, error) {
 	return len(p), nil
 }
 
+// testLogger will allow logging to be done by the testing pkg.
+type testLogger testing.T
+
+func (tl *testLogger) Debug(msg string, args ...interface{})   { tl.log("debug", msg, args...) }
+func (tl *testLogger) Info(msg string, args ...interface{})    { tl.log("info", msg, args...) }
+func (tl *testLogger) Warning(msg string, args ...interface{}) { tl.log("warning", msg, args...) }
+func (tl *testLogger) Error(msg string, args ...interface{})   { tl.log("error", msg, args...) }
+func (tl *testLogger) Fatal(msg string, args ...interface{})   { tl.log("fatal", msg, args...) }
+
+func (tl *testLogger) log(lvl string, msg string, args ...interface{}) {
+	switch lvl {
+	case "debug", "info", "warning", "error", "fatal":
+	default:
+		panic("invalid log level")
+	}
+	msg = lvl + ": " + msg
+	for i := 0; i < len(args); i++ {
+		msg += " %v"
+	}
+	if len(args) == 0 {
+		tl.Log(msg + "\n")
+		return
+	}
+	tl.Logf(msg+"\n", args)
+}
+
 // TestEncodeVideo checks that we can correctly encode some dummy data into a
 // valid MPEG-TS stream. This checks for correct MPEG-TS headers and also that the
 // original data is stored correctly and is retreivable.
@@ -101,7 +127,7 @@ func TestEncodeVideo(t *testing.T) {
 
 	// Create the dst and write the test data to encoder.
 	dst := &destination{}
-	e, err := NewEncoder(nopCloser{dst}, 25, EncodeH264, PacketBasedPSI(psiSendCount))
+	e, err := NewEncoder(nopCloser{dst}, 25, EncodeH264, (*testLogger)(t), PacketBasedPSI(psiSendCount))
 	if err != nil {
 		t.Fatalf("could not create MTS encoder, failed with error: %v", err)
 	}
@@ -165,7 +191,7 @@ func TestEncodePcm(t *testing.T) {
 	sampleSize := 2
 	blockSize := 16000
 	writeFreq := float64(sampleRate*sampleSize) / float64(blockSize)
-	e, err := NewEncoder(nopCloser{&buf}, writeFreq, EncodeAudio)
+	e, err := NewEncoder(nopCloser{&buf}, writeFreq, EncodeAudio, (*testLogger)(t))
 	if err != nil {
 		t.Fatalf("could not create MTS encoder, failed with error: %v", err)
 	}
@@ -269,7 +295,7 @@ const fps = 25
 func TestMetaEncode1(t *testing.T) {
 	Meta = meta.New()
 	var buf bytes.Buffer
-	e, err := NewEncoder(nopCloser{&buf}, fps, EncodeH264)
+	e, err := NewEncoder(nopCloser{&buf}, fps, EncodeH264, (*testLogger)(t))
 	if err != nil {
 		t.Fatalf("could not create encoder, failed with error: %v", err)
 	}
@@ -301,7 +327,7 @@ func TestMetaEncode1(t *testing.T) {
 func TestMetaEncode2(t *testing.T) {
 	Meta = meta.New()
 	var buf bytes.Buffer
-	e, err := NewEncoder(nopCloser{&buf}, fps, EncodeH264)
+	e, err := NewEncoder(nopCloser{&buf}, fps, EncodeH264, (*testLogger)(t))
 	if err != nil {
 		t.Fatalf("could not create MTS encoder, failed with error: %v", err)
 	}
@@ -333,7 +359,7 @@ func TestMetaEncode2(t *testing.T) {
 func TestExtractMeta(t *testing.T) {
 	Meta = meta.New()
 	var buf bytes.Buffer
-	e, err := NewEncoder(nopCloser{&buf}, fps, EncodeH264)
+	e, err := NewEncoder(nopCloser{&buf}, fps, EncodeH264, (*testLogger)(t))
 	if err != nil {
 		t.Fatalf("could not create MTS encoder, failed with error: %v", err)
 	}

container/mts/mpegts_test.go

@@ -365,7 +365,7 @@ func TestTrimToMetaRange(t *testing.T) {
 
 	for i := 0; i < nPSI; i++ {
 		Meta.Add(key, strconv.Itoa((i*2)+1))
-		err := writePSIWithMeta(&clip)
+		err := writePSIWithMeta(&clip, t)
 		if err != nil {
 			t.Fatalf("did not expect to get error writing PSI, error: %v", err)
 		}
@@ -472,7 +472,7 @@ func TestSegmentForMeta(t *testing.T) {
 			} else {
 				Meta.Delete(key)
 			}
-			err := writePSIWithMeta(&clip)
+			err := writePSIWithMeta(&clip, t)
 			if err != nil {
 				t.Fatalf("did not expect to get error writing PSI, error: %v", err)
 			}
@@ -668,7 +668,7 @@ func TestFindPSI(t *testing.T) {
 				}).Bytes()
 
 				Meta.Add(metaKey, test.meta)
-				pmtTable, err = updateMeta(pmtTable)
+				pmtTable, err = updateMeta(pmtTable, (*testLogger)(t))
 				if err != nil {
 					t.Fatalf("could not update meta for test %d", i)
 				}

container/mts/payload_test.go

@@ -70,7 +70,7 @@ func TestExtract(t *testing.T) {
 			// We'll add the frame number as meta.
 			Meta.Add("frameNum", strconv.Itoa(i))
 
-			err = writePSIWithMeta(&clip)
+			err = writePSIWithMeta(&clip, t)
 			if err != nil {
 				t.Fatalf("did not expect error writing psi: %v", err)
 			}
@@ -136,7 +136,7 @@ func TestExtract(t *testing.T) {
 }
 
 // writePSIWithMeta writes PSI to b with updated metadata.
-func writePSIWithMeta(b *bytes.Buffer) error {
+func writePSIWithMeta(b *bytes.Buffer, t *testing.T) error {
 	// Write PAT.
 	pat := Packet{
 		PUSI:    true,
@@ -151,7 +151,7 @@ func writePSIWithMeta(b *bytes.Buffer) error {
 	}
 
 	// Update the meta in the pmt table.
-	pmtTable, err = updateMeta(pmtTable)
+	pmtTable, err = updateMeta(pmtTable, (*testLogger)(t))
 	if err != nil {
 		return err
 	}
@@ -202,7 +202,7 @@ func TestClipBytes(t *testing.T) {
 			// We'll add the frame number as meta.
 			Meta.Add("frameNum", strconv.Itoa(i))
 
-			err = writePSIWithMeta(&clip)
+			err = writePSIWithMeta(&clip, t)
 			if err != nil {
 				t.Fatalf("did not expect error writing psi: %v", err)
 			}

revid/revid.go

@@ -74,6 +74,14 @@ type Logger interface {
 	Log(level int8, message string, params ...interface{})
 }
 
+type encLog struct{ Logger }
+
+func (el *encLog) Debug(msg string, args ...interface{})   { el.Log(logger.Debug, msg, args...) }
+func (el *encLog) Info(msg string, args ...interface{})    { el.Log(logger.Info, msg, args...) }
+func (el *encLog) Warning(msg string, args ...interface{}) { el.Log(logger.Warning, msg, args...) }
+func (el *encLog) Error(msg string, args ...interface{})   { el.Log(logger.Error, msg, args...) }
+func (el *encLog) Fatal(msg string, args ...interface{})   { el.Log(logger.Fatal, msg, args...) }
+
 // Revid provides methods to control a revid session; providing methods
 // to start, stop and change the state of an instance using the Config struct.
 type Revid struct {
@@ -117,7 +125,7 @@ type Revid struct {
 	// bitrate is used for bitrate calculations.
 	bitrate bitrate.Calculator
 
-	// stop used used to signal stopping when looping an input.
+	// stop is used to signal stopping when looping an input.
 	stop chan struct{}
 }
 
@@ -214,7 +222,7 @@ func (r *Revid) reset(c config.Config) error {
 				panic("unknown input type")
 			}
 
-			return mts.NewEncoder(dst, float64(fps), st, encOptions...)
+			return mts.NewEncoder(dst, float64(fps), st, &encLog{r.cfg.Logger}, encOptions...)
 		},
 		func(dst io.WriteCloser, fps int) (io.WriteCloser, error) {
 			return flv.NewEncoder(dst, true, true, fps)

revid/revid_test.go

@@ -51,7 +51,7 @@ func TestRaspivid(t *testing.T) {
 		t.Skip("Skipping TestRaspivid since no raspivid found.")
 	}
 
-	var logger testLogger
+	var logger simpleLogger
 	ns, err := netsender.New(&logger, nil, nil, nil)
 	if err != nil {
 		t.Errorf("netsender.New failed with error %v", err)
@@ -72,12 +72,12 @@ func TestRaspivid(t *testing.T) {
 	}
 }
 
-// testLogger implements a netsender.Logger.
+// simpleLogger implements a netsender.Logger.
-type testLogger struct{}
+type simpleLogger struct{}
 
-func (tl *testLogger) SetLevel(level int8) {}
+func (tl *simpleLogger) SetLevel(level int8) {}
 
-func (tl *testLogger) Log(level int8, msg string, params ...interface{}) {
+func (tl *simpleLogger) Log(level int8, msg string, params ...interface{}) {
 	logLevels := [...]string{"Debug", "Info", "Warn", "Error", "", "", "Fatal"}
 	if level < -1 || level > 5 {
 		panic("Invalid log level")
@@ -216,7 +216,7 @@ func TestResetEncoderSenderSetup(t *testing.T) {
 		},
 	}
 
-	rv, err := New(config.Config{Logger: &testLogger{}}, nil)
+	rv, err := New(config.Config{Logger: &simpleLogger{}}, nil)
 	if err != nil {
 		t.Fatalf("unexpected err: %v", err)
 	}
@@ -225,7 +225,7 @@ func TestResetEncoderSenderSetup(t *testing.T) {
 	for testNum, test := range tests {
 		// Create a new config and reset revid with it.
 		const dummyURL = "rtmp://dummy"
-		c := config.Config{Logger: &testLogger{}, Outputs: test.outputs, RTMPURL: dummyURL}
+		c := config.Config{Logger: &simpleLogger{}, Outputs: test.outputs, RTMPURL: dummyURL}
 		err := rv.setConfig(c)
 		if err != nil {
 			t.Fatalf("unexpected error: %v for test %v", err, testNum)

revid/senders_test.go

@@ -98,10 +98,16 @@ func (ts *destination) Write(d []byte) (int, error) {
 
 func (ts *destination) Close() error { return nil }
 
-// dummyLogger will allow logging to be done by the testing pkg.
+// testLogger will allow logging to be done by the testing pkg.
-type dummyLogger testing.T
+type testLogger testing.T
 
-func (dl *dummyLogger) log(lvl int8, msg string, args ...interface{}) {
+func (tl *testLogger) Debug(msg string, args ...interface{})   { tl.log(logger.Debug, msg, args...) }
+func (tl *testLogger) Info(msg string, args ...interface{})    { tl.log(logger.Info, msg, args...) }
+func (tl *testLogger) Warning(msg string, args ...interface{}) { tl.log(logger.Warning, msg, args...) }
+func (tl *testLogger) Error(msg string, args ...interface{})   { tl.log(logger.Error, msg, args...) }
+func (tl *testLogger) Fatal(msg string, args ...interface{})   { tl.log(logger.Fatal, msg, args...) }
+
+func (dl *testLogger) log(lvl int8, msg string, args ...interface{}) {
 	var l string
 	switch lvl {
 	case logger.Warning:
@@ -136,10 +142,10 @@ func TestMTSSenderSegment(t *testing.T) {
 	dst := &destination{t: t, done: make(chan struct{}), doneAt: numberOfClips}
 	const testRBCapacity = 50000000
 	nElements := testRBCapacity / rbStartingElementSize
-	sender := newMTSSender(dst, (*dummyLogger)(t).log, ring.NewBuffer(nElements, rbStartingElementSize, 0), 0)
+	sender := newMTSSender(dst, (*testLogger)(t).log, ring.NewBuffer(nElements, rbStartingElementSize, 0), 0)
 
 	const psiSendCount = 10
-	encoder, err := mts.NewEncoder(sender, 25, mts.EncodeH264, mts.PacketBasedPSI(psiSendCount))
+	encoder, err := mts.NewEncoder(sender, 25, mts.EncodeH264, (*testLogger)(t), mts.PacketBasedPSI(psiSendCount))
 	if err != nil {
 		t.Fatalf("could not create MTS encoder, failed with error: %v", err)
 	}
@@ -217,10 +223,10 @@ func TestMtsSenderFailedSend(t *testing.T) {
 	dst := &destination{t: t, testFails: true, failAt: clipToFailAt, done: make(chan struct{})}
 	const testRBCapacity = 50000000 // 50MB
 	nElements := testRBCapacity / rbStartingElementSize
-	sender := newMTSSender(dst, (*dummyLogger)(t).log, ring.NewBuffer(nElements, rbStartingElementSize, 0), 0)
+	sender := newMTSSender(dst, (*testLogger)(t).log, ring.NewBuffer(nElements, rbStartingElementSize, 0), 0)
 
 	const psiSendCount = 10
-	encoder, err := mts.NewEncoder(sender, 25, mts.EncodeH264, mts.PacketBasedPSI(psiSendCount))
+	encoder, err := mts.NewEncoder(sender, 25, mts.EncodeH264, (*testLogger)(t), mts.PacketBasedPSI(psiSendCount))
 	if err != nil {
 		t.Fatalf("could not create MTS encoder, failed with error: %v", err)
 	}
@@ -298,10 +304,10 @@ func TestMtsSenderDiscontinuity(t *testing.T) {
 	// Create destination, the mtsSender and the mtsEncoder.
 	const clipToDelay = 3
 	dst := &destination{t: t, sendDelay: 10 * time.Millisecond, delayAt: clipToDelay, done: make(chan struct{})}
-	sender := newMTSSender(dst, (*dummyLogger)(t).log, ring.NewBuffer(1, rbStartingElementSize, 0), 0)
+	sender := newMTSSender(dst, (*testLogger)(t).log, ring.NewBuffer(1, rbStartingElementSize, 0), 0)
 
 	const psiSendCount = 10
-	encoder, err := mts.NewEncoder(sender, 25, mts.EncodeH264, mts.PacketBasedPSI(psiSendCount))
+	encoder, err := mts.NewEncoder(sender, 25, mts.EncodeH264, (*testLogger)(t), mts.PacketBasedPSI(psiSendCount))
 	if err != nil {
 		t.Fatalf("could not create MTS encoder, failed with error: %v", err)
 	}