refactor: improve audio handling

Author: std-microblock

src/client/client_context.h

@@ -20,6 +20,8 @@ struct ClientContext {
   std::shared_ptr<VirtualUSBHubService> virtual_usb_hub_service;
   std::shared_ptr<WebRTCService> webrtc_service;
   std::shared_ptr<audivis_widget> root_widget;
+
+  std::optional<std::thread> audio_thread;
 };
 
 } // namespace client

src/client/main.cc

@@ -649,13 +649,6 @@ ClientContext::ClientContext() {
 
 void ClientContext::init_webrtc_service() {
   webrtc_service = std::make_shared<WebRTCService>(
-      [&](const std::vector<uint8_t> &data) {
-        if (virtual_usb_hub_service && virtual_usb_hub_service->device_) {
-          return virtual_usb_hub_service->device_->submit_audio_data(data);
-        }
-
-        return false;
-      },
       [&](const WebRTCService::WebRTCStatus &status) {
         root_widget->owner_rt->post_loop_thread_task([this, status]() {
           if (status.state == WebRTCService::ConnectionState::Gathering) {
@@ -674,6 +667,34 @@ void ClientContext::init_webrtc_service() {
           }
         });
       });
+
+  if (!audio_thread) {
+    audio_thread.emplace([&]() {
+      while (true) {
+        std::this_thread::yield();
+        std::unique_lock<std::mutex> lock(webrtc_service->audioBufferMutex_);
+        while (webrtc_service->audioBuffer_.size() >= 960) {
+          std::vector<uint8_t> buffer(webrtc_service->audioBuffer_.begin(),
+                                      webrtc_service->audioBuffer_.begin() +
+                                          960);
+          if (virtual_usb_hub_service && virtual_usb_hub_service->device_) {
+            try {
+              if (virtual_usb_hub_service->device_->submit_audio_data(buffer)) {
+                webrtc_service->audioBuffer_.erase(
+                    webrtc_service->audioBuffer_.begin(),
+                    webrtc_service->audioBuffer_.begin() + 960);
+              }
+            } catch (const std::exception &e) {
+              std::cerr << "Error submitting audio data: " << e.what()
+                        << std::endl;
+              lock.unlock();
+              break;
+            }
+          }
+        }
+      }
+    });
+  }
 }
 
 } // namespace client

src/client/webrtc_service.cc

@@ -6,10 +6,8 @@ using json = nlohmann::json;
 
 namespace client {
 
-WebRTCService::WebRTCService(AudioDataCallback audio_data_callback,
-                             StatusCallback status_callback)
+WebRTCService::WebRTCService(StatusCallback status_callback)
     : cli_("http://audivis-signaling-server.microblock.cc"),
-      audio_data_callback_(std::move(audio_data_callback)),
       status_callback_(std::move(status_callback)) {
   cli_.set_connection_timeout(0, 300000); // 5 minutes
   setup_peer_connection();
@@ -109,18 +107,8 @@ void WebRTCService::setup_data_channel() {
       for (const auto &byte : *binary) {
         data.push_back(static_cast<uint8_t>(byte));
       }
+      std::lock_guard<std::mutex> lock(audioBufferMutex_);
       audioBuffer_.insert(audioBuffer_.end(), data.begin(), data.end());
-
-      std::println("Received audio data, buffer size: {}", audioBuffer_.size());
-      while (audioBuffer_.size() >= 960) {
-        std::vector<uint8_t> buffer(audioBuffer_.begin(),
-                                    audioBuffer_.begin() + 960);
-        if (audio_data_callback_ && audio_data_callback_(buffer)) {
-          audioBuffer_.erase(audioBuffer_.begin(), audioBuffer_.begin() + 960);
-        }
-      }
-      std::println("Processed audio data, remaining buffer size: {}",
-                   audioBuffer_.size());
     }
   });
 }

src/client/webrtc_service.h

@@ -27,10 +27,9 @@ struct WebRTCService {
         std::string session_id;
     };
 
-    using AudioDataCallback = std::function<bool(const std::vector<uint8_t>&)>;
     using StatusCallback = std::function<void(const WebRTCStatus&)>;
 
-    WebRTCService(AudioDataCallback audio_data_callback, StatusCallback status_callback);
+    WebRTCService(StatusCallback status_callback);
     ~WebRTCService();
 
     void start_signaling();
@@ -40,7 +39,7 @@ struct WebRTCService {
     std::shared_ptr<rtc::PeerConnection> pc_;
     std::shared_ptr<rtc::DataChannel> dc_;
     std::deque<uint8_t> audioBuffer_;
-    AudioDataCallback audio_data_callback_;
+    std::mutex audioBufferMutex_;
     StatusCallback status_callback_;
     std::string session_id_;