refactor: improve audio capturer buffer management

src/capturer/pa_capturer.cpp

@@ -33,8 +33,8 @@ bool PaCapturer::CreateFloat32Source() {
     ss.rate = sample_rate_;
 
     // Set fragsize so the PulseAudio server delivers data in exact 10ms fragments.
-    const uint32_t frag_bytes =
-        static_cast<uint32_t>(frames_per_chunk()) * channels_ * sizeof(float);
+    const uint32_t frag_bytes = static_cast<uint32_t>(
+        static_cast<size_t>(frames_per_chunk()) * static_cast<size_t>(channels_) * sizeof(float));
     pa_buffer_attr attr{};
     attr.maxlength = static_cast<uint32_t>(-1);
     attr.fragsize = frag_bytes;
@@ -50,8 +50,9 @@ bool PaCapturer::CreateFloat32Source() {
     }
 
     // Pre-allocate the capture buffer for exactly 10ms of stereo float32 audio.
-    const size_t n_samples = static_cast<size_t>(frames_per_chunk()) * channels_;
-    capture_buf_.resize(n_samples * sizeof(float));
+    const size_t n_samples =
+        static_cast<size_t>(frames_per_chunk()) * static_cast<size_t>(channels_);
+    capture_buf_.resize(n_samples);
 
     INFO_PRINT("PulseAudio capture format: FLOAT32LE, %d channels, %d Hz", channels_, sample_rate_);
 
@@ -67,16 +68,16 @@ void PaCapturer::CaptureSamples() {
     int error;
     // Read exactly 10ms of audio per call so all consumers (WebRTC, recorder)
     const size_t n_frames = static_cast<size_t>(frames_per_chunk());
-    const size_t n_samples = n_frames * channels_; // interleaved float32 count
+    const size_t n_samples = n_frames * static_cast<size_t>(channels_); // interleaved float32 count
 
-    if (pa_simple_read(src, capture_buf_.data(), capture_buf_.size(), &error) < 0) {
+    if (pa_simple_read(src, capture_buf_.data(), capture_buf_.size() * sizeof(float), &error) < 0) {
         ERROR_PRINT("pa_simple_read() failed: %s", pa_strerror(error));
         pa_simple_free(src);
         src = nullptr;
         return;
     }
 
-    shared_buffer_ = {.start = capture_buf_.data(),
+    shared_buffer_ = {.start = reinterpret_cast<uint8_t *>(capture_buf_.data()),
                       .length = static_cast<unsigned int>(n_samples),
                       .channels = static_cast<unsigned int>(channels_)};
     Next(shared_buffer_);

src/capturer/pa_capturer.h

@@ -21,7 +21,7 @@ class PaCapturer : public AudioCapturer {
 
   private:
     pa_simple *src;
-    std::vector<uint8_t> capture_buf_;
+    std::vector<float> capture_buf_;
 
     void CaptureSamples();
     bool CreateFloat32Source();

src/main.cpp

@@ -14,7 +14,7 @@ int main(int argc, char *argv[]) {
 
     std::shared_ptr<Conductor> conductor = Conductor::Create(args);
     std::unique_ptr<RecorderManager> bg_recorder_mgr;
-    std::unique_ptr<RecorderManager> ondemand_recorder_mgr;
+    std::shared_ptr<RecorderManager> ondemand_recorder_mgr;
 
     // Background recorder
     if ((args.record_mode == RecordMode::Background || args.record_mode == -1) &&
@@ -31,7 +31,7 @@ int main(int argc, char *argv[]) {
         if (Utils::CreateFolder(ondemand_args.record_path)) {
             ondemand_recorder_mgr = RecorderManager::Create(
                 conductor->VideoSource(), conductor->AudioSource(), ondemand_args, false);
-            conductor->SetOnDemandRecorder(ondemand_recorder_mgr.get());
+            conductor->SetOnDemandRecorder(ondemand_recorder_mgr);
             DEBUG_PRINT("On-demand recorder is ready.");
         }
     }

src/recorder/recorder_manager.cpp

@@ -135,43 +135,36 @@ RecorderManager::RecorderManager(Args config)
       auto_start_(true),
       header_written_(false),
       has_first_keyframe(false),
-      time_reset_pending_(false),
-      record_path(config.record_path),
-      elapsed_time_(0.0) {}
+      record_path(config.record_path) {}
 
 void RecorderManager::SubscribeVideoSource(std::shared_ptr<VideoCapturer> video_src) {
     video_subscription_ = video_src->Subscribe(
         [this](V4L2FrameBufferRef buffer) {
-            // waiting first keyframe to start recorders.
-            if (auto_start_ && !has_first_keyframe &&
-                ((buffer->flags() & V4L2_BUF_FLAG_KEYFRAME) ||
-                 video_src_->format() != V4L2_PIX_FMT_H264)) {
-                Start();
-                last_created_time_ = buffer->timestamp();
-            }
+            bool is_keyframe = (buffer->flags() & V4L2_BUF_FLAG_KEYFRAME) ||
+                               (video_src_->format() != V4L2_PIX_FMT_H264);
 
-            // restart to write in the new file on the next keyframe boundary.
-            if (has_first_keyframe && elapsed_time_ >= config.file_duration &&
-                ((buffer->flags() & V4L2_BUF_FLAG_KEYFRAME) ||
-                 video_src_->format() != V4L2_PIX_FMT_H264)) {
-                Stop();
+            // waiting first keyframe to start recorders.
+            if (auto_start_ && !has_first_keyframe && is_keyframe) {
                 Start();
+                base_start_time_ = buffer->timestamp();
+                next_generate_time_ = ++file_index_ * config.file_duration;
             }
 
-            if (has_first_keyframe && video_recorder) {
-                video_recorder->OnBuffer(buffer);
-            }
-
-            // Sync last_created_time_ to V4L2 time base on first callback after Start()
-            if (time_reset_pending_) {
-                last_created_time_ = buffer->timestamp();
-                elapsed_time_ = 0.0;
-                time_reset_pending_ = false;
-            } else {
-                int64_t elapsed_us =
-                    (int64_t)(buffer->timestamp().tv_sec - last_created_time_.tv_sec) * 1000000LL +
-                    (int64_t)(buffer->timestamp().tv_usec - last_created_time_.tv_usec);
-                elapsed_time_ = elapsed_us / 1e6;
+            int64_t total_elapsed_us =
+                (int64_t)(buffer->timestamp().tv_sec - base_start_time_.tv_sec) * 1000000LL +
+                (int64_t)(buffer->timestamp().tv_usec - base_start_time_.tv_usec);
+            double total_elapsed_time = total_elapsed_us / 1e6;
+
+            if (has_first_keyframe) {
+                if (total_elapsed_time >= next_generate_time_ && is_keyframe) {
+                    Stop();
+                    Start();
+                    next_generate_time_ = ++file_index_ * config.file_duration;
+                }
+
+                if (video_recorder) {
+                    video_recorder->OnBuffer(buffer);
+                }
             }
         },
         config.record_stream_idx);
@@ -283,8 +276,6 @@ void RecorderManager::Start() {
     }
 
     has_first_keyframe = true;
-    elapsed_time_ = 0.0;
-    time_reset_pending_ = true;
 }
 
 void RecorderManager::Stop() {

src/recorder/recorder_manager.h

@@ -54,16 +54,17 @@ class RecorderManager {
     void SubscribeAudioSource(std::shared_ptr<AudioCapturer> audio_src);
 
   private:
-    double elapsed_time_;
     bool auto_start_;
+    int file_index_ = 0;
+    double next_generate_time_;
     std::atomic<bool> header_written_;
     std::atomic<bool> time_reset_pending_;
     std::mutex rotation_mtx_;
     std::condition_variable rotation_cv_;
     std::atomic<bool> rotation_abort_;
     std::atomic<bool> rotation_requested_;
     std::thread rotation_thread_;
-    struct timeval last_created_time_;
+    struct timeval base_start_time_;
     std::shared_ptr<VideoCapturer> video_src_;
 
     std::string current_filepath_;

src/rtc/audio_device_bridge.h

@@ -1,6 +1,7 @@
 #ifndef AUDIO_DEVICE_BRIDGE_H_
 #define AUDIO_DEVICE_BRIDGE_H_
 
+#include <algorithm>
 #include <atomic>
 #include <memory>
 #include <vector>
@@ -52,6 +53,7 @@ class AudioDeviceBridge : public webrtc::AudioDeviceModule {
 
     int32_t Terminate() override {
         subscription_ = {};
+        audio_transport_.store(nullptr, std::memory_order_release);
         initialized_.store(false);
         return 0;
     }
@@ -179,6 +181,10 @@ class AudioDeviceBridge : public webrtc::AudioDeviceModule {
 
   private:
     void OnAudioBuffer(const AudioBuffer &buf) {
+        if (!initialized_.load(std::memory_order_acquire) ||
+            !recording_.load(std::memory_order_acquire))
+            return;
+
         auto *transport = audio_transport_.load(std::memory_order_acquire);
         if (!transport)
             return;

src/rtc/conductor.cpp

@@ -79,9 +79,9 @@ void Conductor::InitializeTracks() {
             }
         })();
 
-        if (adm_ && audio_capture_source_) {
-            adm_->SetCapturer(audio_capture_source_);
-        } else {
+        if (!audio_capture_source_) {
+            ERROR_PRINT("Audio capturer failed to initialize; skipping audio track creation.");
+        } else if (!adm_) {
             ERROR_PRINT("Audio device module is not initialized; cannot set audio capturer.");
         }
 
@@ -261,7 +261,7 @@ void Conductor::InitializeCommandChannel(webrtc::scoped_refptr<RtcPeer> peer) {
         });
 
     cmd_channel->OnClosed([this]() {
-        auto recorder = ondemand_recorder_;
+        auto recorder = ondemand_recorder_.lock();
         if (recorder && recorder->is_recording()) {
             DEBUG_PRINT("Peer disconnected: Auto-stop on-demand recording when peer disconnects "
                         "(kFailed / kClosed)");
@@ -297,30 +297,29 @@ void Conductor::QueryFile(std::shared_ptr<RtcChannel> datachannel, const protoco
 
     auto req = pkt.query_file_request();
     auto type = req.type();
-    auto base_dir = (req.mode() == protocol::VideoMode::TIMELAPSE) ? args.record_path + "timelapse"
-                                                                   : args.record_path;
+    const bool is_timelapse = (req.mode() == protocol::VideoMode::TIMELAPSE);
     const std::string &parameter = req.parameter();
+    auto search_dir = is_timelapse ? args.record_path + "timelapse" : args.record_path;
 
     DEBUG_PRINT("Received query request: mode=%s, type=%d, param=%s",
-                (req.mode() == protocol::VideoMode::TIMELAPSE ? "TIMELAPSE" : "RECORDING"),
-                req.type(), parameter.c_str());
+                (is_timelapse ? "TIMELAPSE" : "RECORDING"), req.type(), parameter.c_str());
 
     if (type == protocol::QueryFileType::LATEST_FILE || parameter.empty()) {
-        auto path = Utils::FindLatestCompleteFile(base_dir, ".mp4");
+        auto path = Utils::FindLatestCompleteFile(search_dir, ".mp4");
         DEBUG_PRINT("LATEST: %s", path.c_str());
         SendFileResponse(datachannel, path, req.mode());
     } else if (type == protocol::QueryFileType::BEFORE_FILE) {
-        auto paths = Utils::FindOlderFiles(base_dir, parameter, 8);
-        if (paths.empty()) {
-            SendFileResponse(datachannel, "", req.mode());
-        } else {
+        auto paths = Utils::FindOlderFiles(search_dir, parameter, 8);
+        if (!paths.empty()) {
             for (auto &path : paths) {
                 DEBUG_PRINT("OLDER: %s", path.c_str());
                 SendFileResponse(datachannel, path, req.mode());
             }
+            return;
         }
+        SendFileResponse(datachannel, "", req.mode());
     } else if (type == protocol::QueryFileType::BEFORE_TIME) {
-        auto path = Utils::FindFilesFromDatetime(base_dir, parameter);
+        auto path = Utils::FindFilesFromDatetime(search_dir, parameter);
         DEBUG_PRINT("TIME_MATCH: %s", path.c_str());
         SendFileResponse(datachannel, path, req.mode());
     }
@@ -395,11 +394,13 @@ void Conductor::ControlCamera(std::shared_ptr<RtcChannel> datachannel,
     }
 }
 
-void Conductor::SetOnDemandRecorder(RecorderManager *recorder) { ondemand_recorder_ = recorder; }
+void Conductor::SetOnDemandRecorder(std::shared_ptr<RecorderManager> recorder) {
+    ondemand_recorder_ = recorder;
+}
 
 void Conductor::StartRecording(std::shared_ptr<RtcChannel> datachannel,
                                const protocol::Packet &pkt) {
-    auto recorder = ondemand_recorder_;
+    auto recorder = ondemand_recorder_.lock();
     if (!recorder) {
         ERROR_PRINT("On-demand recorder is not set.");
         return;
@@ -415,7 +416,7 @@ void Conductor::StartRecording(std::shared_ptr<RtcChannel> datachannel,
 
 void Conductor::StopRecording(std::shared_ptr<RtcChannel> datachannel,
                               const protocol::Packet &pkt) {
-    auto recorder = ondemand_recorder_;
+    auto recorder = ondemand_recorder_.lock();
     if (!recorder) {
         ERROR_PRINT("On-demand recorder is not set.");
         return;

src/rtc/conductor.h

@@ -30,7 +30,7 @@ class Conductor {
     std::shared_ptr<AudioCapturer> AudioSource() const;
     std::shared_ptr<VideoCapturer> VideoSource() const;
     void EnsureTracksAdded(webrtc::scoped_refptr<RtcPeer> peer);
-    void SetOnDemandRecorder(RecorderManager *recorder);
+    void SetOnDemandRecorder(std::shared_ptr<RecorderManager> recorder);
 
   private:
     Args args;
@@ -69,7 +69,7 @@ class Conductor {
     webrtc::scoped_refptr<ScaleTrackSource> video_track_source_;
 
     std::shared_ptr<UnixSocketServer> ipc_server_;
-    RecorderManager *ondemand_recorder_ = nullptr;
+    std::weak_ptr<RecorderManager> ondemand_recorder_;
 };
 
 #endif // CONDUCTOR_H_