fix the timeout issue and a few bugs related to the SimpleRoom example

Author: xianshijing-lk

client-sdk-rust

@@ -319,7 +319,7 @@ int main(int argc, char *argv[]) {
             << "  Creation time (ms): " << info.creation_time << "\n";
 
   // Setup Audio Source / Track
-  auto audioSource = std::make_shared<AudioSource>(44100, 1, 10);
+  auto audioSource = std::make_shared<AudioSource>(44100, 1, 0);
   auto audioTrack =
       LocalAudioTrack::createLocalAudioTrack("micTrack", audioSource);
 
@@ -385,6 +385,8 @@ int main(int argc, char *argv[]) {
   // Shutdown the audio / video capture threads.
   media.stopMic();
   media.stopCamera();
+  media.stopSpeaker();
+  media.shutdownRenderer();
 
   // Drain any queued tasks that might still try to update the renderer /
   // speaker

examples/simple_room/main.cpp

@@ -31,6 +31,7 @@ SDLMediaManager::~SDLMediaManager() {
   stopMic();
   stopCamera();
   stopSpeaker();
+  shutdownRenderer();
 }
 
 bool SDLMediaManager::ensureSDLInit(Uint32 flags) {

examples/simple_room/sdl_media_manager.cpp

@@ -22,6 +22,8 @@
 
 using namespace livekit;
 
+constexpr int kMaxFPS = 60;
+
 SDLVideoRenderer::SDLVideoRenderer() = default;
 
 SDLVideoRenderer::~SDLVideoRenderer() { shutdown(); }
@@ -95,6 +97,16 @@ void SDLVideoRenderer::render() {
     return;
   }
 
+  // Throttle rendering to kMaxFPS
+  const auto now = std::chrono::steady_clock::now();
+  if (last_render_time_.time_since_epoch().count() != 0) {
+    const auto min_interval = std::chrono::microseconds(1'000'000 / kMaxFPS);
+    if (now - last_render_time_ < min_interval) {
+      return;
+    }
+  }
+  last_render_time_ = now;
+
   // 3) Read a frame from VideoStream (blocking until one is available)
   livekit::VideoFrameEvent vfe;
   bool gotFrame = stream_->read(vfe);

examples/simple_room/sdl_video_renderer.cpp

@@ -49,4 +49,5 @@ class SDLVideoRenderer {
   std::shared_ptr<livekit::VideoStream> stream_;
   int width_ = 0;
   int height_ = 0;
+  std::chrono::steady_clock::time_point last_render_time_{};
 };

examples/simple_room/sdl_video_renderer.h

@@ -41,8 +41,34 @@ class AudioSource {
    * @param sample_rate   Sample rate in Hz.
    * @param num_channels  Number of channels.
    * @param queue_size_ms Max buffer duration for the internal queue in ms.
+   *
+   * Buffering behavior:
+   * -------------------
+   * - queue_size_ms == 0 (recommended for real-time capture):
+   *     Disables internal buffering entirely. Audio frames are forwarded
+   *     directly to WebRTC sinks and consumed synchronously.
+   *
+   *     This mode is optimized for real-time audio capture driven by hardware
+   *     media callbacks (e.g. microphone capture). The caller is expected to
+   *     provide fixed-size real-time frames (typically 10 ms per call).
+   *
+   *     Because the native side consumes frames immediately, this mode
+   * minimizes latency and jitter and is the best choice for live capture
+   * scenarios.
+   *
+   * - queue_size_ms > 0 (buffered / blocking mode):
+   *     Enables an internal queue that buffers audio up to the specified
+   * duration. Frames are accumulated and flushed asynchronously once the buffer
+   * reaches its threshold.
+   *
+   *     This mode is intended for non-real-time producers (e.g. TTS engines,
+   *     file-based audio, or agents generating audio faster or slower than
+   *     real-time). The buffering layer smooths timing and allows the audio to
+   * be streamed out in real time even if the producer is bursty.
+   *
+   *     queue_size_ms must be a multiple of 10.
    */
-  AudioSource(int sample_rate, int num_channels, int queue_size_ms = 1000);
+  AudioSource(int sample_rate, int num_channels, int queue_size_ms = 0);
   virtual ~AudioSource() = default;
 
   AudioSource(const AudioSource &) = delete;
@@ -86,19 +112,32 @@ class AudioSource {
    * callback arrives (recommended for production unless the caller needs
    * explicit timeout control).
    *
-   * Notes:
-   *   - This is a blocking call.
-   *   - timeout_ms == 0 (infinite wait) is the safest mode because it
-   * guarantees the callback completes before the function returns, which in
-   * turn guarantees that the audio buffer lifetime is fully protected. The
-   * caller does not need to manage or extend the frame lifetime manually.
+   * Blocking semantics:
+   * The blocking behavior of this call depends on the buffering mode selected
+   * at construction time:
+   *
+   * - queue_size_ms == 0 (real-time capture mode):
+   *     Frames are consumed synchronously by the native layer. The FFI callback
+   *     is invoked immediately as part of the capture call, so this function
+   *     returns quickly.
+   *
+   *     This mode relies on the caller being paced by a real-time media
+   * callback (e.g. audio hardware interrupt / capture thread). It provides the
+   * lowest possible latency and is ideal for live microphone capture.
+   *
+   * - queue_size_ms > 0 (buffered / non-real-time mode):
+   *     Frames are queued internally and flushed asynchronously. This function
+   *     will block until the buffered audio corresponding to this frame has
+   * been consumed by the native side and the FFI callback fires.
    *
-   *   - May throw std::runtime_error if:
-   *       • the FFI reports an error
+   *     This mode is best suited for non-real-time audio producers (such as TTS
+   *     engines or agents) that generate audio independently of real-time
+   * pacing, while still streaming audio out in real time.
    *
-   *   - The underlying FFI request *must* eventually produce a callback for
-   * each frame. If the FFI layer is misbehaving or the event loop is stalled,
-   *     a timeout may occur in bounded-wait mode.
+   * Safety notes:
+   * May throw std::runtime_error if:
+   *   - the FFI reports an error
+   *   - a timeout occurs in bounded-wait mode
    */
   void captureFrame(const AudioFrame &frame, int timeout_ms = 20);
 

include/livekit/audio_source.h

@@ -44,6 +44,93 @@ inline void logAndThrow(const std::string &error_msg) {
   throw std::runtime_error(error_msg);
 }
 
+std::optional<FfiClient::AsyncId> ExtractAsyncId(const proto::FfiEvent &event) {
+  using E = proto::FfiEvent;
+  switch (event.message_case()) {
+  case E::kConnect:
+    return event.connect().async_id();
+  case E::kDisconnect:
+    return event.disconnect().async_id();
+  case E::kDispose:
+    return event.dispose().async_id();
+  case E::kPublishTrack:
+    return event.publish_track().async_id();
+  case E::kUnpublishTrack:
+    return event.unpublish_track().async_id();
+  case E::kPublishData:
+    return event.publish_data().async_id();
+  case E::kPublishTranscription:
+    return event.publish_transcription().async_id();
+  case E::kCaptureAudioFrame:
+    return event.capture_audio_frame().async_id();
+  case E::kSetLocalMetadata:
+    return event.set_local_metadata().async_id();
+  case E::kSetLocalName:
+    return event.set_local_name().async_id();
+  case E::kSetLocalAttributes:
+    return event.set_local_attributes().async_id();
+  case E::kGetStats:
+    return event.get_stats().async_id();
+  case E::kGetSessionStats:
+    return event.get_session_stats().async_id();
+  case E::kPublishSipDtmf:
+    return event.publish_sip_dtmf().async_id();
+  case E::kChatMessage:
+    return event.chat_message().async_id();
+  case E::kPerformRpc:
+    return event.perform_rpc().async_id();
+
+  // low-level data stream callbacks
+  case E::kSendStreamHeader:
+    return event.send_stream_header().async_id();
+  case E::kSendStreamChunk:
+    return event.send_stream_chunk().async_id();
+  case E::kSendStreamTrailer:
+    return event.send_stream_trailer().async_id();
+
+  // high-level
+  case E::kByteStreamReaderReadAll:
+    return event.byte_stream_reader_read_all().async_id();
+  case E::kByteStreamReaderWriteToFile:
+    return event.byte_stream_reader_write_to_file().async_id();
+  case E::kByteStreamOpen:
+    return event.byte_stream_open().async_id();
+  case E::kByteStreamWriterWrite:
+    return event.byte_stream_writer_write().async_id();
+  case E::kByteStreamWriterClose:
+    return event.byte_stream_writer_close().async_id();
+  case E::kSendFile:
+    return event.send_file().async_id();
+
+  case E::kTextStreamReaderReadAll:
+    return event.text_stream_reader_read_all().async_id();
+  case E::kTextStreamOpen:
+    return event.text_stream_open().async_id();
+  case E::kTextStreamWriterWrite:
+    return event.text_stream_writer_write().async_id();
+  case E::kTextStreamWriterClose:
+    return event.text_stream_writer_close().async_id();
+  case E::kSendText:
+    return event.send_text().async_id();
+  case E::kSendBytes:
+    return event.send_bytes().async_id();
+
+  // NOT async completion:
+  case E::kRoomEvent:
+  case E::kTrackEvent:
+  case E::kVideoStreamEvent:
+  case E::kAudioStreamEvent:
+  case E::kByteStreamReaderEvent:
+  case E::kTextStreamReaderEvent:
+  case E::kRpcMethodInvocation:
+  case E::kLogs:
+  case E::kPanic:
+  case E::MESSAGE_NOT_SET:
+  default:
+    return std::nullopt;
+  }
+}
+
 } // namespace
 
 FfiClient::~FfiClient() {
@@ -77,7 +164,7 @@ bool FfiClient::isInitialized() const noexcept {
 FfiClient::ListenerId
 FfiClient::AddListener(const FfiClient::Listener &listener) {
   std::lock_guard<std::mutex> guard(lock_);
-  FfiClient::ListenerId id = nextListenerId++;
+  FfiClient::ListenerId id = next_listener_id++;
   listeners_[id] = listener;
   return id;
 }
@@ -117,34 +204,33 @@ FfiClient::sendRequest(const proto::FfiRequest &request) const {
 }
 
 void FfiClient::PushEvent(const proto::FfiEvent &event) const {
-  std::vector<std::unique_ptr<PendingBase>> to_complete;
+  std::unique_ptr<PendingBase> to_complete;
+  std::vector<Listener> listeners_copy;
   {
     std::lock_guard<std::mutex> guard(lock_);
-    for (auto it = pending_.begin(); it != pending_.end();) {
-      if ((*it)->matches(event)) {
-        to_complete.push_back(std::move(*it));
-        it = pending_.erase(it);
-      } else {
-        ++it;
+
+    // Complete pending future if this event is a callback with async_id
+    if (auto async_id = ExtractAsyncId(event)) {
+      auto it = pending_by_id_.find(*async_id);
+      if (it != pending_by_id_.end() && it->second &&
+          it->second->matches(event)) {
+        to_complete = std::move(it->second);
+        pending_by_id_.erase(it);
       }
     }
-  }
-
-  // Run handlers outside lock
-  for (auto &p : to_complete) {
-    p->complete(event);
-  }
 
-  // Notify listeners. Note, we copy the listeners here to avoid calling into
-  // the listeners under the lock, which could potentially cause deadlock.
-  std::vector<Listener> listeners_copy;
-  {
-    std::lock_guard<std::mutex> guard(lock_);
+    // Snapshot listeners
     listeners_copy.reserve(listeners_.size());
-    for (auto &[_, listener] : listeners_) {
-      listeners_copy.push_back(listener);
+    for (const auto &kv : listeners_) {
+      listeners_copy.push_back(kv.second);
     }
   }
+  // Run handler outside lock
+  if (to_complete) {
+    to_complete->complete(event);
+  }
+
+  // Notify listeners outside lock
   for (auto &listener : listeners_copy) {
     listener(event);
   }
@@ -158,22 +244,19 @@ void LivekitFfiCallback(const uint8_t *buf, size_t len) {
 }
 
 FfiClient::AsyncId FfiClient::generateAsyncId() {
-  return nextAsyncId_.fetch_add(1, std::memory_order_relaxed);
+  return next_async_id_.fetch_add(1, std::memory_order_relaxed);
 }
 
 bool FfiClient::cancelPendingByAsyncId(AsyncId async_id) {
   std::unique_ptr<PendingBase> to_cancel;
   {
     std::lock_guard<std::mutex> guard(lock_);
-    for (auto it = pending_.begin(); it != pending_.end(); ++it) {
-      if ((*it)->async_id == async_id) {
-        to_cancel = std::move(*it);
-        pending_.erase(it);
-        break;
-      }
+    auto it = pending_by_id_.find(async_id);
+    if (it != pending_by_id_.end()) {
+      to_cancel = std::move(it->second);
+      pending_by_id_.erase(it);
     }
   }
-
   if (to_cancel) {
     to_cancel->cancel();
     return true;
@@ -192,7 +275,7 @@ std::future<T> FfiClient::registerAsync(
   pending->handler = std::move(handler);
   {
     std::lock_guard<std::mutex> guard(lock_);
-    pending_.push_back(std::move(pending));
+    pending_by_id_.emplace(async_id, std::move(pending));
   }
   return fut;
 }

src/ffi_client.cpp

@@ -169,8 +169,12 @@ class FfiClient {
     }
 
     void cancel() override {
-      promise.set_exception(std::make_exception_ptr(
-          std::runtime_error("Async operation cancelled")));
+      try {
+        promise.set_exception(std::make_exception_ptr(
+            std::runtime_error("Async operation cancelled")));
+      } catch (const std::future_error &) {
+        // already satisfied
+      }
     }
   };
 
@@ -187,10 +191,11 @@ class FfiClient {
   bool cancelPendingByAsyncId(AsyncId async_id);
 
   std::unordered_map<ListenerId, Listener> listeners_;
-  ListenerId nextListenerId = 1;
+  std::atomic<ListenerId> next_listener_id{1};
   mutable std::mutex lock_;
-  mutable std::vector<std::unique_ptr<PendingBase>> pending_;
-  std::atomic<AsyncId> nextAsyncId_{1};
+  mutable std::unordered_map<AsyncId, std::unique_ptr<PendingBase>>
+      pending_by_id_;
+  std::atomic<AsyncId> next_async_id_{1};
 
   void PushEvent(const proto::FfiEvent &event) const;
   friend void LivekitFfiCallback(const uint8_t *buf, size_t len);