Introduce dynamic endpoints

Bug: webrtc:10138
Change-Id: I7f6922adb93680cada6bea014539fc3089735834
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/128480
Reviewed-by: Sebastian Jansson <srte@webrtc.org>
Reviewed-by: Karl Wiberg <kwiberg@webrtc.org>
Commit-Queue: Artem Titov <titovartem@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#27336}

api/test/network_emulation_manager.h

@@ -46,6 +46,19 @@ struct EmulatedEndpointConfig {
   // If specified will be used as IP address for endpoint node. Must be unique
   // among all created nodes.
   absl::optional<rtc::IPAddress> ip;
+  // Should endpoint be enabled or not, when it will be created.
+  // Enabled endpoints will be available for webrtc to send packets.
+  bool start_as_enabled = true;
+};
+
+// Provide interface to obtain all required objects to inject network emulation
+// layer into PeerConnection.
+class EmulatedNetworkManagerInterface {
+ public:
+  virtual ~EmulatedNetworkManagerInterface() = default;
+
+  virtual rtc::Thread* network_thread() = 0;
+  virtual rtc::NetworkManager* network_manager() = 0;
 };
 
 // Provides an API for creating and configuring emulated network layer.
@@ -63,6 +76,12 @@ class NetworkEmulationManager {
   // Creates an emulated endpoint, which represents single network interface on
   // the peer's device.
   virtual EmulatedEndpoint* CreateEndpoint(EmulatedEndpointConfig config) = 0;
+  // Enable emulated endpoint to make it available for webrtc.
+  // Caller mustn't enable currently enabled endpoint.
+  virtual void EnableEndpoint(EmulatedEndpoint* endpoint) = 0;
+  // Disable emulated endpoint to make it unavailable for webrtc.
+  // Caller mustn't disable currently disabled endpoint.
+  virtual void DisableEndpoint(EmulatedEndpoint* endpoint) = 0;
 
   // Creates a route between endpoints going through specified network nodes.
   // This route is single direction only and describe how traffic that was
@@ -88,16 +107,13 @@ class NetworkEmulationManager {
   // removed earlier.
   virtual void ClearRoute(EmulatedRoute* route) = 0;
 
-  // Creates rtc::Thread that should be used as network thread for peer
-  // connection. Created thread contains special rtc::SocketServer inside it
-  // to enable correct integration between peer connection and emulated network
-  // layer.
-  virtual rtc::Thread* CreateNetworkThread(
-      const std::vector<EmulatedEndpoint*>& endpoints) = 0;
-  // Creates rtc::NetworkManager that should be used inside
-  // cricket::PortAllocator for peer connection to provide correct list of
-  // network interfaces, that exists in emulated network later.
-  virtual rtc::NetworkManager* CreateNetworkManager(
+  // Creates EmulatedNetworkManagerInterface which can be used then to inject
+  // network emulation layer into PeerConnection. |endpoints| - are available
+  // network interfaces for PeerConnection. If endpoint is enabled, it will be
+  // immediately available for PeerConnection, otherwise user will be able to
+  // enable endpoint later to make it available for PeerConnection.
+  virtual EmulatedNetworkManagerInterface*
+  CreateEmulatedNetworkManagerInterface(
       const std::vector<EmulatedEndpoint*>& endpoints) = 0;
 };
 

test/pc/e2e/peer_connection_e2e_smoke_test.cc

@@ -75,25 +75,27 @@ TEST(PeerConnectionE2EQualityTestSmokeTest, RunWithEmulatedNetwork) {
 
   // Setup components. We need to provide rtc::NetworkManager compatible with
   // emulated network layer.
-  fixture->AddPeer(
-      network_emulation_manager->CreateNetworkThread({alice_endpoint}),
-      network_emulation_manager->CreateNetworkManager({alice_endpoint}),
-      [](PeerConfigurer* alice) {
-        VideoConfig alice_video_config(640, 360, 30);
-        alice_video_config.stream_label = "alice-video";
-        alice->AddVideoConfig(std::move(alice_video_config));
-        alice->SetAudioConfig(AudioConfig());
-      });
+  EmulatedNetworkManagerInterface* alice_network =
+      network_emulation_manager->CreateEmulatedNetworkManagerInterface(
+          {alice_endpoint});
+  fixture->AddPeer(alice_network->network_thread(),
+                   alice_network->network_manager(), [](PeerConfigurer* alice) {
+                     VideoConfig alice_video_config(640, 360, 30);
+                     alice_video_config.stream_label = "alice-video";
+                     alice->AddVideoConfig(std::move(alice_video_config));
+                     alice->SetAudioConfig(AudioConfig());
+                   });
 
-  fixture->AddPeer(
-      network_emulation_manager->CreateNetworkThread({bob_endpoint}),
-      network_emulation_manager->CreateNetworkManager({bob_endpoint}),
-      [](PeerConfigurer* bob) {
-        VideoConfig bob_video_config(640, 360, 30);
-        bob_video_config.stream_label = "bob-video";
-        bob->AddVideoConfig(std::move(bob_video_config));
-        bob->SetAudioConfig(AudioConfig());
-      });
+  EmulatedNetworkManagerInterface* bob_network =
+      network_emulation_manager->CreateEmulatedNetworkManagerInterface(
+          {bob_endpoint});
+  fixture->AddPeer(bob_network->network_thread(),
+                   bob_network->network_manager(), [](PeerConfigurer* bob) {
+                     VideoConfig bob_video_config(640, 360, 30);
+                     bob_video_config.stream_label = "bob-video";
+                     bob->AddVideoConfig(std::move(bob_video_config));
+                     bob->SetAudioConfig(AudioConfig());
+                   });
 
   fixture->Run(RunParams{TimeDelta::seconds(5)});
 

test/scenario/network/BUILD.gn

@@ -13,6 +13,8 @@ rtc_source_set("emulated_network") {
   sources = [
     "cross_traffic.cc",
     "cross_traffic.h",
+    "emulated_network_manager.cc",
+    "emulated_network_manager.h",
     "fake_network_socket.cc",
     "fake_network_socket.h",
     "fake_network_socket_server.cc",

test/scenario/network/emulated_network_manager.cc

@@ -0,0 +1,108 @@
+/*
+ *  Copyright (c) 2019 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+
+#include "test/scenario/network/emulated_network_manager.h"
+
+#include <memory>
+#include <utility>
+
+#include "absl/memory/memory.h"
+
+namespace webrtc {
+namespace test {
+
+EmulatedNetworkManager::EmulatedNetworkManager(
+    Clock* clock,
+    EndpointsContainer* endpoints_controller)
+    : endpoints_controller_(endpoints_controller),
+      socket_server_(clock, endpoints_controller),
+      network_thread_(&socket_server_),
+      sent_first_update_(false),
+      start_count_(0) {
+  network_thread_.SetName("net_thread", nullptr);
+  network_thread_.Start();
+}
+
+void EmulatedNetworkManager::EnableEndpoint(EmulatedEndpoint* endpoint) {
+  RTC_CHECK(endpoints_controller_->HasEndpoint(endpoint))
+      << "No such interface: " << endpoint->GetPeerLocalAddress().ToString();
+  network_thread_.PostTask(RTC_FROM_HERE, [this, endpoint]() {
+    endpoint->Enable();
+    UpdateNetworksOnce();
+  });
+}
+
+void EmulatedNetworkManager::DisableEndpoint(EmulatedEndpoint* endpoint) {
+  RTC_CHECK(endpoints_controller_->HasEndpoint(endpoint))
+      << "No such interface: " << endpoint->GetPeerLocalAddress().ToString();
+  network_thread_.PostTask(RTC_FROM_HERE, [this, endpoint]() {
+    endpoint->Disable();
+    UpdateNetworksOnce();
+  });
+}
+
+// Network manager interface. All these methods are supposed to be called from
+// the same thread.
+void EmulatedNetworkManager::StartUpdating() {
+  RTC_DCHECK_RUN_ON(&network_thread_);
+
+  if (start_count_) {
+    // If network interfaces are already discovered and signal is sent,
+    // we should trigger network signal immediately for the new clients
+    // to start allocating ports.
+    if (sent_first_update_)
+      network_thread_.PostTask(RTC_FROM_HERE,
+                               [this]() { MaybeSignalNetworksChanged(); });
+  } else {
+    network_thread_.PostTask(RTC_FROM_HERE, [this]() { UpdateNetworksOnce(); });
+  }
+  ++start_count_;
+}
+
+void EmulatedNetworkManager::StopUpdating() {
+  RTC_DCHECK_RUN_ON(&network_thread_);
+  if (!start_count_)
+    return;
+
+  --start_count_;
+  if (!start_count_) {
+    sent_first_update_ = false;
+  }
+}
+
+void EmulatedNetworkManager::UpdateNetworksOnce() {
+  RTC_DCHECK_RUN_ON(&network_thread_);
+
+  std::vector<rtc::Network*> networks;
+  for (std::unique_ptr<rtc::Network>& net :
+       endpoints_controller_->GetEnabledNetworks()) {
+    net->set_default_local_address_provider(this);
+    networks.push_back(net.release());
+  }
+
+  bool changed;
+  MergeNetworkList(networks, &changed);
+  if (changed || !sent_first_update_) {
+    MaybeSignalNetworksChanged();
+    sent_first_update_ = true;
+  }
+}
+
+void EmulatedNetworkManager::MaybeSignalNetworksChanged() {
+  RTC_DCHECK_RUN_ON(&network_thread_);
+  // If manager is stopped we don't need to signal anything.
+  if (start_count_ == 0) {
+    return;
+  }
+  SignalNetworksChanged();
+}
+
+}  // namespace test
+}  // namespace webrtc

test/scenario/network/emulated_network_manager.h

@@ -0,0 +1,65 @@
+/*
+ *  Copyright (c) 2019 The WebRTC project authors. All Rights Reserved.
+ *
+ *  Use of this source code is governed by a BSD-style license
+ *  that can be found in the LICENSE file in the root of the source
+ *  tree. An additional intellectual property rights grant can be found
+ *  in the file PATENTS.  All contributing project authors may
+ *  be found in the AUTHORS file in the root of the source tree.
+ */
+
+#ifndef TEST_SCENARIO_NETWORK_EMULATED_NETWORK_MANAGER_H_
+#define TEST_SCENARIO_NETWORK_EMULATED_NETWORK_MANAGER_H_
+
+#include <memory>
+#include <vector>
+
+#include "api/test/network_emulation_manager.h"
+#include "rtc_base/critical_section.h"
+#include "rtc_base/ip_address.h"
+#include "rtc_base/network.h"
+#include "rtc_base/socket_server.h"
+#include "rtc_base/thread.h"
+#include "rtc_base/thread_checker.h"
+#include "test/scenario/network/fake_network_socket_server.h"
+#include "test/scenario/network/network_emulation.h"
+
+namespace webrtc {
+namespace test {
+
+// Framework assumes that rtc::NetworkManager is called from network thread.
+class EmulatedNetworkManager : public rtc::NetworkManagerBase,
+                               public sigslot::has_slots<>,
+                               public EmulatedNetworkManagerInterface {
+ public:
+  EmulatedNetworkManager(Clock* clock,
+                         EndpointsContainer* endpoints_controller);
+
+  void EnableEndpoint(EmulatedEndpoint* endpoint);
+  void DisableEndpoint(EmulatedEndpoint* endpoint);
+
+  // NetworkManager interface. All these methods are supposed to be called from
+  // the same thread.
+  void StartUpdating() override;
+  void StopUpdating() override;
+
+  // EmulatedNetworkManagerInterface API
+  rtc::Thread* network_thread() override { return &network_thread_; }
+  rtc::NetworkManager* network_manager() override { return this; }
+
+ private:
+  void UpdateNetworksOnce();
+  void MaybeSignalNetworksChanged();
+
+  EndpointsContainer* const endpoints_controller_;
+  FakeNetworkSocketServer socket_server_;
+  rtc::Thread network_thread_;
+
+  bool sent_first_update_ RTC_GUARDED_BY(network_thread_);
+  int start_count_ RTC_GUARDED_BY(network_thread_);
+};
+
+}  // namespace test
+}  // namespace webrtc
+
+#endif  // TEST_SCENARIO_NETWORK_EMULATED_NETWORK_MANAGER_H_

test/scenario/network/fake_network_socket_server.cc

@@ -17,9 +17,9 @@ namespace test {
 
 FakeNetworkSocketServer::FakeNetworkSocketServer(
     Clock* clock,
-    std::vector<EmulatedEndpoint*> endpoints)
+    EndpointsContainer* endpoints_container)
     : clock_(clock),
-      endpoints_(std::move(endpoints)),
+      endpoints_container_(endpoints_container),
       wakeup_(/*manual_reset=*/false, /*initially_signaled=*/false) {}
 FakeNetworkSocketServer::~FakeNetworkSocketServer() = default;
 
@@ -29,13 +29,7 @@ void FakeNetworkSocketServer::OnMessageQueueDestroyed() {
 
 EmulatedEndpoint* FakeNetworkSocketServer::GetEndpointNode(
     const rtc::IPAddress& ip) {
-  for (auto* endpoint : endpoints_) {
-    rtc::IPAddress peerLocalAddress = endpoint->GetPeerLocalAddress();
-    if (peerLocalAddress == ip) {
-      return endpoint;
-    }
-  }
-  RTC_CHECK(false) << "No network found for address" << ip.ToString();
+  return endpoints_container_->LookupByLocalAddress(ip);
 }
 
 void FakeNetworkSocketServer::Unregister(SocketIoProcessor* io_processor) {

test/scenario/network/fake_network_socket_server.h

@@ -35,7 +35,7 @@ class FakeNetworkSocketServer : public rtc::SocketServer,
                                 public SocketManager {
  public:
   FakeNetworkSocketServer(Clock* clock,
-                          std::vector<EmulatedEndpoint*> endpoints);
+                          EndpointsContainer* endpoints_controller);
   ~FakeNetworkSocketServer() override;
 
   EmulatedEndpoint* GetEndpointNode(const rtc::IPAddress& ip) override;
@@ -57,7 +57,7 @@ class FakeNetworkSocketServer : public rtc::SocketServer,
   Timestamp Now() const;
 
   Clock* const clock_;
-  const std::vector<EmulatedEndpoint*> endpoints_;
+  const EndpointsContainer* endpoints_container_;
   rtc::Event wakeup_;
   rtc::MessageQueue* msg_queue_;
 

test/scenario/network/network_emulation.cc

@@ -128,13 +128,34 @@ void EmulatedNetworkNode::RemoveReceiver(uint64_t dest_endpoint_id) {
   routing_.erase(dest_endpoint_id);
 }
 
-EmulatedEndpoint::EmulatedEndpoint(uint64_t id, rtc::IPAddress ip, Clock* clock)
+EmulatedEndpoint::EmulatedEndpoint(uint64_t id,
+                                   rtc::IPAddress ip,
+                                   bool is_enabled,
+                                   Clock* clock)
     : id_(id),
       peer_local_addr_(ip),
+      is_enabled_(is_enabled),
       send_node_(nullptr),
       clock_(clock),
       next_port_(kFirstEphemeralPort),
-      connected_endpoint_id_(absl::nullopt) {}
+      connected_endpoint_id_(absl::nullopt) {
+  constexpr int kIPv4NetworkPrefixLength = 24;
+  constexpr int kIPv6NetworkPrefixLength = 64;
+
+  int prefix_length = 0;
+  if (ip.family() == AF_INET) {
+    prefix_length = kIPv4NetworkPrefixLength;
+  } else if (ip.family() == AF_INET6) {
+    prefix_length = kIPv6NetworkPrefixLength;
+  }
+  rtc::IPAddress prefix = TruncateIP(ip, prefix_length);
+  network_ = absl::make_unique<rtc::Network>(
+      ip.ToString(), "Endpoint id=" + std::to_string(id_), prefix,
+      prefix_length, rtc::AdapterType::ADAPTER_TYPE_UNKNOWN);
+  network_->AddIP(ip);
+
+  enabled_state_checker_.DetachFromThread();
+}
 EmulatedEndpoint::~EmulatedEndpoint() = default;
 
 uint64_t EmulatedEndpoint::GetId() const {
@@ -226,6 +247,23 @@ void EmulatedEndpoint::OnPacketReceived(EmulatedIpPacket packet) {
   it->second->OnPacketReceived(std::move(packet));
 }
 
+void EmulatedEndpoint::Enable() {
+  RTC_DCHECK_RUN_ON(&enabled_state_checker_);
+  RTC_CHECK(!is_enabled_);
+  is_enabled_ = true;
+}
+
+void EmulatedEndpoint::Disable() {
+  RTC_DCHECK_RUN_ON(&enabled_state_checker_);
+  RTC_CHECK(is_enabled_);
+  is_enabled_ = false;
+}
+
+bool EmulatedEndpoint::Enabled() const {
+  RTC_DCHECK_RUN_ON(&enabled_state_checker_);
+  return is_enabled_;
+}
+
 EmulatedNetworkNode* EmulatedEndpoint::GetSendNode() const {
   return send_node_;
 }
@@ -234,4 +272,40 @@ void EmulatedEndpoint::SetConnectedEndpointId(uint64_t endpoint_id) {
   connected_endpoint_id_ = endpoint_id;
 }
 
+EndpointsContainer::EndpointsContainer(
+    const std::vector<EmulatedEndpoint*>& endpoints)
+    : endpoints_(endpoints) {}
+
+EmulatedEndpoint* EndpointsContainer::LookupByLocalAddress(
+    const rtc::IPAddress& local_ip) const {
+  for (auto* endpoint : endpoints_) {
+    rtc::IPAddress peerLocalAddress = endpoint->GetPeerLocalAddress();
+    if (peerLocalAddress == local_ip) {
+      return endpoint;
+    }
+  }
+  RTC_CHECK(false) << "No network found for address" << local_ip.ToString();
+}
+
+bool EndpointsContainer::HasEndpoint(EmulatedEndpoint* endpoint) const {
+  for (auto* e : endpoints_) {
+    if (e->GetId() == endpoint->GetId()) {
+      return true;
+    }
+  }
+  return false;
+}
+
+std::vector<std::unique_ptr<rtc::Network>>
+EndpointsContainer::GetEnabledNetworks() const {
+  std::vector<std::unique_ptr<rtc::Network>> networks;
+  for (auto* endpoint : endpoints_) {
+    if (endpoint->Enabled()) {
+      networks.emplace_back(
+          absl::make_unique<rtc::Network>(endpoint->network()));
+    }
+  }
+  return networks;
+}
+
 }  // namespace webrtc

test/scenario/network/network_emulation.h

@@ -24,8 +24,10 @@
 #include "rtc_base/async_socket.h"
 #include "rtc_base/copy_on_write_buffer.h"
 #include "rtc_base/critical_section.h"
+#include "rtc_base/network.h"
 #include "rtc_base/socket_address.h"
 #include "rtc_base/thread.h"
+#include "rtc_base/thread_checker.h"
 #include "system_wrappers/include/clock.h"
 
 namespace webrtc {
@@ -120,7 +122,10 @@ class EmulatedNetworkNode : public EmulatedNetworkReceiverInterface {
 // from other EmulatedNetworkNodes.
 class EmulatedEndpoint : public EmulatedNetworkReceiverInterface {
  public:
-  EmulatedEndpoint(uint64_t id, rtc::IPAddress, Clock* clock);
+  EmulatedEndpoint(uint64_t id,
+                   rtc::IPAddress ip,
+                   bool is_enabled,
+                   Clock* clock);
   ~EmulatedEndpoint() override;
 
   uint64_t GetId() const;
@@ -155,6 +160,12 @@ class EmulatedEndpoint : public EmulatedNetworkReceiverInterface {
   // Will be called to deliver packet into endpoint from network node.
   void OnPacketReceived(EmulatedIpPacket packet) override;
 
+  void Enable();
+  void Disable();
+  bool Enabled() const;
+
+  const rtc::Network& network() const { return *network_.get(); }
+
  protected:
   friend class test::NetworkEmulationManagerImpl;
 
@@ -166,12 +177,15 @@ class EmulatedEndpoint : public EmulatedNetworkReceiverInterface {
   uint16_t NextPort() RTC_EXCLUSIVE_LOCKS_REQUIRED(receiver_lock_);
 
   rtc::CriticalSection receiver_lock_;
+  rtc::ThreadChecker enabled_state_checker_;
 
   uint64_t id_;
   // Peer's local IP address for this endpoint network interface.
   const rtc::IPAddress peer_local_addr_;
+  bool is_enabled_ RTC_GUARDED_BY(enabled_state_checker_);
   EmulatedNetworkNode* send_node_;
   Clock* const clock_;
+  std::unique_ptr<rtc::Network> network_;
 
   uint16_t next_port_ RTC_GUARDED_BY(receiver_lock_);
   std::map<uint16_t, EmulatedNetworkReceiverInterface*> port_to_receiver_
@@ -193,6 +207,20 @@ class EmulatedRoute {
   bool active;
 };
 
+class EndpointsContainer {
+ public:
+  EndpointsContainer(const std::vector<EmulatedEndpoint*>& endpoints);
+
+  EmulatedEndpoint* LookupByLocalAddress(const rtc::IPAddress& local_ip) const;
+  bool HasEndpoint(EmulatedEndpoint* endpoint) const;
+  // Returns list of networks for enabled endpoints. Caller takes ownership of
+  // returned rtc::Network objects.
+  std::vector<std::unique_ptr<rtc::Network>> GetEnabledNetworks() const;
+
+ private:
+  const std::vector<EmulatedEndpoint*> endpoints_;
+};
+
 }  // namespace webrtc
 
 #endif  // TEST_SCENARIO_NETWORK_NETWORK_EMULATION_H_

test/scenario/network/network_emulation_manager.cc

@@ -80,12 +80,27 @@ EmulatedEndpoint* NetworkEmulationManagerImpl::CreateEndpoint(
 
   bool res = used_ip_addresses_.insert(*ip).second;
   RTC_CHECK(res) << "IP=" << ip->ToString() << " already in use";
-  auto node = absl::make_unique<EmulatedEndpoint>(next_node_id_++, *ip, clock_);
+  auto node = absl::make_unique<EmulatedEndpoint>(
+      next_node_id_++, *ip, config.start_as_enabled, clock_);
   EmulatedEndpoint* out = node.get();
   endpoints_.push_back(std::move(node));
   return out;
 }
 
+void NetworkEmulationManagerImpl::EnableEndpoint(EmulatedEndpoint* endpoint) {
+  EmulatedNetworkManager* network_manager =
+      endpoint_to_network_manager_[endpoint];
+  RTC_CHECK(network_manager);
+  network_manager->EnableEndpoint(endpoint);
+}
+
+void NetworkEmulationManagerImpl::DisableEndpoint(EmulatedEndpoint* endpoint) {
+  EmulatedNetworkManager* network_manager =
+      endpoint_to_network_manager_[endpoint];
+  RTC_CHECK(network_manager);
+  network_manager->DisableEndpoint(endpoint);
+}
+
 EmulatedRoute* NetworkEmulationManagerImpl::CreateRoute(
     EmulatedEndpoint* from,
     const std::vector<EmulatedNetworkNode*>& via_nodes,
@@ -182,37 +197,26 @@ NetworkEmulationManagerImpl::CreatePulsedPeaksCrossTraffic(
   return out;
 }
 
-rtc::Thread* NetworkEmulationManagerImpl::CreateNetworkThread(
+EmulatedNetworkManagerInterface*
+NetworkEmulationManagerImpl::CreateEmulatedNetworkManagerInterface(
     const std::vector<EmulatedEndpoint*>& endpoints) {
-  FakeNetworkSocketServer* socket_server = CreateSocketServer(endpoints);
-  std::unique_ptr<rtc::Thread> network_thread =
-      absl::make_unique<rtc::Thread>(socket_server);
-  network_thread->SetName("network_thread" + std::to_string(threads_.size()),
-                          nullptr);
-  network_thread->Start();
-  rtc::Thread* out = network_thread.get();
-  threads_.push_back(std::move(network_thread));
-  return out;
-}
-
-rtc::NetworkManager* NetworkEmulationManagerImpl::CreateNetworkManager(
-    const std::vector<EmulatedEndpoint*>& endpoints) {
-  auto network_manager = absl::make_unique<rtc::FakeNetworkManager>();
+  auto endpoints_controller = absl::make_unique<EndpointsContainer>(endpoints);
+  auto network_manager = absl::make_unique<EmulatedNetworkManager>(
+      clock_, endpoints_controller.get());
   for (auto* endpoint : endpoints) {
-    network_manager->AddInterface(
-        rtc::SocketAddress(endpoint->GetPeerLocalAddress(), /*port=*/0));
+    // Associate endpoint with network manager.
+    bool insertion_result =
+        endpoint_to_network_manager_.insert({endpoint, network_manager.get()})
+            .second;
+    RTC_CHECK(insertion_result)
+        << "Endpoint ip=" << endpoint->GetPeerLocalAddress().ToString()
+        << " is already used for another network";
   }
-  rtc::NetworkManager* out = network_manager.get();
-  managers_.push_back(std::move(network_manager));
-  return out;
-}
 
-FakeNetworkSocketServer* NetworkEmulationManagerImpl::CreateSocketServer(
-    const std::vector<EmulatedEndpoint*>& endpoints) {
-  auto socket_server =
-      absl::make_unique<FakeNetworkSocketServer>(clock_, endpoints);
-  FakeNetworkSocketServer* out = socket_server.get();
-  socket_servers_.push_back(std::move(socket_server));
+  EmulatedNetworkManagerInterface* out = network_manager.get();
+
+  endpoints_controllers_.push_back(std::move(endpoints_controller));
+  network_managers_.push_back(std::move(network_manager));
   return out;
 }
 

test/scenario/network/network_emulation_manager.h

@@ -26,6 +26,7 @@
 #include "rtc_base/thread.h"
 #include "system_wrappers/include/clock.h"
 #include "test/scenario/network/cross_traffic.h"
+#include "test/scenario/network/emulated_network_manager.h"
 #include "test/scenario/network/fake_network_socket_server.h"
 #include "test/scenario/network/network_emulation.h"
 #include "test/scenario/network/traffic_route.h"
@@ -42,6 +43,8 @@ class NetworkEmulationManagerImpl : public NetworkEmulationManager {
       std::unique_ptr<NetworkBehaviorInterface> network_behavior) override;
 
   EmulatedEndpoint* CreateEndpoint(EmulatedEndpointConfig config) override;
+  void EnableEndpoint(EmulatedEndpoint* endpoint) override;
+  void DisableEndpoint(EmulatedEndpoint* endpoint) override;
 
   EmulatedRoute* CreateRoute(EmulatedEndpoint* from,
                              const std::vector<EmulatedNetworkNode*>& via_nodes,
@@ -57,9 +60,7 @@ class NetworkEmulationManagerImpl : public NetworkEmulationManager {
       TrafficRoute* traffic_route,
       PulsedPeaksConfig config);
 
-  rtc::Thread* CreateNetworkThread(
-      const std::vector<EmulatedEndpoint*>& endpoints) override;
-  rtc::NetworkManager* CreateNetworkManager(
+  EmulatedNetworkManagerInterface* CreateEmulatedNetworkManagerInterface(
       const std::vector<EmulatedEndpoint*>& endpoints) override;
 
  private:
@@ -84,9 +85,11 @@ class NetworkEmulationManagerImpl : public NetworkEmulationManager {
   std::vector<std::unique_ptr<TrafficRoute>> traffic_routes_;
   std::vector<std::unique_ptr<RandomWalkCrossTraffic>> random_cross_traffics_;
   std::vector<std::unique_ptr<PulsedPeaksCrossTraffic>> pulsed_cross_traffics_;
-  std::vector<std::unique_ptr<FakeNetworkSocketServer>> socket_servers_;
-  std::vector<std::unique_ptr<rtc::Thread>> threads_;
-  std::vector<std::unique_ptr<rtc::NetworkManager>> managers_;
+  std::vector<std::unique_ptr<EndpointsContainer>> endpoints_controllers_;
+  std::vector<std::unique_ptr<EmulatedNetworkManager>> network_managers_;
+
+  std::map<EmulatedEndpoint*, EmulatedNetworkManager*>
+      endpoint_to_network_manager_;
 
   // Must be the last field, so it will be deleted first, because tasks
   // in the TaskQueue can access other fields of the instance of this class.

test/scenario/network/network_emulation_pc_unittest.cc

@@ -113,36 +113,32 @@ TEST(NetworkEmulationManagerPCTest, Run) {
   emulation.CreateRoute(alice_endpoint, {alice_node}, bob_endpoint);
   emulation.CreateRoute(bob_endpoint, {bob_node}, alice_endpoint);
 
-  rtc::Thread* alice_network_thread =
-      emulation.CreateNetworkThread({alice_endpoint});
-  rtc::Thread* bob_network_thread =
-      emulation.CreateNetworkThread({bob_endpoint});
+  EmulatedNetworkManagerInterface* alice_network =
+      emulation.CreateEmulatedNetworkManagerInterface({alice_endpoint});
+  EmulatedNetworkManagerInterface* bob_network =
+      emulation.CreateEmulatedNetworkManagerInterface({bob_endpoint});
 
   // Setup peer connections.
   rtc::scoped_refptr<PeerConnectionFactoryInterface> alice_pcf;
   rtc::scoped_refptr<PeerConnectionInterface> alice_pc;
   std::unique_ptr<MockPeerConnectionObserver> alice_observer =
       absl::make_unique<MockPeerConnectionObserver>();
-  rtc::NetworkManager* alice_network_manager =
-      emulation.CreateNetworkManager({alice_endpoint});
 
   rtc::scoped_refptr<PeerConnectionFactoryInterface> bob_pcf;
   rtc::scoped_refptr<PeerConnectionInterface> bob_pc;
   std::unique_ptr<MockPeerConnectionObserver> bob_observer =
       absl::make_unique<MockPeerConnectionObserver>();
-  rtc::NetworkManager* bob_network_manager =
-      emulation.CreateNetworkManager({bob_endpoint});
 
   signaling_thread->Invoke<void>(RTC_FROM_HERE, [&]() {
     alice_pcf = CreatePeerConnectionFactory(signaling_thread.get(),
-                                            alice_network_thread);
+                                            alice_network->network_thread());
     alice_pc = CreatePeerConnection(alice_pcf, alice_observer.get(),
-                                    alice_network_manager);
+                                    alice_network->network_manager());
 
-    bob_pcf =
-        CreatePeerConnectionFactory(signaling_thread.get(), bob_network_thread);
-    bob_pc =
-        CreatePeerConnection(bob_pcf, bob_observer.get(), bob_network_manager);
+    bob_pcf = CreatePeerConnectionFactory(signaling_thread.get(),
+                                          bob_network->network_thread());
+    bob_pc = CreatePeerConnection(bob_pcf, bob_observer.get(),
+                                  bob_network->network_manager());
   });
 
   std::unique_ptr<PeerConnectionWrapper> alice =

test/scenario/network/network_emulation_unittest.cc

@@ -97,12 +97,16 @@ TEST(NetworkEmulationManagerTest, Run) {
   network_manager.CreateRoute(alice_endpoint, {alice_node}, bob_endpoint);
   network_manager.CreateRoute(bob_endpoint, {bob_node}, alice_endpoint);
 
-  auto* nt1 = network_manager.CreateNetworkThread({alice_endpoint});
-  auto* nt2 = network_manager.CreateNetworkThread({bob_endpoint});
+  EmulatedNetworkManagerInterface* nt1 =
+      network_manager.CreateEmulatedNetworkManagerInterface({alice_endpoint});
+  EmulatedNetworkManagerInterface* nt2 =
+      network_manager.CreateEmulatedNetworkManagerInterface({bob_endpoint});
 
   for (uint64_t j = 0; j < 2; j++) {
-    auto* s1 = nt1->socketserver()->CreateAsyncSocket(AF_INET, SOCK_DGRAM);
-    auto* s2 = nt2->socketserver()->CreateAsyncSocket(AF_INET, SOCK_DGRAM);
+    auto* s1 = nt1->network_thread()->socketserver()->CreateAsyncSocket(
+        AF_INET, SOCK_DGRAM);
+    auto* s2 = nt2->network_thread()->socketserver()->CreateAsyncSocket(
+        AF_INET, SOCK_DGRAM);
 
     SocketReader r1(s1);
     SocketReader r2(s2);
@@ -113,8 +117,8 @@ TEST(NetworkEmulationManagerTest, Run) {
     s1->Bind(a1);
     s2->Bind(a2);
 
-    s1->Connect(s1->GetLocalAddress());
-    s2->Connect(s2->GetLocalAddress());
+    s1->Connect(s2->GetLocalAddress());
+    s2->Connect(s1->GetLocalAddress());
 
     rtc::CopyOnWriteBuffer data("Hello");
     for (uint64_t i = 0; i < 1000; i++) {