diff --git a/machinery/src/capture/main.go b/machinery/src/capture/main.go index 757ea2b..b8b398f 100644 --- a/machinery/src/capture/main.go +++ b/machinery/src/capture/main.go @@ -155,9 +155,9 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat nextPkt, cursorError = recordingCursor.ReadPacket() - now := time.Now().UnixMilli() - hardMaxReached := now-startRecording > maxRecordingPeriod-500 - postRecordingElapsed := startRecording+postRecording-now <= 0 + packetTime := pkt.CurrentTime + hardMaxReached := packetTime-startRecording > maxRecordingPeriod-500 + postRecordingElapsed := startRecording+postRecording-packetTime <= 0 if start && (postRecordingElapsed || hardMaxReached) { rolloverRequested = true if hardMaxReached && !rolloverMaxLogged { @@ -166,18 +166,22 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat } } - if start && // If already recording and current frame is a keyframe and we should stop recording - rolloverRequested && nextPkt.IsKeyFrame { + closeOnCurrentKeyframe := rolloverRequested && pkt.IsKeyFrame && pkt.CurrentTime > startRecording + closeOnNextKeyframe := rolloverRequested && nextPkt.IsKeyFrame - pts := convertPTS(pkt.TimeLegacy) - if pkt.IsVideo { - // Write the last packet - if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil { - log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error()) - } - } else if pkt.IsAudio { - if err := audioWriter.WritePacket(pkt); err != nil { - log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error()) + if start && (closeOnCurrentKeyframe || closeOnNextKeyframe) { + + if !closeOnCurrentKeyframe { + pts := convertPTS(pkt.TimeLegacy) + if pkt.IsVideo { + // Write the last packet before the rollover keyframe. + if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil { + log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error()) + } + } else if pkt.IsAudio { + if err := audioWriter.WritePacket(pkt); err != nil { + log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error()) + } } } @@ -549,18 +553,18 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat log.Log.Error("capture.main.HandleRecordStream(motiondetection): " + cursorError.Error()) } - now = time.Now().UnixMilli() select { case motion := <-communication.HandleMotion: - motionTimestamp = now + motionTimestamp = pkt.CurrentTime log.Log.Info("capture.main.HandleRecordStream(motiondetection): motion detected while recording. Expanding recording.") numberOfChanges := motion.NumberOfChanges log.Log.Info("capture.main.HandleRecordStream(motiondetection): Received message with recording data, detected changes to save: " + strconv.Itoa(numberOfChanges)) default: } - hardMaxReached := now-startRecording > maxRecordingPeriod-500 - postRecordingElapsed := motionTimestamp+postRecording-now < 0 + packetTime := pkt.CurrentTime + hardMaxReached := packetTime-startRecording > maxRecordingPeriod-500 + postRecordingElapsed := motionTimestamp+postRecording-packetTime < 0 if start && (postRecordingElapsed || hardMaxReached) { rolloverRequested = true if hardMaxReached && !rolloverMaxLogged { @@ -569,10 +573,13 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat } } - if start && rolloverRequested && nextPkt.IsKeyFrame { - log.Log.Info("capture.main.HandleRecordStream(motiondetection): timestamp+postRecording-now < 0 - " + strconv.FormatInt(motionTimestamp+postRecording-now, 10) + " < 0") - log.Log.Info("capture.main.HandleRecordStream(motiondetection): now-startRecording > maxRecordingPeriod-500 - " + strconv.FormatInt(now-startRecording, 10) + " > " + strconv.FormatInt(maxRecordingPeriod-500, 10)) - log.Log.Info("capture.main.HandleRecordStream(motiondetection): closing recording (timestamp: " + strconv.FormatInt(motionTimestamp, 10) + ", postRecording: " + strconv.FormatInt(postRecording, 10) + ", now: " + strconv.FormatInt(now, 10) + ", startRecording: " + strconv.FormatInt(startRecording, 10) + ", maxRecordingPeriod: " + strconv.FormatInt(maxRecordingPeriod, 10)) + closeOnCurrentKeyframe := rolloverRequested && pkt.IsKeyFrame && pkt.CurrentTime > startRecording + closeOnNextKeyframe := rolloverRequested && nextPkt.IsKeyFrame + + if start && (closeOnCurrentKeyframe || closeOnNextKeyframe) { + log.Log.Info("capture.main.HandleRecordStream(motiondetection): timestamp+postRecording-packetTime < 0 - " + strconv.FormatInt(motionTimestamp+postRecording-packetTime, 10) + " < 0") + log.Log.Info("capture.main.HandleRecordStream(motiondetection): packetTime-startRecording > maxRecordingPeriod-500 - " + strconv.FormatInt(packetTime-startRecording, 10) + " > " + strconv.FormatInt(maxRecordingPeriod-500, 10)) + log.Log.Info("capture.main.HandleRecordStream(motiondetection): closing recording (timestamp: " + strconv.FormatInt(motionTimestamp, 10) + ", postRecording: " + strconv.FormatInt(postRecording, 10) + ", packetTime: " + strconv.FormatInt(packetTime, 10) + ", startRecording: " + strconv.FormatInt(startRecording, 10) + ", maxRecordingPeriod: " + strconv.FormatInt(maxRecordingPeriod, 10)) break } if pkt.IsKeyFrame && !start && pkt.CurrentTime >= startRecording { diff --git a/machinery/src/capture/pcmu_to_aac_transcoder.go b/machinery/src/capture/pcmu_to_aac_transcoder.go index 98f33cc..50b03b4 100644 --- a/machinery/src/capture/pcmu_to_aac_transcoder.go +++ b/machinery/src/capture/pcmu_to_aac_transcoder.go @@ -44,6 +44,9 @@ type ffmpegToAACTranscoder struct { closed bool stdinClosed bool closeOnce sync.Once + stdoutDone chan struct{} + waitDone chan struct{} + waitErr error } func newFFmpegToAACTranscoder(inputFormat string, sampleRate int, channels int) (*ffmpegToAACTranscoder, error) { @@ -88,13 +91,16 @@ func newFFmpegToAACTranscoder(inputFormat string, sampleRate int, channels int) } t := &ffmpegToAACTranscoder{ - cmd: cmd, - stdin: stdin, - stdout: stdout, - stderr: stderr, + cmd: cmd, + stdin: stdin, + stdout: stdout, + stderr: stderr, + stdoutDone: make(chan struct{}), + waitDone: make(chan struct{}), } go func() { + defer close(t.stdoutDone) buf := make([]byte, 4096) for { n, readErr := stdout.Read(buf) @@ -112,6 +118,11 @@ func newFFmpegToAACTranscoder(inputFormat string, sampleRate int, channels int) } }() + go func() { + t.waitErr = cmd.Wait() + close(t.waitDone) + }() + log.Log.Info("capture.audio_to_aac: " + strings.ToUpper(inputFormat) + " -> AAC transcoder initialised (ffmpeg process)") return t, nil } @@ -161,17 +172,10 @@ func (t *ffmpegToAACTranscoder) Transcode(input []byte) ([]byte, error) { return nil, err } - deadline := time.Now().Add(75 * time.Millisecond) - for { - data := t.readAvailable() - if len(data) > 0 { - return data, nil - } - if time.Now().After(deadline) { - return nil, nil - } - time.Sleep(5 * time.Millisecond) - } + // Do not block the recording loop waiting for the encoder to emit output. + // FFmpeg can buffer for a while, and polling here per RTP packet causes the + // recorder to fall behind real time and drop trailing media before close. + return t.readAvailable(), nil } func (t *ffmpegToAACTranscoder) Flush() ([]byte, error) { @@ -193,25 +197,39 @@ func (t *ffmpegToAACTranscoder) Flush() ([]byte, error) { } t.mu.Unlock() - deadline := time.Now().Add(750 * time.Millisecond) - previousLen := -1 - stableReads := 0 - for { - buffered := t.bufferedLen() - if buffered == previousLen { - stableReads++ - } else { - stableReads = 0 - previousLen = buffered + processExited := false + readerFinished := false + deadline := time.Now().Add(3 * time.Second) + for !processExited || !readerFinished { + if !processExited { + select { + case <-t.waitDone: + processExited = true + default: + } + } + if !readerFinished { + select { + case <-t.stdoutDone: + readerFinished = true + default: + } } - if stableReads >= 3 || time.Now().After(deadline) { + if processExited && readerFinished { + break + } + if time.Now().After(deadline) { break } time.Sleep(15 * time.Millisecond) } + if processExited && t.waitErr != nil { + return t.readAvailable(), t.waitErr + } + return t.readAvailable(), nil } @@ -229,13 +247,21 @@ func (t *ffmpegToAACTranscoder) Close() { } t.mu.Unlock() - if t.stdout != nil { - _ = t.stdout.Close() + processExited := false + select { + case <-t.waitDone: + processExited = true + case <-time.After(250 * time.Millisecond): } - if t.cmd != nil && t.cmd.Process != nil { - _ = t.cmd.Process.Kill() - _, _ = t.cmd.Process.Wait() + if !processExited { + if t.stdout != nil { + _ = t.stdout.Close() + } + if t.cmd != nil && t.cmd.Process != nil { + _ = t.cmd.Process.Kill() + <-t.waitDone + } } if stderr := t.stderrString(); stderr != "" {