Events Being Dropped on Connection

.gitignore

@@ -34,3 +34,4 @@ lib/
 web/
 *trace.json
 compile_commands.json
+user_stories

AGENTS.md

@@ -8,7 +8,7 @@ This is **client-sdk-cpp**, the official LiveKit C++ client SDK. It wraps a Rust
 
 ### Core Principle
 Rust owns as much of the business logic as possible. The C++ layer should be a thin wrapper around the Rust core. If a feature may be used by another SDK it should be implemented in Rust. Since this is an SDK,
-ensure backwards compatibility is maintained when possible.
+ensure backwards compatibility is maintained when possible. Do not update auto generated code.
 
 ### Platform Support
 The SDK must be supported on the following platforms:

include/livekit/rpc_error.h

@@ -77,6 +77,7 @@ class LIVEKIT_API RpcError : public std::runtime_error {
    * @param data     Optional extra data, e.g. JSON. Empty string means no data.
    */
   RpcError(ErrorCode code, std::string message, std::string data = {});
+  ~RpcError() override;
 
   /**
    * Numeric error code.

src/ffi_client.cpp

@@ -35,7 +35,6 @@
 namespace livekit {
 
 namespace {
-
 inline void logAndThrow(const std::string& error_msg) {
   LK_LOG_ERROR("LiveKit SDK Error: {}", error_msg);
   throw std::runtime_error(error_msg);

src/room.cpp

@@ -18,14 +18,11 @@
 
 #include <functional>
 
-#include "data_track.pb.h"
 #include "ffi.pb.h"
 #include "ffi_client.h"
 #include "livekit/audio_stream.h"
 #include "livekit/e2ee.h"
-#include "livekit/local_data_track.h"
 #include "livekit/local_participant.h"
-#include "livekit/local_track_publication.h"
 #include "livekit/remote_audio_track.h"
 #include "livekit/remote_data_track.h"
 #include "livekit/remote_participant.h"
@@ -65,6 +62,16 @@ std::shared_ptr<livekit::RemoteParticipant> createRemoteParticipant(const proto:
                                                       pinfo.metadata(), std::move(attrs), kind, reason);
 }
 
+void readyForRoomEvent(std::uint64_t room_handle) {
+  FfiRequest req;
+  req.mutable_ready_for_room_event()->set_room_handle(room_handle);
+
+  const auto resp = FfiClient::instance().sendRequest(req);
+  if (!resp.has_ready_for_room_event()) {
+    throw std::runtime_error("FfiResponse missing ready_for_room_event");
+  }
+}
+
 } // namespace
 Room::Room() : subscription_thread_dispatcher_(std::make_unique<SubscriptionThreadDispatcher>()) {}
 
@@ -104,20 +111,28 @@ void Room::setDelegate(RoomDelegate* delegate) {
 
 bool Room::Connect(const std::string& url, const std::string& token, const RoomOptions& options) {
   TRACE_EVENT0("livekit", "Room::Connect");
-
   {
     const std::scoped_lock<std::mutex> g(lock_);
     if (connection_state_ != ConnectionState::Disconnected) {
       throw std::runtime_error("already connected");
     }
     connection_state_ = ConnectionState::Reconnecting;
   }
-  auto fut = FfiClient::instance().connectAsync(url, token, options);
+
+  FfiClient::ListenerId listenerId = 0;
   try {
+    listenerId = FfiClient::instance().AddListener([this](const proto::FfiEvent& e) { OnEvent(e); });
+    {
+      const std::scoped_lock<std::mutex> g(lock_);
+      listener_id_ = listenerId;
+    }
+
+    auto fut = FfiClient::instance().connectAsync(url, token, options);
     auto connectCb = fut.get(); // fut will throw if it fails to connect to the room
 
     const auto& owned_room = connectCb.result().room();
     auto new_room_handle = std::make_shared<FfiHandle>(owned_room.handle().id());
+    const auto room_handle_id = static_cast<std::uint64_t>(new_room_handle->get());
     auto new_room_info = fromProto(owned_room.info());
 
     // Setup local particpant
@@ -141,6 +156,7 @@ bool Room::Connect(const std::string& url, const std::string& token, const RoomO
           std::make_unique<LocalParticipant>(std::move(participant_handle), pinfo.sid(), pinfo.name(), pinfo.identity(),
                                              pinfo.metadata(), std::move(attrs), kind, reason);
     }
+
     // Setup remote participants
     std::unordered_map<std::string, std::shared_ptr<RemoteParticipant>> new_remote_participants;
     {
@@ -178,17 +194,31 @@ bool Room::Connect(const std::string& url, const std::string& token, const RoomO
       connection_state_ = ConnectionState::Connected;
     }
 
-    // Install listener (Room is fully initialized)
-    auto listenerId = FfiClient::instance().AddListener([this](const proto::FfiEvent& e) { OnEvent(e); });
+    readyForRoomEvent(room_handle_id);
+    return true;
+  } catch (const std::exception& e) {
+    int listener_to_remove = 0;
+    std::unique_ptr<LocalParticipant> local_participant_to_cleanup;
     {
       const std::scoped_lock<std::mutex> g(lock_);
-      listener_id_ = listenerId;
+      connection_state_ = ConnectionState::Disconnected;
+      if (listener_id_ == listenerId) {
+        listener_to_remove = listener_id_;
+        listener_id_ = 0;
+      }
+      local_participant_to_cleanup = std::move(local_participant_);
+      remote_participants_.clear();
+      room_handle_.reset();
+      e2ee_manager_.reset();
+      text_stream_readers_.clear();
+      byte_stream_readers_.clear();
+    }
+    if (local_participant_to_cleanup) {
+      local_participant_to_cleanup->shutdown();
+    }
+    if (listener_to_remove != 0) {
+      FfiClient::instance().RemoveListener(listener_to_remove);
     }
-
-    return true;
-  } catch (const std::exception& e) {
-    // On error, set the connection_state_ to Disconnected
-    connection_state_ = ConnectionState::Disconnected;
     LK_LOG_ERROR("Room::Connect failed: {}", e.what());
     return false;
   }

src/rpc_error.cpp

@@ -26,6 +26,8 @@ RpcError::RpcError(std::uint32_t code, std::string message, std::string data)
 RpcError::RpcError(ErrorCode code, std::string message, std::string data)
     : RpcError(static_cast<std::uint32_t>(code), std::move(message), std::move(data)) {}
 
+RpcError::~RpcError() = default;
+
 std::uint32_t RpcError::code() const noexcept { return code_; }
 
 const std::string& RpcError::message() const noexcept { return message_; }