diff --git a/src/rtc/rtc_peer.cpp b/src/rtc/rtc_peer.cpp index 80fcc61..7dd6997 100644 --- a/src/rtc/rtc_peer.cpp +++ b/src/rtc/rtc_peer.cpp @@ -30,13 +30,29 @@ void RtcPeer::CreateOffer() { peer_connection_->CreateOffer(this, webrtc::PeerConnectionInterface::RTCOfferAnswerOptions()); } +void RtcPeer::RenewSafetyFlag(webrtc::scoped_refptr &flag) { + if (flag) { + flag->SetNotAlive(); + } + flag = webrtc::PendingTaskSafetyFlag::CreateDetached(); +} + void RtcPeer::Terminate() { is_connected_.store(false); - is_complete_.store(true); + is_expired_.store(true); on_local_sdp_fn_ = nullptr; on_local_ice_fn_ = nullptr; - sdp_emit_safety_ = webrtc::ScopedTaskSafetyDetached(); + + if (peer_timeout_safety_) { + peer_timeout_safety_->SetNotAlive(); + } + if (sdp_emit_safety_) { + sdp_emit_safety_->SetNotAlive(); + } + if (reconnect_grace_safety_) { + reconnect_grace_safety_->SetNotAlive(); + } if (peer_connection_) { peer_connection_->Close(); peer_connection_ = nullptr; @@ -62,6 +78,8 @@ bool RtcPeer::isPublisher() const { return is_publisher_; } bool RtcPeer::isConnected() const { return is_connected_.load(); } +bool RtcPeer::isExpired() const { return is_expired_.load(); } + void RtcPeer::SetSink(webrtc::VideoSinkInterface *video_sink_obj) { custom_video_sink_ = std::move(video_sink_obj); } @@ -166,12 +184,12 @@ void RtcPeer::OnSignalingChange(webrtc::PeerConnectionInterface::SignalingState new_state == webrtc::PeerConnectionInterface::SignalingState::kHaveRemoteOffer); if (new_state == webrtc::PeerConnectionInterface::SignalingState::kHaveRemoteOffer) { // Cancel any previous timeout and schedule a new one on the signaling thread. - peer_timeout_safety_ = webrtc::ScopedTaskSafetyDetached(); + RenewSafetyFlag(peer_timeout_safety_); webrtc::Thread::Current()->PostDelayedTask( webrtc::SafeTask( - peer_timeout_safety_.flag(), + peer_timeout_safety_, [this]() { - if (peer_connection_ && !is_complete_.load() && !is_connected_.load()) { + if (peer_connection_ && !is_expired_.load() && !is_connected_.load()) { DEBUG_PRINT("Connection timeout after kConnecting. Closing connection."); peer_connection_->Close(); peer_connection_ = nullptr; @@ -219,6 +237,8 @@ void RtcPeer::OnConnectionChange(webrtc::PeerConnectionInterface::PeerConnection DEBUG_PRINT("OnConnectionChange => %s", std::string(state).c_str()); if (new_state == webrtc::PeerConnectionInterface::PeerConnectionState::kConnected) { is_connected_.store(true); + // Cancel the pending reconnect-grace. + RenewSafetyFlag(reconnect_grace_safety_); if (needs_renegotiation_ && signaling_state_ == webrtc::PeerConnectionInterface::SignalingState::kStable) { needs_renegotiation_ = false; @@ -227,11 +247,28 @@ void RtcPeer::OnConnectionChange(webrtc::PeerConnectionInterface::PeerConnection } } else if (new_state == webrtc::PeerConnectionInterface::PeerConnectionState::kFailed) { is_connected_.store(false); - peer_connection_->Close(); - peer_connection_ = nullptr; + RenewSafetyFlag(reconnect_grace_safety_); + auto *current_thread = webrtc::Thread::Current(); + DEBUG_PRINT("Arming reconnect-grace timer (thread=%p) for %d seconds (%s).", + (void *)current_thread, timeout_, id_.c_str()); + current_thread->PostDelayedTask( + webrtc::SafeTask( + reconnect_grace_safety_, + [this]() { + DEBUG_PRINT("Reconnect-grace timer fired (%s): has_pc=%d, connected=%d.", + id_.c_str(), peer_connection_ != nullptr, is_connected_.load()); + if (peer_connection_ && !is_connected_.load()) { + DEBUG_PRINT("No reconnect within %d seconds. Closing connection (%s).", + timeout_, id_.c_str()); + peer_connection_->Close(); + peer_connection_ = nullptr; + is_expired_.store(true); + } + }), + webrtc::TimeDelta::Seconds(timeout_)); } else if (new_state == webrtc::PeerConnectionInterface::PeerConnectionState::kClosed) { is_connected_.store(false); - is_complete_.store(true); + is_expired_.store(true); } } @@ -305,7 +342,7 @@ void RtcPeer::EmitLocalSdp(int delay_sec) { } // Cancel any previously scheduled SDP emit. - sdp_emit_safety_ = webrtc::ScopedTaskSafetyDetached(); + RenewSafetyFlag(sdp_emit_safety_); auto send_sdp = [this]() { std::string type = webrtc::SdpTypeToString(modified_desc_->GetType()); @@ -314,7 +351,7 @@ void RtcPeer::EmitLocalSdp(int delay_sec) { }; if (delay_sec > 0) { - webrtc::Thread::Current()->PostDelayedTask(webrtc::SafeTask(sdp_emit_safety_.flag(), + webrtc::Thread::Current()->PostDelayedTask(webrtc::SafeTask(sdp_emit_safety_, [this, send_sdp]() { send_sdp(); }), diff --git a/src/rtc/rtc_peer.h b/src/rtc/rtc_peer.h index 2a3247f..0f478fc 100644 --- a/src/rtc/rtc_peer.h +++ b/src/rtc/rtc_peer.h @@ -112,6 +112,7 @@ class RtcPeer : public webrtc::PeerConnectionObserver, bool isSfuPeer() const; bool isPublisher() const; bool isConnected() const; + bool isExpired() const; std::string id() const; void SetSink(webrtc::VideoSinkInterface *video_sink_obj); @@ -145,6 +146,7 @@ class RtcPeer : public webrtc::PeerConnectionObserver, std::string ModifySetupAttribute(const std::string &sdp, const std::string &new_setup); void EmitLocalSdp(int delay_sec = 0); void FlushPendingIce(); + void RenewSafetyFlag(webrtc::scoped_refptr &flag); struct PendingIceCandidate { std::string sdp_mid; @@ -161,10 +163,11 @@ class RtcPeer : public webrtc::PeerConnectionObserver, bool has_candidates_in_sdp_; bool needs_renegotiation_ = false; std::atomic is_connected_ = false; - std::atomic is_complete_ = false; + std::atomic is_expired_ = false; std::atomic is_negotiating_ = false; - webrtc::ScopedTaskSafetyDetached peer_timeout_safety_; - webrtc::ScopedTaskSafetyDetached sdp_emit_safety_; + webrtc::scoped_refptr peer_timeout_safety_; + webrtc::scoped_refptr sdp_emit_safety_; + webrtc::scoped_refptr reconnect_grace_safety_; std::string modified_sdp_; webrtc::PeerConnectionInterface::SignalingState signaling_state_; diff --git a/src/signaling/mqtt_service.cpp b/src/signaling/mqtt_service.cpp index 8323071..4e6cccb 100644 --- a/src/signaling/mqtt_service.cpp +++ b/src/signaling/mqtt_service.cpp @@ -261,10 +261,11 @@ void MqttService::RefreshPeerMap() { continue; } - DEBUG_PRINT("Found peer_id key: %s, connected value: %d", peer_id.c_str(), - peer->isConnected()); + const char *status = + peer->isExpired() ? "expired" : (peer->isConnected() ? "connected" : "reconnecting"); + DEBUG_PRINT("Found peer_id key: %s, status: %s", peer_id.c_str(), status); - if (!peer->isConnected()) { + if (peer->isExpired()) { auto it_c = peer_id_to_client_id_.find(peer_id); if (it_c != peer_id_to_client_id_.end()) { std::string client_id = it_c->second; diff --git a/src/signaling/signaling_service.h b/src/signaling/signaling_service.h index da8c8de..2f25428 100644 --- a/src/signaling/signaling_service.h +++ b/src/signaling/signaling_service.h @@ -52,7 +52,7 @@ class SignalingService { while (pm_it != peer_map_.end()) { auto peer_id = pm_it->second->id(); - if (pm_it->second && !pm_it->second->isConnected()) { + if (pm_it->second && pm_it->second->isExpired()) { pm_it = peer_map_.erase(pm_it); DEBUG_PRINT("peer_map (%s) was erased.", peer_id.c_str()); } else {