Merge pull request #29 from edwinwutw/cl-350492846

Roll forward to Cl/350492846
2026-09-14 22:56:12 -04:00 · 2021-01-07 10:36:56 -08:00
parent 183aed7f6c 085907fea9
commit d9ff717369
11 changed files with 856 additions and 896 deletions
@@ -239,7 +239,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;
  }

@@ -77,7 +77,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_);
@@ -87,9 +87,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_);

@@ -100,138 +102,152 @@ class WebRtc {
    kAnswerer = 2,
  };

-  absl::flat_hash_map<Role, std::string> 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<ConnectionFlow> 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<WebRtcSignalingMessenger> 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<WebRtcSocketWrapper> ListenForWebRtcSocketFuture(
-      const Role& role, const std::string& connection_id,
-      Future<rtc::scoped_refptr<webrtc::DataChannelInterface>>
-          data_channel_future,
-      AcceptedConnectionCallback callback);
+    // Allows us to communicate with the Tachyon web server.
+    std::unique_ptr<WebRtcSignalingMessenger> signaling_messenger;

-  WebRtcSocketWrapper CreateWebRtcSocketWrapper(
-      const Role& role, const std::string& connection_id,
-      rtc::scoped_refptr<webrtc::DataChannelInterface> 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<WebRtcSocketWrapper> 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<std::unique_ptr<webrtc::IceCandidateInterface>>
+          ice_candidates) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  // Runs on |single_thread_executor_|.
+  std::unique_ptr<ConnectionFlow> CreateConnectionFlow(
+      const std::string& service_id, const PeerId& remote_peer_id)
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  // Runs on |single_thread_executor_|.
+  std::unique_ptr<ConnectionFlow> 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<webrtc::DataChannelInterface> 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<std::string, AcceptingConnectionsInfo>
+      accepting_connections_info_ ABSL_GUARDED_BY(mutex_);

-  // Use service_id as key for accepting connections.
-  absl::flat_hash_map<std::string, ConnectionInfo> accepting_map_
-      ABSL_GUARDED_BY(mutex_);
-  // Use remote peer_id as key for connecting connections.
-  absl::flat_hash_map<std::string, ConnectionInfo> 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<std::string, ConnectionRequestInfo>
+      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<std::string, std::unique_ptr<ConnectionFlow>>
+      connection_flows_ ABSL_GUARDED_BY(mutex_);
+
+  // A map of a remote PeerId -> Socket. Non-empty while we have active
+  // connections.
+  absl::flat_hash_map<std::string, WebRtcSocketWrapper> sockets_
      ABSL_GUARDED_BY(mutex_);
 };

