From f93ad3beb1420626ecb9b0d2d937e334f2287a42 Mon Sep 17 00:00:00 2001 From: edwinwutw Date: Fri, 8 Jan 2021 01:00:39 +0800 Subject: [PATCH] Roll forward to cl/350492846 Signed-off-by: edwinwutw --- cpp/core/internal/base_pcp_handler.cc | 5 +- cpp/core/internal/mediums/webrtc.cc | 1295 ++++++++--------- cpp/core/internal/mediums/webrtc.h | 236 +-- .../mediums/webrtc/connection_flow.cc | 42 +- .../internal/mediums/webrtc/connection_flow.h | 34 +- .../mediums/webrtc/connection_flow_test.cc | 23 +- .../mediums/webrtc/data_channel_listener.h | 8 +- cpp/core/internal/mediums/webrtc_test.cc | 98 +- cpp/core/internal/p2p_cluster_pcp_handler.cc | 3 +- cpp/core/internal/payload_manager.cc | 5 + cpp/core/internal/webrtc_bwu_handler.cc | 3 +- 11 files changed, 856 insertions(+), 896 deletions(-) diff --git a/cpp/core/internal/base_pcp_handler.cc b/cpp/core/internal/base_pcp_handler.cc index 759c6d14..a0bd9f1f 100644 --- a/cpp/core/internal/base_pcp_handler.cc +++ b/cpp/core/internal/base_pcp_handler.cc @@ -225,7 +225,10 @@ void BasePcpHandler::OnEncryptionSuccessRunnable( if (!ukey2) { // Fail early, if there is no crypto context. - ProcessPreConnectionResultFailure(connection_info.client, endpoint_id); + ProcessPreConnectionInitiationFailure( + endpoint_id, connection_info.channel.get(), {Status::kEndpointIoError}, + connection_info.result.get()); + connection_info.result.reset(); return; } diff --git a/cpp/core/internal/mediums/webrtc.cc b/cpp/core/internal/mediums/webrtc.cc index dddef4cf..7fc70115 100644 --- a/cpp/core/internal/mediums/webrtc.cc +++ b/cpp/core/internal/mediums/webrtc.cc @@ -2,10 +2,10 @@ #include #include -#include #include "core/internal/mediums/webrtc/session_description_wrapper.h" #include "core/internal/mediums/webrtc/signaling_frames.h" +#include "core/internal/mediums/webrtc/webrtc_socket_wrapper.h" #include "platform/base/byte_array.h" #include "platform/base/listeners.h" #include "platform/public/cancelable_alarm.h" @@ -13,7 +13,6 @@ #include "platform/public/logging.h" #include "platform/public/mutex_lock.h" #include "location/nearby/mediums/proto/web_rtc_signaling_frames.pb.h" -#include "absl/container/flat_hash_map.h" #include "absl/strings/str_cat.h" #include "absl/time/time.h" #include "webrtc/api/jsep.h" @@ -26,7 +25,7 @@ namespace mediums { namespace { // The maximum amount of time to wait to connect to a data channel via WebRTC. -constexpr absl::Duration kDataChannelTimeout = absl::Milliseconds(5000); +constexpr absl::Duration kDataChannelTimeout = absl::Seconds(10); // Delay between restarting signaling messenger to receive messages. constexpr absl::Duration kRestartReceiveMessagesDuration = absl::Seconds(60); @@ -36,32 +35,27 @@ constexpr absl::Duration kRestartReceiveMessagesDuration = absl::Seconds(60); WebRtc::WebRtc() = default; WebRtc::~WebRtc() { - { - MutexLock lock(&mutex_); - NEARBY_LOGS(WARNING) << "Destructing Webrtc: " << InternalStatesToString(); - } // This ensures that all pending callbacks are run before we reset the medium // and we are not accepting new runnables. - restart_receive_messages_executor_.Shutdown(); single_thread_executor_.Shutdown(); - // Disconnect will also erase the connection info from map. Use a separate - // set to save the connection ids to avoid the iterator violation issue. - absl::flat_hash_set connection_ids; - for (auto& item : accepting_map_) { - connection_ids.emplace(item.first); + // Stop accepting all connections + absl::flat_hash_set service_ids; + for (auto& item : accepting_connections_info_) { + service_ids.emplace(item.first); } - for (const auto& connection_id : connection_ids) { - Disconnect(Role::kOfferer, connection_id); + for (const auto& service_id : service_ids) { + StopAcceptingConnections(service_id); } - connection_ids.clear(); - for (auto& item : connecting_map_) { - connection_ids.emplace(item.first); + + // Disconnect all connections + absl::flat_hash_set peer_ids; + for (auto& item : sockets_) { + peer_ids.emplace(item.first); } - for (const auto& connection_id : connection_ids) { - Disconnect(Role::kAnswerer, connection_id); + for (const auto& peer_id : peer_ids) { + sockets_.find(peer_id)->second.Close(); } - connection_ids.clear(); } const std::string WebRtc::GetDefaultCountryCode() { @@ -72,684 +66,681 @@ bool WebRtc::IsAvailable() { return medium_.IsValid(); } bool WebRtc::IsAcceptingConnections(const std::string& service_id) { MutexLock lock(&mutex_); - ConnectionInfo* connection_info = - GetConnectionInfo(Role::kOfferer, service_id); - return connection_info && connection_info->self_id.IsValid(); + return IsAcceptingConnectionsLocked(service_id); +} + +bool WebRtc::IsAcceptingConnectionsLocked(const std::string& service_id) { + return accepting_connections_info_.contains(service_id); } bool WebRtc::StartAcceptingConnections(const std::string& service_id, - const PeerId& self_id, + const PeerId& self_peer_id, const LocationHint& location_hint, AcceptedConnectionCallback callback) { + MutexLock lock(&mutex_); if (!IsAvailable()) { - MutexLock lock(&mutex_); - LogAndDisconnect(Role::kOfferer, service_id, - "WebRTC is not available for data transfer."); + NEARBY_LOG(WARNING, + "Cannot start accepting WebRTC connections because WebRTC is " + "not available."); return false; } - if (IsAcceptingConnections(service_id)) { - NEARBY_LOG(WARNING, "Already accepting WebRTC connections."); - return false; - } - { - MutexLock lock(&mutex_); - NEARBY_LOGS(WARNING) << "StartAcceptingConnections: " - << InternalStatesToString(); - accepting_map_.emplace(service_id, - ConnectionInfo{.socket = WebRtcSocketWrapper()}); - ConnectionInfo* connection_info = &accepting_map_[service_id]; - if (!InitWebRtcFlow(Role::kOfferer, self_id, location_hint, service_id)) - return false; - - connection_info->restart_receive_messages_alarm = CancelableAlarm( - "restart_receiving_messages_webrtc", - std::bind(&WebRtc::RestartReceiveMessages, this, location_hint, - service_id), - kRestartReceiveMessagesDuration, &restart_receive_messages_executor_); - - SessionDescriptionWrapper offer = - connection_info->connection_flow->CreateOffer(); - connection_info->pending_local_offer = - webrtc_frames::EncodeOffer(self_id, offer.GetSdp()); - if (!SetLocalSessionDescription(std::move(offer), Role::kOfferer, - service_id)) { - NEARBY_LOG(WARNING, "Failed to set local session description."); - return false; - } - - // There is no timeout set for the future returned since we do not know how - // much time it will take for the two devices to discover each other before - // the actual transport can begin. - ListenForWebRtcSocketFuture( - Role::kOfferer, service_id, - connection_info->connection_flow->GetDataChannel(), - std::move(callback)); - NEARBY_LOG(WARNING, "Started listening for WebRtc connections as %s", - self_id.GetId().c_str()); - } - - return true; -} - -WebRtcSocketWrapper WebRtc::Connect(const PeerId& peer_id, - const LocationHint& location_hint) { - if (!IsAvailable()) { - MutexLock lock(&mutex_); - LogAndDisconnect(Role::kAnswerer, peer_id.GetId(), - "WebRTC is not available for data transfer."); - return WebRtcSocketWrapper(); - } - - { - MutexLock lock(&mutex_); - NEARBY_LOGS(WARNING) << "Start Connecting to " << peer_id.GetId() << ":\n" - << InternalStatesToString(); - if (connecting_map_.contains(peer_id.GetId())) { - NEARBY_LOG( - ERROR, - "Cannot connect with WebRtc because we are already connecting."); - return WebRtcSocketWrapper(); - } - connecting_map_.emplace(peer_id.GetId(), - ConnectionInfo{.socket = WebRtcSocketWrapper()}); - ConnectionInfo* connection_info = &connecting_map_[peer_id.GetId()]; - connection_info->peer_id = peer_id; - if (!InitWebRtcFlow(Role::kAnswerer, PeerId::FromRandom(), location_hint, - peer_id.GetId())) { - return WebRtcSocketWrapper(); - } - } - - NEARBY_LOG(WARNING, "Attempting to make a WebRTC connection to %s.", - peer_id.GetId().c_str()); - Future socket_future; - { - MutexLock lock(&mutex_); - socket_future = ListenForWebRtcSocketFuture( - Role::kAnswerer, peer_id.GetId(), - connecting_map_[peer_id.GetId()].connection_flow->GetDataChannel(), - AcceptedConnectionCallback()); - } - - // The two devices have discovered each other, hence we have a timeout for - // establishing the transport channel. - // NOTE - We should not hold |mutex_| while waiting for the data channel since - // it would block incoming signaling messages from being processed, resulting - // in a timeout in creating the socket. - ExceptionOr result = - socket_future.Get(kDataChannelTimeout); - if (result.ok()) { - NEARBY_LOGS(WARNING) << "Succeeded to make WebRTC connection to " - << peer_id.GetId(); - return result.result(); - } - - NEARBY_LOGS(WARNING) << "Failed to make WebRTC connection to " - << peer_id.GetId(); - Disconnect(Role::kAnswerer, peer_id.GetId()); - return WebRtcSocketWrapper(); -} - -bool WebRtc::SetLocalSessionDescription(SessionDescriptionWrapper sdp, - Role role, - const std::string& connection_id) { - ConnectionInfo* connection_info = GetConnectionInfo(role, connection_id); - if (!connection_info) return false; - if (!connection_info->connection_flow->SetLocalSessionDescription( - std::move(sdp))) { - LogAndDisconnect(role, connection_id, - "Unable to set local session description"); + if (IsAcceptingConnectionsLocked(service_id)) { + NEARBY_LOG(WARNING, + "Cannot start accepting WebRTC connections because service %s " + "is already accepting WebRTC connections.", + service_id.c_str()); return false; } + // We'll track our state here, so that we're separated from the other services + // who may be also using WebRTC. + AcceptingConnectionsInfo info = AcceptingConnectionsInfo(); + info.self_peer_id = self_peer_id; + info.accepted_connection_callback = callback; + + // Create a new SignalingMessenger so that we can communicate w/ Tachyon. + info.signaling_messenger = + medium_.GetSignalingMessenger(self_peer_id.GetId(), location_hint); + if (!info.signaling_messenger->IsValid()) { + return false; + } + + // This registers ourselves w/ Tachyon, creating a room from the PeerId. + // This allows a remote device to message us over Tachyon. + auto signaling_message_callback = [this, service_id](ByteArray message) { + OffloadFromThread([this, service_id{std::move(service_id)}, + message{std::move(message)}]() { + ProcessTachyonInboxMessage(service_id, message); + }); + }; + if (!info.signaling_messenger->StartReceivingMessages( + signaling_message_callback)) { + info.signaling_messenger.reset(); + return false; + } + + // We'll automatically disconnect from Tachyon after 60sec. When this alarm + // fires, we'll recreate our room so we continue to receive messages. + info.restart_tachyon_receive_messages_alarm = CancelableAlarm( + "restart_receiving_messages_webrtc", + std::bind(&WebRtc::ProcessRestartTachyonReceiveMessages, this, + service_id), + kRestartReceiveMessagesDuration, &single_thread_executor_); + + // Now that we're set up to receive messages, we'll save our state and return + // a successful result. + accepting_connections_info_.emplace(service_id, std::move(info)); + NEARBY_LOG(INFO, + "Started listening for WebRTC connections as %s on service %s", + self_peer_id.GetId().c_str(), service_id.c_str()); return true; } void WebRtc::StopAcceptingConnections(const std::string& service_id) { - if (!IsAcceptingConnections(service_id)) { + MutexLock lock(&mutex_); + if (!IsAcceptingConnectionsLocked(service_id)) { NEARBY_LOG(WARNING, - "Skipped StopAcceptingConnections since we are not currently " - "accepting WebRTC connections for %s", + "Cannot stop accepting WebRTC connections because service %s " + "is not accepting WebRTC connections.", service_id.c_str()); return; } - { - MutexLock lock(&mutex_); - LogAndShutdownSignaling(Role::kOfferer, service_id, - "Invoked by StopAcceptingConnections."); + // Grab our info from the map. + auto& info = accepting_connections_info_.find(service_id)->second; + + // Stop receiving messages from Tachyon. + info.signaling_messenger->StopReceivingMessages(); + info.signaling_messenger.reset(); + + // Cancel the scheduled alarm. + if (info.restart_tachyon_receive_messages_alarm.IsValid()) { + info.restart_tachyon_receive_messages_alarm.Cancel(); + info.restart_tachyon_receive_messages_alarm = CancelableAlarm(); } - NEARBY_LOG(WARNING, "Stopped accepting WebRTC connections for %s", + + // If we had any in-progress connections that haven't materialized into full + // DataChannels yet, it's time to shut them down since they can't reach us + // anymore. + absl::flat_hash_set peer_ids; + for (auto& item : connection_flows_) { + peer_ids.emplace(item.first); + } + for (const auto& peer_id : peer_ids) { + const auto& entry = connection_flows_.find(peer_id); + // Skip outgoing connections in this step. Start/StopAcceptingConnections + // only deals with incoming connections. + if (requesting_connections_info_.contains(peer_id)) { + continue; + } + + // Skip fully connected connections in this step. If the connection was + // formed while we were accepting connections, then it will stay alive until + // it's explicitly closed. + if (entry->second->GetState() == ConnectionFlow::State::kConnected) { + continue; + } + + entry->second->Close(); + connection_flows_.erase(peer_id); + } + + // Clean up our state. We're now no longer listening for connections. + accepting_connections_info_.erase(service_id); + NEARBY_LOG(INFO, "Stopped listening for WebRTC connections for service %s", service_id.c_str()); } -Future WebRtc::ListenForWebRtcSocketFuture( - const Role& role, const std::string& connection_id, - Future> - data_channel_future, - AcceptedConnectionCallback callback) { - Future socket_future; - auto data_channel_runnable = [this, role, connection_id, socket_future, - data_channel_future, - callback{std::move(callback)}]() mutable { - // The overall timeout of creating the socket and data channel is controlled - // by the caller of this function. - ExceptionOr> res = - data_channel_future.Get(); - if (res.ok()) { - WebRtcSocketWrapper wrapper = - CreateWebRtcSocketWrapper(role, connection_id, res.result()); - callback.accepted_cb(wrapper); - { - MutexLock lock(&mutex_); - ConnectionInfo* connection_info = - GetConnectionInfo(role, connection_id); - if (connection_info) { - connection_info->socket = wrapper; - } - } - socket_future.Set(wrapper); - } else { - NEARBY_LOG(WARNING, "Failed to get WebRtcSocket."); - socket_future.Set(WebRtcSocketWrapper()); +WebRtcSocketWrapper WebRtc::Connect(const std::string& service_id, + const PeerId& remote_peer_id, + const LocationHint& location_hint) { + for (int attempts_count = 0; attempts_count < kConnectAttemptsLimit; + attempts_count++) { + auto wrapper_result = + AttemptToConnect(service_id, remote_peer_id, location_hint); + if (wrapper_result.IsValid()) { + return wrapper_result; } - }; - - data_channel_future.AddListener(std::move(data_channel_runnable), - &single_thread_executor_); - - return socket_future; + } + return WebRtcSocketWrapper(); } -WebRtcSocketWrapper WebRtc::CreateWebRtcSocketWrapper( - const Role& role, const std::string& connection_id, - rtc::scoped_refptr data_channel) { - if (data_channel == nullptr) { - return WebRtcSocketWrapper(); - } +WebRtcSocketWrapper WebRtc::AttemptToConnect( + const std::string& service_id, const PeerId& remote_peer_id, + const LocationHint& location_hint) { + ConnectionRequestInfo info = ConnectionRequestInfo(); + info.self_peer_id = PeerId::FromRandom(); + Future socket_future = info.socket_future; - auto socket = std::make_unique("WebRtcSocket", data_channel); - socket->SetOnSocketClosedListener({[this, role, connection_id]() { - OffloadFromSignalingThread( - [this, role, connection_id]() { Disconnect(role, connection_id); }); - }}); - return WebRtcSocketWrapper(std::move(socket)); -} - -bool WebRtc::InitWebRtcFlow(const Role& role, const PeerId& self_id, - const LocationHint& location_hint, - const std::string& connection_id) { - ConnectionInfo* connection_info = GetConnectionInfo(role, connection_id); - if (!connection_info) { - NEARBY_LOGS(WARNING) << "Can not find matching connection info."; - return false; - } - connection_info->self_id = self_id; - - if (connection_info->connection_flow) { - LogAndShutdownSignaling( - role, connection_id, - "Tried to initialize WebRTC without shutting down the previous " - "connection"); - return false; - } - - if (connection_info->signaling_messenger) { - LogAndShutdownSignaling( - role, connection_id, - "Tried to initialize WebRTC without shutting down signaling messenger"); - return false; - } - connection_info->signaling_messenger = - medium_.GetSignalingMessenger(self_id.GetId(), location_hint); - auto signaling_message_callback = std::bind( - [this](ByteArray message, Role role, const std::string& connection_id) { - OffloadFromSignalingThread([this, message{std::move(message)}, - role{role}, - connection_id{connection_id}]() { - ProcessSignalingMessage(role, connection_id, message); - }); - }, - std::placeholders::_1, role, connection_id); - - if (!connection_info->signaling_messenger->IsValid() || - !connection_info->signaling_messenger->StartReceivingMessages( - signaling_message_callback)) { - LogAndDisconnect(role, connection_id, - "Could not receive from signaling messenger."); - return false; - } - - if (role == Role::kAnswerer && - !connection_info->signaling_messenger->SendMessage( - connection_info->peer_id.GetId(), - webrtc_frames::EncodeReadyForSignalingPoke(self_id))) { - LogAndDisconnect(Role::kAnswerer, connection_info->peer_id.GetId(), - absl::StrCat("Could not send signaling poke to peer ", - connection_info->peer_id.GetId())); - return false; - } - - connection_info->connection_flow = ConnectionFlow::Create( - GetLocalIceCandidateListener(role, connection_id), - GetDataChannelListener(role, connection_id), medium_); - if (!connection_info->connection_flow) { - LogAndDisconnect(role, connection_id, "Failed to create connection flow"); - return false; - } - - NEARBY_LOGS(WARNING) << "Succeeded to create connection flow for role:" - << role_names_[role] - << ", connection_id: " << connection_id; - return true; -} - -void WebRtc::OnLocalIceCandidate( - const Role& role, const std::string& connection_id, - const webrtc::IceCandidateInterface* local_ice_candidate) { - ::location::nearby::mediums::IceCandidate ice_candidate = - webrtc_frames::EncodeIceCandidate(*local_ice_candidate); - - OffloadFromSignalingThread([this, ice_candidate{std::move(ice_candidate)}, - role{role}, connection_id{connection_id}]() { - MutexLock lock(&mutex_); - ConnectionInfo* connection_info = GetConnectionInfo(role, connection_id); - if (IsSignaling(role, connection_id)) { - if (connection_info && connection_info->signaling_messenger) { - NEARBY_LOG(WARNING, "Sending local ice candidates to %s", - connection_info->peer_id.GetId().c_str()); - connection_info->signaling_messenger->SendMessage( - connection_info->peer_id.GetId(), - webrtc_frames::EncodeIceCandidates(connection_info->self_id, - {std::move(ice_candidate)})); - } else { - connection_info->pending_local_ice_candidates.push_back( - std::move(ice_candidate)); - } - } else { - connection_info->pending_local_ice_candidates.push_back( - std::move(ice_candidate)); - } - }); -} - -LocalIceCandidateListener WebRtc::GetLocalIceCandidateListener( - const Role& role, const std::string& connection_id) { - return {std::bind(&WebRtc::OnLocalIceCandidate, this, role, connection_id, - std::placeholders::_1)}; -} - -void WebRtc::OnDataChannelClosed(const Role& role, - const std::string& connection_id) { - OffloadFromSignalingThread([this, role, connection_id]() { - MutexLock lock(&mutex_); - LogAndDisconnect(role, connection_id, "WebRTC data channel closed"); - }); -} - -void WebRtc::OnDataChannelMessageReceived(const Role& role, - const std::string& connection_id, - const ByteArray& message) { - OffloadFromSignalingThread([this, role, connection_id, message]() { - { - MutexLock lock(&mutex_); - ConnectionInfo* connection_info = GetConnectionInfo(role, connection_id); - if (!connection_info) return; - if (!connection_info->socket.IsValid()) { - LogAndDisconnect(role, connection_id, - "Received a data channel message without a socket"); - return; - } - connection_info->socket.NotifyDataChannelMsgReceived(message); - } - }); -} - -void WebRtc::OnDataChannelBufferedAmountChanged( - const Role& role, const std::string& connection_id) { - OffloadFromSignalingThread([this, role, connection_id]() { - { - MutexLock lock(&mutex_); - ConnectionInfo* connection_info = GetConnectionInfo(role, connection_id); - if (!connection_info) return; - if (!connection_info->socket.IsValid()) { - LogAndDisconnect(role, connection_id, - "Data channel buffer changed without a socket"); - return; - } - connection_info->socket.NotifyDataChannelBufferedAmountChanged(); - } - }); -} - -DataChannelListener WebRtc::GetDataChannelListener( - const Role& role, const std::string& connection_id) { - return { - .data_channel_closed_cb = - std::bind(&WebRtc::OnDataChannelClosed, this, role, connection_id), - .data_channel_message_received_cb = - std::bind(&WebRtc::OnDataChannelMessageReceived, this, role, - connection_id, std::placeholders::_1), - .data_channel_buffered_amount_changed_cb = - std::bind(&WebRtc::OnDataChannelBufferedAmountChanged, this, role, - connection_id), - }; -} - -bool WebRtc::IsSignaling(const Role& role, const std::string& connection_id) { - ConnectionInfo* connection_info = GetConnectionInfo(role, connection_id); - if (!connection_info) return false; - return (connection_info->self_id.IsValid() && - connection_info->peer_id.IsValid()); -} - -void WebRtc::ProcessSignalingMessage(const Role& role, - const std::string& connection_id, - const ByteArray& message) { - MutexLock lock(&mutex_); - ConnectionInfo* connection_info = GetConnectionInfo(role, connection_id); - if (!connection_info) { - NEARBY_LOG(ERROR, - "Could not find connection info for role: %s, connection_id: %s", - role_names_[role].c_str(), connection_id.c_str()); - return; - } - - if (!connection_info->connection_flow) { - LogAndDisconnect(role, connection_id, - "Received WebRTC frame before signaling was started"); - return; - } - - location::nearby::mediums::WebRtcSignalingFrame frame; - if (!frame.ParseFromString(std::string(message))) { - LogAndDisconnect(role, connection_id, "Failed to parse signaling message"); - return; - } - - if (!frame.has_sender_id()) { - LogAndDisconnect(role, connection_id, - "Invalid WebRTC frame: Sender ID is missing"); - return; - } - - if (frame.has_ready_for_signaling_poke() && - !connection_info->peer_id.IsValid()) { - connection_info->peer_id = PeerId(frame.sender_id().id()); - NEARBY_LOG(WARNING, "Peer %s is ready for signaling", - connection_info->peer_id.GetId().c_str()); - } - - if (!IsSignaling(role, connection_id)) { - NEARBY_LOG(WARNING, - "Ignoring WebRTC frame: we are not currently listening for " - "signaling messages"); - return; - } - - if (frame.sender_id().id() != connection_info->peer_id.GetId()) { - NEARBY_LOG( - WARNING, - "Ignoring WebRTC frame: we are only listening for another peer."); - return; - } - - if (frame.has_ready_for_signaling_poke()) { - NEARBY_LOG(WARNING, - "Received ready-for-poke for role: %s connection_id: %s", - role_names_[role].c_str(), connection_id.c_str()); - SendOfferAndIceCandidatesToPeer(connection_id); - } else if (frame.has_offer()) { - NEARBY_LOG(WARNING, "Received offer for role: %s connection_id: %s", - role_names_[role].c_str(), connection_id.c_str()); - DCHECK(role == Role::kAnswerer); - connection_info->connection_flow->OnOfferReceived( - SessionDescriptionWrapper(webrtc_frames::DecodeOffer(frame).release())); - SendAnswerToPeer(connection_id); - } else if (frame.has_answer()) { - NEARBY_LOG(WARNING, "Received answer for role: %s connection_id: %s", - role_names_[role].c_str(), connection_id.c_str()); - DCHECK(role == Role::kOfferer); - connection_info->connection_flow->OnAnswerReceived( - SessionDescriptionWrapper( - webrtc_frames::DecodeAnswer(frame).release())); - } else if (frame.has_ice_candidates()) { - NEARBY_LOG(WARNING, - "Received ice candidates for role: %s connection_id: %s", - role_names_[role].c_str(), connection_id.c_str()); - if (!connection_info->connection_flow->OnRemoteIceCandidatesReceived( - webrtc_frames::DecodeIceCandidates(frame))) { - LogAndDisconnect(role, connection_id, - "Could not add remote ice candidates."); - } - } else { - NEARBY_LOG(WARNING, - "Received unknown frame type for role: %s connection_id: %s", - role_names_[role].c_str(), connection_id.c_str()); - } -} - -void WebRtc::SendOfferAndIceCandidatesToPeer(const std::string& service_id) { - ConnectionInfo* connection_info = - GetConnectionInfo(Role::kOfferer, service_id); - if (!connection_info) return; - if (connection_info->pending_local_offer.Empty()) { - LogAndDisconnect( - Role::kOfferer, service_id, - "Unable to send pending offer to remote peer: local offer not set"); - return; - } - - NEARBY_LOG(WARNING, "Sending offer from %s", service_id.c_str()); - if (!connection_info->signaling_messenger->SendMessage( - connection_info->peer_id.GetId(), - connection_info->pending_local_offer)) { - LogAndDisconnect(Role::kOfferer, service_id, - "Failed to send local offer via signaling messenger"); - return; - } - connection_info->pending_local_offer = ByteArray(); - - if (!connection_info->pending_local_ice_candidates.empty()) { - NEARBY_LOG(WARNING, "Sending local pending ice candidates from %s", - service_id.c_str()); - connection_info->signaling_messenger->SendMessage( - connection_info->peer_id.GetId(), - webrtc_frames::EncodeIceCandidates( - connection_info->self_id, - std::move(connection_info->pending_local_ice_candidates))); - } -} - -void WebRtc::SendAnswerToPeer(const std::string& peer_id) { - ConnectionInfo* connection_info = GetConnectionInfo(Role::kAnswerer, peer_id); - if (!connection_info) return; - SessionDescriptionWrapper answer = - connection_info->connection_flow->CreateAnswer(); - ByteArray answer_message( - webrtc_frames::EncodeAnswer(connection_info->self_id, answer.GetSdp())); - - if (!SetLocalSessionDescription(std::move(answer), Role::kAnswerer, peer_id)) - return; - - NEARBY_LOGS(WARNING) << "Sending answer to peer " << peer_id; - if (!connection_info->signaling_messenger->SendMessage( - connection_info->peer_id.GetId(), answer_message)) { - LogAndDisconnect(Role::kAnswerer, peer_id, - "Failed to send local answer via signaling messenger"); - return; - } -} - -void WebRtc::LogAndDisconnect(const Role& role, - const std::string& connection_id, - const std::string& error_message) { - NEARBY_LOG(WARNING, - "Disconnecting WebRTC role: %d, connection id: %s, msg: %s", role, - connection_id.c_str(), error_message.c_str()); - DisconnectLocked(role, connection_id); -} - -void WebRtc::LogAndShutdownSignaling(const Role& role, - const std::string& connection_id, - const std::string& error_message) { - NEARBY_LOG(WARNING, - "Stopping WebRTC role: %s, connection id: %s, msg: %s:\n%s", - role_names_[role].c_str(), connection_id.c_str(), - error_message.c_str(), InternalStatesToString().c_str()); - ShutdownSignaling(role, connection_id); -} - -void WebRtc::ShutdownSignaling(const Role& role, - const std::string& connection_id) { - ConnectionInfo* connection_info = GetConnectionInfo(role, connection_id); - if (!connection_info) { - return; - } - - connection_info->self_id = PeerId(); - connection_info->peer_id = PeerId(); - connection_info->pending_local_offer = ByteArray(); - connection_info->pending_local_ice_candidates.clear(); - - if (connection_info->restart_receive_messages_alarm.IsValid()) { - connection_info->restart_receive_messages_alarm.Cancel(); - connection_info->restart_receive_messages_alarm = CancelableAlarm(); - } - - if (connection_info->signaling_messenger) { - connection_info->signaling_messenger->StopReceivingMessages(); - connection_info->signaling_messenger.reset(); - } - - if (!connection_info->socket.IsValid()) - ShutdownIceCandidateCollection(role, connection_id); -} - -void WebRtc::Disconnect(const Role& role, const std::string& connection_id) { - MutexLock lock(&mutex_); - DisconnectLocked(role, connection_id); -} - -void WebRtc::DisconnectLocked(const Role& role, - const std::string& connection_id) { - NEARBY_LOGS(WARNING) << "Disconnecting role: " << role_names_[role] - << " connection_id: " << connection_id << ":\n" - << InternalStatesToString(); - ShutdownSignaling(role, connection_id); - ShutdownWebRtcSocket(role, connection_id); - ShutdownIceCandidateCollection(role, connection_id); - - if (role == Role::kOfferer && accepting_map_.contains(connection_id)) { - accepting_map_.erase(connection_id); - } else if (role == Role::kAnswerer && - connecting_map_.contains(connection_id)) { - connecting_map_.erase(connection_id); - } -} - -void WebRtc::ShutdownWebRtcSocket(const Role& role, - const std::string& connection_id) { - ConnectionInfo* connection_info = GetConnectionInfo(role, connection_id); - if (connection_info && connection_info->socket.IsValid()) { - connection_info->socket.Close(); - connection_info->socket = WebRtcSocketWrapper(); - } -} - -void WebRtc::ShutdownIceCandidateCollection(const Role& role, - const std::string& connection_id) { - ConnectionInfo* connection_info = GetConnectionInfo(role, connection_id); - if (connection_info && connection_info->connection_flow) { - connection_info->connection_flow->Close(); - connection_info->connection_flow.reset(); - } -} - -void WebRtc::OffloadFromSignalingThread(Runnable runnable) { - single_thread_executor_.Execute(std::move(runnable)); -} - -void WebRtc::RestartReceiveMessages(const LocationHint& location_hint, - const std::string& service_id) { - if (!IsAcceptingConnections(service_id)) { - NEARBY_LOG(WARNING, - "Skipping restart since we are not accepting connections."); - return; - } - - NEARBY_LOG(WARNING, "Restarting listening for receiving signaling messages."); { MutexLock lock(&mutex_); - ConnectionInfo* connection_info = - GetConnectionInfo(Role::kOfferer, service_id); - if (!connection_info) { - NEARBY_LOG(ERROR, - "Can't find connection info in RestartReceiveMessages for %s", - service_id.c_str()); - return; + if (!IsAvailable()) { + NEARBY_LOG( + WARNING, + "Cannot connect to WebRTC peer %s because WebRTC is not available.", + remote_peer_id.GetId().c_str()); + return WebRtcSocketWrapper(); } - connection_info->signaling_messenger->StopReceivingMessages(); - connection_info->signaling_messenger = medium_.GetSignalingMessenger( - connection_info->self_id.GetId(), location_hint); + // Create a new ConnectionFlow for this connection attempt. + std::unique_ptr connection_flow = + CreateConnectionFlow(service_id, remote_peer_id); + if (!connection_flow) { + NEARBY_LOG( + INFO, + "Cannot connect to WebRTC peer %s because we failed to create a " + "ConnectionFlow.", + remote_peer_id.GetId().c_str()); + return WebRtcSocketWrapper(); + } - auto signaling_message_callback = std::bind( - [this](ByteArray message, const Role& role, - const std::string& connection_id) { - OffloadFromSignalingThread([this, message{std::move(message)}, - role{role}, - connection_id{connection_id}]() { - ProcessSignalingMessage(role, connection_id, message); - }); - }, - std::placeholders::_1, Role::kOfferer, service_id); + // Create a new SignalingMessenger so that we can communicate over Tachyon. + info.signaling_messenger = + medium_.GetSignalingMessenger(info.self_peer_id.GetId(), location_hint); + if (!info.signaling_messenger->IsValid()) { + NEARBY_LOG( + INFO, + "Cannot connect to WebRTC peer %s because we failed to create a " + "SignalingMessenger.", + remote_peer_id.GetId().c_str()); + connection_flow->Close(); + return WebRtcSocketWrapper(); + } - if (!connection_info->signaling_messenger->IsValid() || - !connection_info->signaling_messenger->StartReceivingMessages( + // This registers ourselves w/ Tachyon, creating a room from the PeerId. + // This allows a remote device to message us over Tachyon. + auto signaling_message_callback = [this, service_id](ByteArray message) { + OffloadFromThread([this, service_id{std::move(service_id)}, + message{std::move(message)}]() { + ProcessTachyonInboxMessage(service_id, message); + }); + }; + if (!info.signaling_messenger->StartReceivingMessages( signaling_message_callback)) { - DisconnectLocked(Role::kOfferer, service_id); + NEARBY_LOG(INFO, + "Cannot connect to WebRTC peer %s because we failed to start " + "receiving messages over Tachyon.", + remote_peer_id.GetId().c_str()); + info.signaling_messenger.reset(); + connection_flow->Close(); + return WebRtcSocketWrapper(); } + + // Poke the remote device. This will cause them to send us an Offer. + if (!info.signaling_messenger->SendMessage( + remote_peer_id.GetId(), + webrtc_frames::EncodeReadyForSignalingPoke(info.self_peer_id))) { + NEARBY_LOG(INFO, + "Cannot connect to WebRTC peer %s because we failed to poke " + "the peer over Tachyon.", + remote_peer_id.GetId().c_str()); + info.signaling_messenger.reset(); + connection_flow->Close(); + return WebRtcSocketWrapper(); + } + + // Create a new ConnectionRequest entry. This map will be used later to look + // up state as we negotiate the connection over Tachyon. + requesting_connections_info_.emplace(remote_peer_id.GetId(), + std::move(info)); + connection_flows_.emplace(remote_peer_id.GetId(), + std::move(connection_flow)); + } + + // Wait for the connection to go through. Don't hold the mutex here so that + // we're not blocking necessary operations. + ExceptionOr socket_result = + socket_future.Get(kDataChannelTimeout); + + { + MutexLock lock(&mutex_); + + // Reclaim our info, since we had released ownership while talking to + // Tachyon. + auto& info = + requesting_connections_info_.find(remote_peer_id.GetId())->second; + + // Verify that the connection went through. + if (!socket_result.ok()) { + NEARBY_LOG(INFO, "Failed to connect to WebRTC peer %s.", + remote_peer_id.GetId().c_str()); + RemoveConnectionFlow(remote_peer_id); + info.signaling_messenger.reset(); + requesting_connections_info_.erase(remote_peer_id.GetId()); + return WebRtcSocketWrapper(); + } + + // Clean up our ConnectionRequest. + info.signaling_messenger.reset(); + requesting_connections_info_.erase(remote_peer_id.GetId()); + + // Return the result. + return socket_result.GetResult(); } } -WebRtc::ConnectionInfo* WebRtc::GetConnectionInfo( - const Role& role, const std::string& connection_id) { - if (role == Role::kOfferer && accepting_map_.contains(connection_id)) { - return &accepting_map_[connection_id]; - } else if (role == Role::kAnswerer && - connecting_map_.contains(connection_id)) { - return &connecting_map_[connection_id]; +void WebRtc::ProcessLocalIceCandidate( + const std::string& service_id, const PeerId& remote_peer_id, + const ::location::nearby::mediums::IceCandidate ice_candidate) { + MutexLock lock(&mutex_); + + // Check first if we have an outgoing request w/ this peer. As this request is + // tied to a specific peer, it takes precedence. + const auto& connection_request_entry = + requesting_connections_info_.find(remote_peer_id.GetId()); + if (connection_request_entry != requesting_connections_info_.end()) { + // Pass the ice candidate to the remote side. + if (!connection_request_entry->second.signaling_messenger->SendMessage( + remote_peer_id.GetId(), + webrtc_frames::EncodeIceCandidates( + connection_request_entry->second.self_peer_id, + {ice_candidate}))) { + NEARBY_LOG(INFO, "Failed to send ice candidate to %s.", + remote_peer_id.GetId().c_str()); + } + + NEARBY_LOG(INFO, "Sent ice candidate to %s.", + remote_peer_id.GetId().c_str()); + return; } - return nullptr; + + // Check next if we're expecting incoming connection requests. + const auto& accepting_connection_entry = + accepting_connections_info_.find(service_id); + if (accepting_connection_entry != accepting_connections_info_.end()) { + // Pass the ice candidate to the remote side. + // TODO(xlythe) Consider not blocking here, since this can eat into the + // connection time + if (!accepting_connection_entry->second.signaling_messenger->SendMessage( + remote_peer_id.GetId(), + webrtc_frames::EncodeIceCandidates( + accepting_connection_entry->second.self_peer_id, + {ice_candidate}))) { + NEARBY_LOG(INFO, "Failed to send ice candidate to %s.", + remote_peer_id.GetId().c_str()); + } + + NEARBY_LOG(INFO, "Sent ice candidate to %s.", + remote_peer_id.GetId().c_str()); + return; + } + + NEARBY_LOG(INFO, + "Skipping restart listening for tachyon inbox messages since we " + "are not accepting connections for service %s.", + service_id.c_str()); } -std::string WebRtc::ConnectionInfo::ToString() const { - std::ostringstream result; - result << (connection_flow == nullptr ? "connection_flow is null, " - : "connection_flow is valid, "); - result << (signaling_messenger == nullptr ? "signaling_messenger is null, " - : "signaling_messenger is valid, "); - result << (socket.IsValid() ? "socket is valid, " : "socket is not valid, "); - result << "remote peer_id: " << peer_id.GetId() << ", "; - result << "self peer_id: " << self_id.GetId(); - return result.str(); +void WebRtc::ProcessTachyonInboxMessage(const std::string& service_id, + const ByteArray& message) { + MutexLock lock(&mutex_); + + // Attempt to parse the incoming message as a WebRtcSignalingFrame. + location::nearby::mediums::WebRtcSignalingFrame frame; + if (!frame.ParseFromString(std::string(message))) { + NEARBY_LOG(WARNING, "Failed to parse signaling message."); + return; + } + + // Ensure that the frame is valid (no missing fields). + if (!frame.has_sender_id()) { + NEARBY_LOG(WARNING, "Invalid WebRTC frame: Sender ID is missing."); + return; + } + PeerId remote_peer_id = PeerId(frame.sender_id().id()); + + // Depending on the message type, we'll respond as appropriate. + if (requesting_connections_info_.contains(remote_peer_id.GetId())) { + // This is from a peer we have an outgoing connection request with, so we'll + // only process the Answer path. + if (frame.has_offer()) { + ReceiveOffer(remote_peer_id, + SessionDescriptionWrapper( + webrtc_frames::DecodeOffer(frame).release())); + SendAnswer(remote_peer_id); + } else if (frame.has_ice_candidates()) { + ReceiveIceCandidates(remote_peer_id, + webrtc_frames::DecodeIceCandidates(frame)); + } else { + NEARBY_LOG(INFO, "Received unknown WebRTC frame: ignoring."); + } + } else if (IsAcceptingConnectionsLocked(service_id)) { + // We don't have an outgoing connection request with this peer, but we are + // accepting incoming requests so we'll only process the Offer path. + if (frame.has_ready_for_signaling_poke()) { + SendOffer(service_id, remote_peer_id); + } else if (frame.has_answer()) { + ReceiveAnswer(remote_peer_id, + SessionDescriptionWrapper( + webrtc_frames::DecodeAnswer(frame).release())); + } else if (frame.has_ice_candidates()) { + ReceiveIceCandidates(remote_peer_id, + webrtc_frames::DecodeIceCandidates(frame)); + } else { + NEARBY_LOG(INFO, "Received unknown WebRTC frame: ignoring."); + } + } else { + NEARBY_LOG( + INFO, + "Ignoring Tachyon message since we are not accepting connections."); + } } -std::string WebRtc::InternalStatesToString() { - std::ostringstream map_values; - map_values << "connecting map size: " << connecting_map_.size() - << ", accepting map size: " << accepting_map_.size() << "\n"; - for (auto& item : connecting_map_) { - map_values << "connecting " << item.first << ": " << item.second.ToString() - << "\n"; +void WebRtc::SendOffer(const std::string& service_id, + const PeerId& remote_peer_id) { + std::unique_ptr connection_flow = + CreateConnectionFlow(service_id, remote_peer_id); + if (!connection_flow) { + NEARBY_LOG(INFO, + "Unable to send offer. Failed to create a ConnectionFlow."); + return; } - for (auto& item : accepting_map_) { - map_values << "accepting " << item.first << ": " << item.second.ToString() - << "\n"; + + SessionDescriptionWrapper offer = connection_flow->CreateOffer(); + const webrtc::SessionDescriptionInterface& sdp = offer.GetSdp(); + if (!connection_flow->SetLocalSessionDescription(std::move(offer))) { + NEARBY_LOG(INFO, + "Unable to send offer. Failed to register our offer locally."); + RemoveConnectionFlow(remote_peer_id); + return; } - return map_values.str(); + + // Grab our info from the map. + auto& info = accepting_connections_info_.find(service_id)->second; + + // Pass the offer to the remote side. + if (!info.signaling_messenger->SendMessage( + remote_peer_id.GetId(), + webrtc_frames::EncodeOffer(info.self_peer_id, sdp))) { + NEARBY_LOG(INFO, + "Unable to send offer. Failed to write the offer to the remote " + "peer %s.", + remote_peer_id.GetId().c_str()); + RemoveConnectionFlow(remote_peer_id); + return; + } + + // Store the ConnectionFlow so that other methods can use it later. + connection_flows_.emplace(remote_peer_id.GetId(), std::move(connection_flow)); + NEARBY_LOG(INFO, "Sent offer to %s.", remote_peer_id.GetId().c_str()); +} + +void WebRtc::ReceiveOffer(const PeerId& remote_peer_id, + SessionDescriptionWrapper offer) { + const auto& entry = connection_flows_.find(remote_peer_id.GetId()); + if (entry == connection_flows_.end()) { + NEARBY_LOG(INFO, + "Unable to receive offer. Failed to create a ConnectionFlow."); + return; + } + + if (!entry->second->OnOfferReceived(offer)) { + NEARBY_LOG(INFO, "Unable to receive offer. Failed to process the offer."); + RemoveConnectionFlow(remote_peer_id); + } +} + +void WebRtc::SendAnswer(const PeerId& remote_peer_id) { + const auto& entry = connection_flows_.find(remote_peer_id.GetId()); + if (entry == connection_flows_.end()) { + NEARBY_LOG(INFO, + "Unable to send answer. Failed to create a ConnectionFlow."); + return; + } + + SessionDescriptionWrapper answer = entry->second->CreateAnswer(); + const webrtc::SessionDescriptionInterface& sdp = answer.GetSdp(); + if (!entry->second->SetLocalSessionDescription(std::move(answer))) { + NEARBY_LOG(INFO, + "Unable to send answer. Failed to register our answer locally."); + RemoveConnectionFlow(remote_peer_id); + return; + } + + // Grab our info from the map. + const auto& connection_request_entry = + requesting_connections_info_.find(remote_peer_id.GetId()); + if (connection_request_entry == requesting_connections_info_.end()) { + NEARBY_LOG(INFO, + "Unable to send answer. Failed to find an outgoing connection " + "request."); + RemoveConnectionFlow(remote_peer_id); + return; + } + + // Pass the answer to the remote side. + if (!connection_request_entry->second.signaling_messenger->SendMessage( + remote_peer_id.GetId(), + webrtc_frames::EncodeAnswer( + connection_request_entry->second.self_peer_id, sdp))) { + NEARBY_LOG( + INFO, + "Unable to send answer. Failed to write the answer to the remote " + "peer %s.", + remote_peer_id.GetId().c_str()); + RemoveConnectionFlow(remote_peer_id); + return; + } + + NEARBY_LOG(INFO, "Sent answer to %s.", remote_peer_id.GetId().c_str()); +} + +void WebRtc::ReceiveAnswer(const PeerId& remote_peer_id, + SessionDescriptionWrapper answer) { + const auto& entry = connection_flows_.find(remote_peer_id.GetId()); + if (entry == connection_flows_.end()) { + NEARBY_LOG(INFO, + "Unable to receive answer. Failed to create a ConnectionFlow."); + return; + } + + if (!entry->second->OnAnswerReceived(answer)) { + NEARBY_LOG(INFO, "Unable to receive answer. Failed to process the answer."); + RemoveConnectionFlow(remote_peer_id); + } +} + +void WebRtc::ReceiveIceCandidates( + const PeerId& remote_peer_id, + std::vector> + ice_candidates) { + const auto& entry = connection_flows_.find(remote_peer_id.GetId()); + if (entry == connection_flows_.end()) { + NEARBY_LOG( + INFO, + "Unable to receive ice candidates. Failed to create a ConnectionFlow."); + return; + } + + entry->second->OnRemoteIceCandidatesReceived(std::move(ice_candidates)); +} + +void WebRtc::ProcessRestartTachyonReceiveMessages( + const std::string& service_id) { + MutexLock lock(&mutex_); + if (!IsAcceptingConnectionsLocked(service_id)) { + NEARBY_LOG(INFO, + "Skipping restart listening for tachyon inbox messages since we " + "are not accepting connections for service %s.", + service_id.c_str()); + return; + } + + // Grab our info from the map. + auto& info = accepting_connections_info_.find(service_id)->second; + + // Ensure we've disconnected from Tachyon. + info.signaling_messenger->StopReceivingMessages(); + + // Attempt to re-register. + auto signaling_message_callback = [this, service_id](ByteArray message) { + OffloadFromThread([this, service_id{std::move(service_id)}, + message{std::move(message)}]() { + ProcessTachyonInboxMessage(service_id, message); + }); + }; + if (!info.signaling_messenger->StartReceivingMessages( + signaling_message_callback)) { + NEARBY_LOG(WARNING, + "Failed to restart listening for tachyon inbox messages for " + "service %s since we failed to reach Tachyon.", + service_id.c_str()); + return; + } + + NEARBY_LOG(INFO, + "Successfully restarted listening for tachyon inbox messages on " + "service %s.", + service_id.c_str()); +} + +void WebRtc::ProcessDataChannelCreated( + const std::string& service_id, const PeerId& remote_peer_id, + rtc::scoped_refptr data_channel) { + MutexLock lock(&mutex_); + + // Transform the DataChannel into a socket. + auto socket = std::make_unique("WebRtcSocket", data_channel); + socket->SetOnSocketClosedListener({[this, remote_peer_id]() { + OffloadFromThread( + [this, remote_peer_id]() { ProcessDataChannelClosed(remote_peer_id); }); + }}); + WebRtcSocketWrapper wrapper = WebRtcSocketWrapper(std::move(socket)); + + // Store this DataChannel so that we can update it later. + sockets_.emplace(remote_peer_id.GetId(), wrapper); + + // Notify the client of the newly formed socket. + const auto& connection_request_entry = + requesting_connections_info_.find(remote_peer_id.GetId()); + if (connection_request_entry != requesting_connections_info_.end()) { + connection_request_entry->second.socket_future.Set(wrapper); + return; + } + + const auto& accepting_connection_entry = + accepting_connections_info_.find(service_id); + if (accepting_connection_entry != accepting_connections_info_.end()) { + accepting_connection_entry->second.accepted_connection_callback.accepted_cb( + wrapper); + return; + } + + // No one to handle the newly created DataChannel, so we'll just close it. + wrapper.Close(); + NEARBY_LOG(INFO, + "Ignoring new DataChannel because we " + "are not accepting connections for service %s.", + service_id.c_str()); +} + +void WebRtc::ProcessDataChannelMessage(const PeerId& remote_peer_id, + const ByteArray& message) { + MutexLock lock(&mutex_); + const auto& entry = sockets_.find(remote_peer_id.GetId()); + if (entry == sockets_.end()) { + return; + } + + entry->second.NotifyDataChannelMsgReceived(message); +} + +void WebRtc::ProcessDataChannelBufferAmountChanged( + const PeerId& remote_peer_id) { + MutexLock lock(&mutex_); + const auto& entry = sockets_.find(remote_peer_id.GetId()); + if (entry == sockets_.end()) { + return; + } + + entry->second.NotifyDataChannelBufferedAmountChanged(); +} + +void WebRtc::ProcessDataChannelClosed(const PeerId& remote_peer_id) { + MutexLock lock(&mutex_); + const auto& entry = sockets_.find(remote_peer_id.GetId()); + if (entry == sockets_.end()) { + return; + } + + entry->second.Close(); + sockets_.erase(remote_peer_id.GetId()); + + RemoveConnectionFlow(remote_peer_id); +} + +std::unique_ptr WebRtc::CreateConnectionFlow( + const std::string& service_id, const PeerId& remote_peer_id) { + RemoveConnectionFlow(remote_peer_id); + + return ConnectionFlow::Create( + {.local_ice_candidate_found_cb = + {[this, service_id, remote_peer_id]( + const webrtc::IceCandidateInterface* ice_candidate) { + // Note: We need to encode the ice candidate here, before we jump + // off the thread. Otherwise, it gets destroyed and we can't read + // it later. + ::location::nearby::mediums::IceCandidate encoded_ice_candidate = + webrtc_frames::EncodeIceCandidate(*ice_candidate); + OffloadFromThread( + [this, service_id, remote_peer_id, encoded_ice_candidate]() { + ProcessLocalIceCandidate(service_id, remote_peer_id, + encoded_ice_candidate); + }); + }}}, + { + .data_channel_created_cb = + {[this, service_id, + remote_peer_id](rtc::scoped_refptr + data_channel) { + OffloadFromThread( + [this, service_id, remote_peer_id, data_channel]() { + ProcessDataChannelCreated(service_id, remote_peer_id, + data_channel); + }); + }}, + .data_channel_message_received_cb = {[this, remote_peer_id]( + const ByteArray& message) { + OffloadFromThread([this, remote_peer_id, message]() { + ProcessDataChannelMessage(remote_peer_id, message); + }); + }}, + .data_channel_buffered_amount_changed_cb = {[this, remote_peer_id]() { + OffloadFromThread([this, remote_peer_id]() { + ProcessDataChannelBufferAmountChanged(remote_peer_id); + }); + }}, + .data_channel_closed_cb = {[this, remote_peer_id]() { + OffloadFromThread([this, remote_peer_id]() { + ProcessDataChannelClosed(remote_peer_id); + }); + }}, + }, + medium_); +} + +void WebRtc::RemoveConnectionFlow(const PeerId& remote_peer_id) { + const auto& entry = connection_flows_.find(remote_peer_id.GetId()); + if (entry == connection_flows_.end()) { + return; + } + + entry->second->Close(); + connection_flows_.erase(remote_peer_id.GetId()); + + // If we had an outgoing connection request w/ this peer, report the failure + // to the future that's being waited on. + const auto& connection_request_entry = + requesting_connections_info_.find(remote_peer_id.GetId()); + if (connection_request_entry != requesting_connections_info_.end()) { + connection_request_entry->second.socket_future.SetException( + {Exception::kFailed}); + } +} + +void WebRtc::OffloadFromThread(Runnable runnable) { + single_thread_executor_.Execute(std::move(runnable)); } } // namespace mediums diff --git a/cpp/core/internal/mediums/webrtc.h b/cpp/core/internal/mediums/webrtc.h index 3f098b5b..99cccc69 100644 --- a/cpp/core/internal/mediums/webrtc.h +++ b/cpp/core/internal/mediums/webrtc.h @@ -63,7 +63,7 @@ class WebRtc { // boolean value indicating if the device has started accepting connections. // Runs on @MainThread. bool StartAcceptingConnections(const std::string& service_id, - const PeerId& self_id, + const PeerId& self_peer_id, const LocationHint& location_hint, AcceptedConnectionCallback callback) ABSL_LOCKS_EXCLUDED(mutex_); @@ -73,9 +73,11 @@ class WebRtc { void StopAcceptingConnections(const std::string& service_id) ABSL_LOCKS_EXCLUDED(mutex_); - // Initiates a WebRtc connection with peer device identified by |peer_id|. + // Initiates a WebRtc connection with peer device identified by |peer_id| + // with internal retry for maximum attempts of kConnectAttemptsLimit. // Runs on @MainThread. - WebRtcSocketWrapper Connect(const PeerId& peer_id, + WebRtcSocketWrapper Connect(const std::string& service_id, + const PeerId& peer_id, const LocationHint& location_hint) ABSL_LOCKS_EXCLUDED(mutex_); @@ -86,138 +88,152 @@ class WebRtc { kAnswerer = 2, }; - absl::flat_hash_map role_names_{ - {Role::kNone, "None"}, - {Role::kOfferer, "Offerer"}, - {Role::kAnswerer, "Answerer"}}; + struct AcceptingConnectionsInfo { + // The self_peer_id is generated from the BT/WiFi advertisements and allows + // the scanner to message us over Tachyon. + PeerId self_peer_id; - struct ConnectionInfo { - std::unique_ptr connection_flow; + // The registered callback. When there's an incoming connection, this + // callback is notified. + AcceptedConnectionCallback accepted_connection_callback; + + // Allows us to communicate with the Tachyon web server. std::unique_ptr signaling_messenger; - WebRtcSocketWrapper socket; - CancelableAlarm restart_receive_messages_alarm; - PeerId self_id; - PeerId peer_id; - ByteArray pending_local_offer; - std::vector<::location::nearby::mediums::IceCandidate> - pending_local_ice_candidates; - std::string ToString() const; + // Restarts the tachyon inbox receives messages streaming rpc if the + // streaming rpc times out. The streaming rpc times out after 60s while + // advertising. Non-null when listening for WebRTC connections as an + // offerer. + CancelableAlarm restart_tachyon_receive_messages_alarm; }; - bool InitWebRtcFlow(const Role& role, const PeerId& self_id, - const LocationHint& location_hint, - const std::string& connection_id) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + struct ConnectionRequestInfo { + // The self_peer_id is randomly generated and allows the advertiser to + // message us over Tachyon. + PeerId self_peer_id; - Future ListenForWebRtcSocketFuture( - const Role& role, const std::string& connection_id, - Future> - data_channel_future, - AcceptedConnectionCallback callback); + // Allows us to communicate with the Tachyon web server. + std::unique_ptr signaling_messenger; - WebRtcSocketWrapper CreateWebRtcSocketWrapper( - const Role& role, const std::string& connection_id, - rtc::scoped_refptr data_channel); + // The pending DataChannel future. Our client will be blocked on this while + // they wait for us to set up the channel over Tachyon. + Future socket_future; + }; - LocalIceCandidateListener GetLocalIceCandidateListener( - const Role& role, const std::string& connection_id); - void OnLocalIceCandidate( - const Role& role, const std::string& connection_id, - const webrtc::IceCandidateInterface* local_ice_candidate); - - DataChannelListener GetDataChannelListener(const Role& role, - const std::string& connection_id); - void OnDataChannelClosed(const Role& role, const std::string& connection_id); - void OnDataChannelMessageReceived(const Role& role, - const std::string& connection_id, - const ByteArray& message); - void OnDataChannelBufferedAmountChanged(const Role& role, - const std::string& connection_id); - - // Runs on @MainThread and |single_thread_executor_|. - bool SetLocalSessionDescription(SessionDescriptionWrapper sdp, Role role, - const std::string& connection_id) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on |single_thread_executor_|. - bool IsSignaling(const Role& role, const std::string& connection_id) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on |single_thread_executor_|. - void ProcessSignalingMessage(const Role& role, - const std::string& connection_id, - const ByteArray& message) - ABSL_LOCKS_EXCLUDED(mutex_); - - // Runs on |single_thread_executor_|. - void SendOfferAndIceCandidatesToPeer(const std::string& service_id) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on |single_thread_executor_|. - void SendAnswerToPeer(const std::string& peer_id) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on @MainThread and |single_thread_executor_|. - void LogAndDisconnect(const Role& role, const std::string& connection_id, - const std::string& error_message) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + static constexpr int kConnectAttemptsLimit = 3; + // Attempt to initiates a WebRtc connection with peer device identified by + // |peer_id|. // Runs on @MainThread. - void Disconnect(const Role& role, const std::string& connection_id) + WebRtcSocketWrapper AttemptToConnect(const std::string& service_id, + const PeerId& peer_id, + const LocationHint& location_hint) ABSL_LOCKS_EXCLUDED(mutex_); - // Runs on @MainThread and |single_thread_executor_|. - void DisconnectLocked(const Role& role, const std::string& connection_id) + // Returns if the device is accepting connection with specific service id. + // Runs on @MainThread. + bool IsAcceptingConnectionsLocked(const std::string& service_id) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - void LogAndShutdownSignaling(const Role& role, - const std::string& connection_id, - const std::string& error_message) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on @MainThread and |single_thread_executor_|. - void ShutdownSignaling(const Role& role, const std::string& connection_id) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on @MainThread and |single_thread_executor_|. - void ShutdownWebRtcSocket(const Role& role, const std::string& connection_id) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - // Runs on @MainThread and |single_thread_executor_|. - void ShutdownIceCandidateCollection(const Role& role, - const std::string& connection_id) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - - void OffloadFromSignalingThread(Runnable runnable); - - // Runs on |restart_receive_messages_executor_|. - void RestartReceiveMessages(const LocationHint& location_hint, - const std::string& service_id) + // Runs on |single_thread_executor_|. + void ProcessTachyonInboxMessage(const std::string& service_id, + const ByteArray& message) ABSL_LOCKS_EXCLUDED(mutex_); - void PrintStatus(const std::string& func); - - ConnectionInfo* GetConnectionInfo(const Role& role, - const std::string& connection_id) + // Runs on |single_thread_executor_|. + void SendOffer(const std::string& service_id, const PeerId& remote_peer_id) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - std::string InternalStatesToString() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + // Runs on |single_thread_executor_|. + void ReceiveOffer(const PeerId& remote_peer_id, + SessionDescriptionWrapper offer) + ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + // Runs on |single_thread_executor_|. + void SendAnswer(const PeerId& remote_peer_id) + ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + // Runs on |single_thread_executor_|. + void ReceiveAnswer(const PeerId& remote_peer_id, + SessionDescriptionWrapper answer) + ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + // Runs on |single_thread_executor_|. + void ReceiveIceCandidates( + const PeerId& remote_peer_id, + std::vector> + ice_candidates) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + // Runs on |single_thread_executor_|. + std::unique_ptr CreateConnectionFlow( + const std::string& service_id, const PeerId& remote_peer_id) + ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + // Runs on |single_thread_executor_|. + std::unique_ptr GetConnectionFlow( + const PeerId& remote_peer_id) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + // Runs on |single_thread_executor_|. + void RemoveConnectionFlow(const PeerId& remote_peer_id) + ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + + // Runs on |single_thread_executor_|. + void ProcessDataChannelCreated( + const std::string& service_id, const PeerId& remote_peer_id, + rtc::scoped_refptr data_channel) + ABSL_LOCKS_EXCLUDED(mutex_); + + // Runs on |single_thread_executor_|. + void ProcessDataChannelMessage(const PeerId& remote_peer_id, + const ByteArray& message) + ABSL_LOCKS_EXCLUDED(mutex_); + + // Runs on |single_thread_executor_|. + void ProcessDataChannelBufferAmountChanged(const PeerId& remote_peer_id) + ABSL_LOCKS_EXCLUDED(mutex_); + + // Runs on |single_thread_executor_|. + void ProcessDataChannelClosed(const PeerId& remote_peer_id) + ABSL_LOCKS_EXCLUDED(mutex_); + + // Runs on |single_thread_executor_|. + void ProcessLocalIceCandidate( + const std::string& service_id, const PeerId& remote_peer_id, + const ::location::nearby::mediums::IceCandidate ice_candidate) + ABSL_LOCKS_EXCLUDED(mutex_); + + // Runs on |single_thread_executor_|. + void ProcessRestartTachyonReceiveMessages(const std::string& service_id) + ABSL_LOCKS_EXCLUDED(mutex_); + + void OffloadFromThread(Runnable runnable); Mutex mutex_; WebRtcMedium medium_; - SingleThreadExecutor single_thread_executor_; + // The single thread we throw the potentially blocking work on to. + ScheduledExecutor single_thread_executor_; - // Restarts the signaling messenger for receiving messages. - ScheduledExecutor restart_receive_messages_executor_; + // A map of ServiceID -> State for all services that are listening for + // incoming connections. + absl::flat_hash_map + accepting_connections_info_ ABSL_GUARDED_BY(mutex_); - // Use service_id as key for accepting connections. - absl::flat_hash_map accepting_map_ - ABSL_GUARDED_BY(mutex_); - // Use remote peer_id as key for connecting connections. - absl::flat_hash_map connecting_map_ + // A map of a remote PeerId -> State for pending connection requests. As + // messages from Tachyon come in, this lets us look up the connection request + // info to handle the interaction. + absl::flat_hash_map + requesting_connections_info_ ABSL_GUARDED_BY(mutex_); + + // A map of a remote PeerId -> ConnectionFlow. For each connection, we create + // a unique ConnectionFlow. + absl::flat_hash_map> + connection_flows_ ABSL_GUARDED_BY(mutex_); + + // A map of a remote PeerId -> Socket. Non-empty while we have active + // connections. + absl::flat_hash_map sockets_ ABSL_GUARDED_BY(mutex_); }; diff --git a/cpp/core/internal/mediums/webrtc/connection_flow.cc b/cpp/core/internal/mediums/webrtc/connection_flow.cc index fb724487..58c70649 100644 --- a/cpp/core/internal/mediums/webrtc/connection_flow.cc +++ b/cpp/core/internal/mediums/webrtc/connection_flow.cc @@ -91,6 +91,11 @@ ConnectionFlow::ConnectionFlow( ConnectionFlow::~ConnectionFlow() { Close(); } +ConnectionFlow::State ConnectionFlow::GetState() { + MutexLock lock(&mutex_); + return state_; +} + SessionDescriptionWrapper ConnectionFlow::CreateOffer() { MutexLock lock(&mutex_); @@ -219,11 +224,6 @@ bool ConnectionFlow::OnRemoteIceCandidatesReceived( return true; } -Future> -ConnectionFlow::GetDataChannel() { - return data_channel_future_; -} - bool ConnectionFlow::Close() { MutexLock lock(&mutex_); return CloseLocked(); @@ -275,16 +275,20 @@ void ConnectionFlow::ProcessOnPeerConnectionChange( if (new_state == PeerConnectionState::kClosed || new_state == PeerConnectionState::kFailed || new_state == PeerConnectionState::kDisconnected) { - MutexLock lock(&mutex_); - CloseAndNotifyLocked(); + Close(); } } -void ConnectionFlow::ProcessDataChannelConnected() { +void ConnectionFlow::ProcessDataChannelConnected( + rtc::scoped_refptr data_channel) { MutexLock lock(&mutex_); NEARBY_LOG(INFO, "Data channel state changed to connected."); - if (!TransitionState(State::kWaitingToConnect, State::kConnected)) - CloseAndNotifyLocked(); + if (!TransitionState(State::kWaitingToConnect, State::kConnected)) { + data_channel->Close(); + return; + } + + data_channel_listener_.data_channel_created_cb(std::move(data_channel)); } webrtc::DataChannelObserver* ConnectionFlow::CreateDataChannelObserver( @@ -294,15 +298,14 @@ webrtc::DataChannelObserver* ConnectionFlow::CreateDataChannelObserver( data_channel{std::move(data_channel)}]() { if (data_channel->state() == webrtc::DataChannelInterface::DataState::kOpen) { - data_channel_future_.Set(std::move(data_channel)); - OffloadFromSignalingThread([this]() { ProcessDataChannelConnected(); }); + OffloadFromSignalingThread( + [this, data_channel{std::move(data_channel)}]() { + ProcessDataChannelConnected(std::move(data_channel)); + }); } else if (data_channel->state() == webrtc::DataChannelInterface::DataState::kClosed) { data_channel->UnregisterObserver(); - OffloadFromSignalingThread([this]() { - MutexLock lock(&mutex_); - CloseAndNotifyLocked(); - }); + data_channel_listener_.data_channel_closed_cb(); } }; data_channel_observer_ = absl::make_unique( @@ -325,19 +328,12 @@ bool ConnectionFlow::TransitionState(State current_state, State new_state) { return true; } -void ConnectionFlow::CloseAndNotifyLocked() { - if (CloseLocked()) { - data_channel_listener_.data_channel_closed_cb(); - } -} - bool ConnectionFlow::CloseLocked() { if (state_ == State::kEnded) { return false; } state_ = State::kEnded; - data_channel_future_.SetException({Exception::kInterrupted}); if (peer_connection_) peer_connection_->Close(); data_channel_observer_.reset(); diff --git a/cpp/core/internal/mediums/webrtc/connection_flow.h b/cpp/core/internal/mediums/webrtc/connection_flow.h index 47fbb58f..969ecb7f 100644 --- a/cpp/core/internal/mediums/webrtc/connection_flow.h +++ b/cpp/core/internal/mediums/webrtc/connection_flow.h @@ -9,7 +9,6 @@ #include "core/internal/mediums/webrtc/peer_connection_observer_impl.h" #include "core/internal/mediums/webrtc/session_description_wrapper.h" #include "platform/base/runnable.h" -#include "platform/public/future.h" #include "platform/public/single_thread_executor.h" #include "platform/public/webrtc.h" #include "webrtc/api/data_channel_interface.h" @@ -55,12 +54,26 @@ namespace mediums { */ class ConnectionFlow { public: + enum class State { + kInitialized, + kCreatingOffer, + kWaitingForAnswer, + kReceivedOffer, + kCreatingAnswer, + kWaitingToConnect, + kConnected, + kEnded, + }; + // This method blocks on the creation of the peer connection object. static std::unique_ptr Create( LocalIceCandidateListener local_ice_candidate_listener, DataChannelListener data_channel_listener, WebRtcMedium& webrtc_medium); ~ConnectionFlow(); + // Returns the current state of the ConnectionFlow. + State GetState() ABSL_LOCKS_EXCLUDED(mutex_); + // Create the offer that will be sent to the remote. Mirrors the behaviour of // PeerConnectionInterface::CreateOffer. SessionDescriptionWrapper CreateOffer() ABSL_LOCKS_EXCLUDED(mutex_); @@ -86,8 +99,6 @@ class ConnectionFlow { bool OnRemoteIceCandidatesReceived( std::vector> ice_candidates) ABSL_LOCKS_EXCLUDED(mutex_); - // Get a future for the data channel. - Future> GetDataChannel(); // Close the peer connection and data channel. bool Close() ABSL_LOCKS_EXCLUDED(mutex_); @@ -103,17 +114,6 @@ class ConnectionFlow { ABSL_LOCKS_EXCLUDED(mutex_); private: - enum class State { - kInitialized, - kCreatingOffer, - kWaitingForAnswer, - kReceivedOffer, - kCreatingAnswer, - kWaitingToConnect, - kConnected, - kEnded, - }; - ConnectionFlow(LocalIceCandidateListener local_ice_candidate_listener, DataChannelListener data_channel_listener); @@ -127,7 +127,9 @@ class ConnectionFlow { bool SetRemoteSessionDescription(SessionDescriptionWrapper sdp); - void ProcessDataChannelConnected() ABSL_LOCKS_EXCLUDED(mutex_); + void ProcessDataChannelConnected( + rtc::scoped_refptr) + ABSL_LOCKS_EXCLUDED(mutex_); void CloseAndNotifyLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); bool CloseLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); @@ -141,8 +143,6 @@ class ConnectionFlow { std::unique_ptr data_channel_observer_; - Future> data_channel_future_; - PeerConnectionObserverImpl peer_connection_observer_; rtc::scoped_refptr peer_connection_; diff --git a/cpp/core/internal/mediums/webrtc/connection_flow_test.cc b/cpp/core/internal/mediums/webrtc/connection_flow_test.cc index 6dec5c59..deb605fc 100644 --- a/cpp/core/internal/mediums/webrtc/connection_flow_test.cc +++ b/cpp/core/internal/mediums/webrtc/connection_flow_test.cc @@ -43,6 +43,11 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) { Future message_received_future; + Future> + offerer_data_channel_future; + Future> + answerer_data_channel_future; + std::unique_ptr offerer, answerer; // Send Ice Candidates immediately when you retrieve them @@ -56,7 +61,12 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) { if (answerer) answerer->OnRemoteIceCandidatesReceived(std::move(vec)); }}, - DataChannelListener(), webrtc_medium_offerer); + {.data_channel_created_cb = + [&offerer_data_channel_future]( + rtc::scoped_refptr data_channel) { + offerer_data_channel_future.Set(std::move(data_channel)); + }}, + webrtc_medium_offerer); ASSERT_NE(offerer, nullptr); answerer = ConnectionFlow::Create( {.local_ice_candidate_found_cb = @@ -68,7 +78,12 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) { if (offerer) offerer->OnRemoteIceCandidatesReceived(std::move(vec)); }}, - {.data_channel_message_received_cb = + {.data_channel_created_cb = + [&answerer_data_channel_future]( + rtc::scoped_refptr data_channel) { + answerer_data_channel_future.Set(std::move(data_channel)); + }, + .data_channel_message_received_cb = [&message_received_future](ByteArray bytes) { message_received_future.Set(std::move(bytes)); }}, @@ -89,10 +104,10 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) { // Retrieve Data Channels ExceptionOr> - offerer_channel = offerer->GetDataChannel().Get(absl::Seconds(1)); + offerer_channel = offerer_data_channel_future.Get(absl::Seconds(1)); EXPECT_TRUE(offerer_channel.ok()); ExceptionOr> - answerer_channel = answerer->GetDataChannel().Get(absl::Seconds(1)); + answerer_channel = answerer_data_channel_future.Get(absl::Seconds(1)); EXPECT_TRUE(answerer_channel.ok()); // Send message on data channel diff --git a/cpp/core/internal/mediums/webrtc/data_channel_listener.h b/cpp/core/internal/mediums/webrtc/data_channel_listener.h index 319889fd..92b4c05a 100644 --- a/cpp/core/internal/mediums/webrtc/data_channel_listener.h +++ b/cpp/core/internal/mediums/webrtc/data_channel_listener.h @@ -11,7 +11,10 @@ namespace mediums { // Callbacks from the data channel. struct DataChannelListener { - std::function data_channel_closed_cb = DefaultCallback<>(); + // Called when the data channel is created. + std::function)> + data_channel_created_cb = + DefaultCallback>(); // Called when a new message was received on the data channel. std::function data_channel_message_received_cb = @@ -21,6 +24,9 @@ struct DataChannelListener { // changed. std::function data_channel_buffered_amount_changed_cb = DefaultCallback<>(); + + // Called when the data channel is closed. + std::function data_channel_closed_cb = DefaultCallback<>(); }; } // namespace mediums diff --git a/cpp/core/internal/mediums/webrtc_test.cc b/cpp/core/internal/mediums/webrtc_test.cc index 117ffad7..e5d9e36e 100644 --- a/cpp/core/internal/mediums/webrtc_test.cc +++ b/cpp/core/internal/mediums/webrtc_test.cc @@ -13,7 +13,7 @@ namespace connections { namespace mediums { namespace { -const int kTwoMBSize = 2000000; + class WebRtcTest : public ::testing::Test { protected: WebRtcTest() { @@ -61,7 +61,8 @@ TEST_F(WebRtcTest, Connect_DataChannelTimeOut) { LocationHint location_hint; ASSERT_TRUE(webrtc.IsAvailable()); - WebRtcSocketWrapper wrapper_1 = webrtc.Connect(peer_id, location_hint); + WebRtcSocketWrapper wrapper_1 = + webrtc.Connect(service_id, peer_id, location_hint); EXPECT_FALSE(wrapper_1.IsValid()); EXPECT_TRUE(webrtc.StartAcceptingConnections( @@ -85,7 +86,7 @@ TEST_F(WebRtcTest, StartAcceptingConnection_ThenConnect) { service_id, self_id, location_hint, {mock_accepted_callback_.AsStdFunction()})); WebRtcSocketWrapper wrapper = - webrtc.Connect(PeerId("random_peer_id"), location_hint); + webrtc.Connect(service_id, PeerId("random_peer_id"), location_hint); EXPECT_TRUE(webrtc.IsAcceptingConnections(service_id)); EXPECT_FALSE(wrapper.IsValid()); EXPECT_FALSE(webrtc.StartAcceptingConnections( @@ -135,14 +136,15 @@ TEST_F(WebRtcTest, ConnectTwice) { device_c.StartAcceptingConnections(service_id, other_id, location_hint, {[](WebRtcSocketWrapper wrapper) {}}); - sender_socket = sender.Connect(self_id, location_hint); + sender_socket = sender.Connect(service_id, self_id, location_hint); EXPECT_TRUE(sender_socket.IsValid()); ExceptionOr devices_connected = connected.Get(); ASSERT_TRUE(devices_connected.ok()); EXPECT_TRUE(devices_connected.result()); - WebRtcSocketWrapper socket = sender.Connect(other_id, location_hint); + WebRtcSocketWrapper socket = + sender.Connect(service_id, other_id, location_hint); EXPECT_TRUE(socket.IsValid()); socket.Close(); @@ -176,7 +178,7 @@ TEST_F(WebRtcTest, ConnectBothDevicesAndAbort) { connected.Set(receiver_socket.IsValid()); }}); - sender_socket = sender.Connect(self_id, location_hint); + sender_socket = sender.Connect(service_id, self_id, location_hint); EXPECT_TRUE(sender_socket.IsValid()); ExceptionOr devices_connected = connected.Get(); @@ -204,7 +206,7 @@ TEST_F(WebRtcTest, ConnectBothDevicesAndSendData) { connected.Set(receiver_socket.IsValid()); }}); - sender_socket = sender.Connect(self_id, location_hint); + sender_socket = sender.Connect(service_id, self_id, location_hint); EXPECT_TRUE(sender_socket.IsValid()); ExceptionOr devices_connected = connected.Get(); @@ -238,7 +240,7 @@ TEST_F(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) { connected.Set(receiver_socket.IsValid()); }}); - sender_socket = sender.Connect(self_id, location_hint); + sender_socket = sender.Connect(service_id, self_id, location_hint); EXPECT_TRUE(sender_socket.IsValid()); ExceptionOr devices_connected = connected.Get(); @@ -255,83 +257,6 @@ TEST_F(WebRtcTest, ConnectBothDevices_ShutdownSignaling_SendData) { EXPECT_EQ(message, received_msg.result()); } -// Tests the flow when the two devices created two data channel and transfer -// data in the same time. -TEST_F(WebRtcTest, TwoChannels_SendData) { - WebRtc receiver, sender; - WebRtcSocketWrapper receiver_socket1, receiver_socket2, sender_socket1, - sender_socket2; - const PeerId self_id1("self_id1"), self_id2("self_id2"); - const std::string service_id1("service1"), service_id2("service2"); - LocationHint location_hint; - Future connected1, connected2; - ByteArray message; - message.SetData(kTwoMBSize / 10, 'c'); - - receiver.StartAcceptingConnections( - service_id1, self_id1, location_hint, - {[&receiver_socket1, connected1](WebRtcSocketWrapper wrapper) mutable { - receiver_socket1 = wrapper; - connected1.Set(receiver_socket1.IsValid()); - }}); - - receiver.StartAcceptingConnections( - service_id2, self_id2, location_hint, - {[&receiver_socket2, connected2](WebRtcSocketWrapper wrapper) mutable { - receiver_socket2 = wrapper; - connected2.Set(receiver_socket2.IsValid()); - }}); - - sender_socket1 = sender.Connect(self_id1, location_hint); - EXPECT_TRUE(sender_socket1.IsValid()); - - sender_socket2 = sender.Connect(self_id2, location_hint); - EXPECT_TRUE(sender_socket2.IsValid()); - - ExceptionOr devices_connected1 = connected1.Get(); - ASSERT_TRUE(devices_connected1.ok()); - EXPECT_TRUE(devices_connected1.result()); - - ExceptionOr devices_connected2 = connected1.Get(); - ASSERT_TRUE(devices_connected2.ok()); - EXPECT_TRUE(devices_connected2.result()); - - // Only shuts down signaling channel. - receiver.StopAcceptingConnections(service_id1); - receiver.StopAcceptingConnections(service_id2); - - for (int i = 0; i < 10; i++) { - sender_socket1.GetOutputStream().Write(message); - sender_socket2.GetOutputStream().Write(message); - ExceptionOr received_msg1 = - receiver_socket1.GetInputStream().Read(kTwoMBSize / 10); - ASSERT_TRUE(received_msg1.ok()); - ExceptionOr received_msg2 = - receiver_socket2.GetInputStream().Read(kTwoMBSize / 10); - EXPECT_EQ(message, received_msg1.result()); - EXPECT_EQ(message, received_msg2.result()); - } -} - -TEST_F(WebRtcTest, StartAcceptingConnections_NullPeerConnection) { - using MockAcceptedCallback = - testing::MockFunction; - testing::StrictMock mock_accepted_callback_; - - MediumEnvironment::Instance().SetUseValidPeerConnection( - /*use_valid_peer_connection=*/false); - - WebRtc webrtc; - PeerId self_id("peer_id"); - const std::string service_id("NearbySharing"); - LocationHint location_hint; - - ASSERT_TRUE(webrtc.IsAvailable()); - EXPECT_FALSE(webrtc.StartAcceptingConnections( - service_id, self_id, location_hint, - {mock_accepted_callback_.AsStdFunction()})); -} - TEST_F(WebRtcTest, Connect_NullPeerConnection) { using MockAcceptedCallback = testing::MockFunction; @@ -341,12 +266,13 @@ TEST_F(WebRtcTest, Connect_NullPeerConnection) { /*use_valid_peer_connection=*/false); WebRtc webrtc; + const std::string service_id("NearbySharing"); PeerId self_id("peer_id"); LocationHint location_hint; ASSERT_TRUE(webrtc.IsAvailable()); WebRtcSocketWrapper wrapper = - webrtc.Connect(PeerId("random_peer_id"), location_hint); + webrtc.Connect(service_id, PeerId("random_peer_id"), location_hint); EXPECT_FALSE(wrapper.IsValid()); } diff --git a/cpp/core/internal/p2p_cluster_pcp_handler.cc b/cpp/core/internal/p2p_cluster_pcp_handler.cc index f5071877..96a1ea89 100644 --- a/cpp/core/internal/p2p_cluster_pcp_handler.cc +++ b/cpp/core/internal/p2p_cluster_pcp_handler.cc @@ -1189,7 +1189,8 @@ BasePcpHandler::ConnectImplResult P2pClusterPcpHandler::WebRtcConnectImpl( ClientProxy* client, WebRtcEndpoint* webrtc_endpoint) { std::string empty_country_code; mediums::WebRtcSocketWrapper socket_wrapper = webrtc_medium_.Connect( - webrtc_endpoint->peer_id, Utils::BuildLocationHint(empty_country_code)); + webrtc_endpoint->service_id, webrtc_endpoint->peer_id, + Utils::BuildLocationHint(empty_country_code)); if (!socket_wrapper.IsValid()) { return BasePcpHandler::ConnectImplResult{.status = {Status::kError}}; } diff --git a/cpp/core/internal/payload_manager.cc b/cpp/core/internal/payload_manager.cc index 58a0c91c..2fbb40e9 100644 --- a/cpp/core/internal/payload_manager.cc +++ b/cpp/core/internal/payload_manager.cc @@ -1027,6 +1027,11 @@ void PayloadManager::PendingPayloads::StartTrackingPayload( Payload::Id payload_id, std::unique_ptr pending_payload) { MutexLock lock(&mutex_); + // If the |payload_id| is being re-used, always prefer the newer payload. + auto it = pending_payloads_.find(payload_id); + if (it != pending_payloads_.end()) { + pending_payloads_.erase(payload_id); + } auto pair = pending_payloads_.emplace(payload_id, std::move(pending_payload)); NEARBY_LOG(INFO, "StartTrackingPayload: payload_id=%" PRIX64 "; inserted=%d", payload_id, pair.second); diff --git a/cpp/core/internal/webrtc_bwu_handler.cc b/cpp/core/internal/webrtc_bwu_handler.cc index 0a2be7d9..85f53d14 100644 --- a/cpp/core/internal/webrtc_bwu_handler.cc +++ b/cpp/core/internal/webrtc_bwu_handler.cc @@ -108,7 +108,8 @@ WebrtcBwuHandler::CreateUpgradedEndpointChannel( "location hint %s", peer_id.GetId().c_str(), location_hint.DebugString().c_str()); - mediums::WebRtcSocketWrapper socket = webrtc_.Connect(peer_id, location_hint); + mediums::WebRtcSocketWrapper socket = + webrtc_.Connect(service_id, peer_id, location_hint); if (!socket.IsValid()) { NEARBY_LOG(ERROR, "WebRtcBwuHandler failed to connect to remote peer (%s) on "