@@ -105,6 +105,11 @@ ConnectionFlow::ConnectionFlow(

 ConnectionFlow::~ConnectionFlow() { Close(); }

+ConnectionFlow::State ConnectionFlow::GetState() {
+  MutexLock lock(&mutex_);
+  return state_;
+}
+
 SessionDescriptionWrapper ConnectionFlow::CreateOffer() {
  MutexLock lock(&mutex_);

@@ -233,11 +238,6 @@ bool ConnectionFlow::OnRemoteIceCandidatesReceived(
  return true;
 }

-Future<rtc::scoped_refptr<webrtc::DataChannelInterface>>
-ConnectionFlow::GetDataChannel() {
-  return data_channel_future_;
-}
-
 bool ConnectionFlow::Close() {
  MutexLock lock(&mutex_);
  return CloseLocked();
@@ -289,16 +289,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<webrtc::DataChannelInterface> 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(
@@ -308,15 +312,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<DataChannelObserverImpl>(
@@ -339,19 +342,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();
@@ -23,7 +23,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"
@@ -69,12 +68,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<ConnectionFlow> 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_);
@@ -100,8 +113,6 @@ class ConnectionFlow {
  bool OnRemoteIceCandidatesReceived(
      std::vector<std::unique_ptr<webrtc::IceCandidateInterface>>
          ice_candidates) ABSL_LOCKS_EXCLUDED(mutex_);
-  // Get a future for the data channel.
-  Future<rtc::scoped_refptr<webrtc::DataChannelInterface>> GetDataChannel();
  // Close the peer connection and data channel.
  bool Close() ABSL_LOCKS_EXCLUDED(mutex_);

@@ -117,17 +128,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);

@@ -141,7 +141,9 @@ class ConnectionFlow {

  bool SetRemoteSessionDescription(SessionDescriptionWrapper sdp);

-  void ProcessDataChannelConnected() ABSL_LOCKS_EXCLUDED(mutex_);
+  void ProcessDataChannelConnected(
+      rtc::scoped_refptr<webrtc::DataChannelInterface>)
+      ABSL_LOCKS_EXCLUDED(mutex_);

  void CloseAndNotifyLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
  bool CloseLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
@@ -155,8 +157,6 @@ class ConnectionFlow {

  std::unique_ptr<DataChannelObserverImpl> data_channel_observer_;

-  Future<rtc::scoped_refptr<webrtc::DataChannelInterface>> data_channel_future_;
-
  PeerConnectionObserverImpl peer_connection_observer_;
  rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection_;

@@ -57,6 +57,11 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) {

  Future<ByteArray> message_received_future;

+  Future<rtc::scoped_refptr<webrtc::DataChannelInterface>>
+      offerer_data_channel_future;
+  Future<rtc::scoped_refptr<webrtc::DataChannelInterface>>
+      answerer_data_channel_future;
+
  std::unique_ptr<ConnectionFlow> offerer, answerer;

  // Send Ice Candidates immediately when you retrieve them
@@ -70,7 +75,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<webrtc::DataChannelInterface> 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 =
@@ -82,7 +92,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<webrtc::DataChannelInterface> 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));
           }},
@@ -103,10 +118,10 @@ TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) {

  // Retrieve Data Channels
  ExceptionOr<rtc::scoped_refptr<webrtc::DataChannelInterface>>
-      offerer_channel = offerer->GetDataChannel().Get(absl::Seconds(1));
+      offerer_channel = offerer_data_channel_future.Get(absl::Seconds(1));
  EXPECT_TRUE(offerer_channel.ok());
  ExceptionOr<rtc::scoped_refptr<webrtc::DataChannelInterface>>
-      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
@@ -25,7 +25,10 @@ namespace mediums {

 // Callbacks from the data channel.
 struct DataChannelListener {
-  std::function<void()> data_channel_closed_cb = DefaultCallback<>();
+  // Called when the data channel is created.
+  std::function<void(rtc::scoped_refptr<webrtc::DataChannelInterface>)>
+      data_channel_created_cb =
+          DefaultCallback<rtc::scoped_refptr<webrtc::DataChannelInterface>>();

  // Called when a new message was received on the data channel.
  std::function<void(const ByteArray&)> data_channel_message_received_cb =
@@ -35,6 +38,9 @@ struct DataChannelListener {
  // changed.
  std::function<void()> data_channel_buffered_amount_changed_cb =
      DefaultCallback<>();
+
+  // Called when the data channel is closed.
+  std::function<void()> data_channel_closed_cb = DefaultCallback<>();
 };

 }  // namespace mediums
@@ -27,7 +27,7 @@ namespace connections {
 namespace mediums {

 namespace {
-const int kTwoMBSize = 2000000;
+
 class WebRtcTest : public ::testing::Test {
 protected:
  WebRtcTest() {
@@ -75,7 +75,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(
@@ -99,7 +100,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(
@@ -149,14 +150,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<bool> 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();

@@ -190,7 +192,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<bool> devices_connected = connected.Get();
@@ -218,7 +220,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<bool> devices_connected = connected.Get();
@@ -252,7 +254,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<bool> devices_connected = connected.Get();
@@ -269,83 +271,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<bool> 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<bool> devices_connected1 = connected1.Get();
-  ASSERT_TRUE(devices_connected1.ok());
-  EXPECT_TRUE(devices_connected1.result());
-
-  ExceptionOr<bool> 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<ByteArray> received_msg1 =
-        receiver_socket1.GetInputStream().Read(kTwoMBSize / 10);
-    ASSERT_TRUE(received_msg1.ok());
-    ExceptionOr<ByteArray> 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<void(WebRtcSocketWrapper socket)>;
-  testing::StrictMock<MockAcceptedCallback> 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<void(WebRtcSocketWrapper socket)>;
@@ -355,12 +280,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());
 }

@@ -1203,7 +1203,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}};
  }
@@ -1041,6 +1041,11 @@ void PayloadManager::PendingPayloads::StartTrackingPayload(
    Payload::Id payload_id, std::unique_ptr<PendingPayload> 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);
@@ -122,7 +122,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 "