Adds platform factories for network and task threads

Adds wrappers around NetworkReader and TaskRunnerImpl to simplify
the creation and destruction flows as used in the demo code

Change-Id: I350bd06af3d78963e516d3ee674426682db6b343
Reviewed-on: https://chromium-review.googlesource.com/c/openscreen/+/1773409
Commit-Queue: Ryan Keane <rwkeane@google.com>
Reviewed-by: mark a. foltz <mfoltz@chromium.org>
Reviewed-by: Max Yakimakha <yakimakha@chromium.org>

2 files changed, 24 insertions, 43 deletions

diff --git a/osp/demo/demo.cc b/osp/demo/demo.cc
index 2e3aefa0..3966bc4b 100644
--- a/osp/demo/demo.cc
+++ b/osp/demo/demo.cc
@@ -31,7 +31,9 @@
 #include "platform/api/time.h"
 #include "platform/api/trace_logging.h"
 #include "platform/impl/network_reader.h"
+#include "platform/impl/network_reader_thread.h"
 #include "platform/impl/task_runner.h"
+#include "platform/impl/task_runner_thread.h"
 #include "platform/impl/text_trace_logging_platform.h"
 #include "third_party/tinycbor/src/src/cbor.h"
 
@@ -422,25 +424,21 @@ void RunControllerPollLoop(presentation::Controller* controller) {
 void ListenerDemo() {
   SignalThings();
 
-  // TODO(rwkeane): Wrap these start/end calls in a class.
-  platform::TaskRunnerImpl task_runner(platform::Clock::now);
-  platform::NetworkReader network_reader;
-  std::thread task_runner_thread(
-      [&task_runner]() { task_runner.RunUntilStopped(); });
-  std::thread network_reader_thread(
-      [&network_reader]() { network_reader.RunUntilStopped(); });
-  platform::UdpSocket::SetLifetimeObserver(&network_reader);
+  platform::TaskRunnerThread task_runner_thread(platform::Clock::now);
+  platform::NetworkReaderThread network_reader_thread;
+  platform::UdpSocket::SetLifetimeObserver(
+      network_reader_thread.network_reader());
 
   ListenerObserver listener_observer;
   MdnsServiceListenerConfig listener_config;
   auto mdns_listener = MdnsServiceListenerFactory::Create(
-      listener_config, &listener_observer, &task_runner);
+      listener_config, &listener_observer, task_runner_thread.task_runner());
 
   MessageDemuxer demuxer(platform::Clock::now,
                          MessageDemuxer::kDefaultBufferLimit);
   ConnectionClientObserver client_observer;
   auto connection_client = ProtocolConnectionClientFactory::Create(
-      &demuxer, &client_observer, &task_runner);
+      &demuxer, &client_observer, task_runner_thread.task_runner());
 
   auto* network_service = NetworkServiceManager::Create(
       std::move(mdns_listener), nullptr, std::move(connection_client), nullptr);
@@ -458,11 +456,6 @@ void ListenerDemo() {
   controller.reset();
 
   NetworkServiceManager::Dispose();
-
-  network_reader.RequestStopSoon();
-  task_runner.RequestStopSoon();
-  network_reader_thread.join();
-  task_runner_thread.join();
 }
 
 void HandleReceiverCommand(absl::string_view command,
@@ -522,14 +515,10 @@ void PublisherDemo(absl::string_view friendly_name) {
 
   constexpr uint16_t server_port = 6667;
 
-  // TODO(rwkeane): Wrap these start/end calls in a class.
-  platform::TaskRunnerImpl task_runner(platform::Clock::now);
-  platform::NetworkReader network_reader;
-  std::thread task_runner_thread(
-      [&task_runner]() { task_runner.RunUntilStopped(); });
-  std::thread network_reader_thread(
-      [&network_reader]() { network_reader.RunUntilStopped(); });
-  platform::UdpSocket::SetLifetimeObserver(&network_reader);
+  platform::TaskRunnerThread task_runner_thread(platform::Clock::now);
+  platform::NetworkReaderThread network_reader_thread;
+  platform::UdpSocket::SetLifetimeObserver(
+      network_reader_thread.network_reader());
 
   PublisherObserver publisher_observer;
   // TODO(btolsch): aggregate initialization probably better?
@@ -540,7 +529,7 @@ void PublisherDemo(absl::string_view friendly_name) {
   publisher_config.connection_server_port = server_port;
 
   auto mdns_publisher = MdnsServicePublisherFactory::Create(
-      publisher_config, &publisher_observer, &task_runner);
+      publisher_config, &publisher_observer, task_runner_thread.task_runner());
 
   ServerConfig server_config;
   std::vector<platform::InterfaceAddresses> interfaces =
@@ -554,7 +543,8 @@ void PublisherDemo(absl::string_view friendly_name) {
                          MessageDemuxer::kDefaultBufferLimit);
   ConnectionServerObserver server_observer;
   auto connection_server = ProtocolConnectionServerFactory::Create(
-      server_config, &demuxer, &server_observer, &task_runner);
+      server_config, &demuxer, &server_observer,
+      task_runner_thread.task_runner());
 
   auto* network_service =
       NetworkServiceManager::Create(nullptr, std::move(mdns_publisher), nullptr,
@@ -572,11 +562,6 @@ void PublisherDemo(absl::string_view friendly_name) {
 
   receiver_delegate.connection.reset();
   CleanupPublisherDemo(network_service);
-
-  network_reader.RequestStopSoon();
-  task_runner.RequestStopSoon();
-  network_reader_thread.join();
-  task_runner_thread.join();
 }
 
 }  // namespace
diff --git a/osp/impl/discovery/mdns/embedder_demo.cc b/osp/impl/discovery/mdns/embedder_demo.cc
index b9a5f917..33c921b5 100644
--- a/osp/impl/discovery/mdns/embedder_demo.cc
+++ b/osp/impl/discovery/mdns/embedder_demo.cc
@@ -16,7 +16,9 @@
 #include "platform/api/time.h"
 #include "platform/base/error.h"
 #include "platform/impl/network_reader.h"
+#include "platform/impl/network_reader_thread.h"
 #include "platform/impl/task_runner.h"
+#include "platform/impl/task_runner_thread.h"
 
 // This file contains a demo of our mDNSResponder wrapper code.  It can both
 // listen for mDNS services and advertise an mDNS service.  The command-line
@@ -356,22 +358,16 @@ int main(int argc, char** argv) {
 
   openscreen::ServiceMap services;
   openscreen::g_services = &services;
-  auto task_runner = std::make_unique<openscreen::platform::TaskRunnerImpl>(
+
+  openscreen::platform::TaskRunnerThread task_runner_thread(
       openscreen::platform::Clock::now);
-  std::thread task_runner_thread(
-      [&task_runner]() { task_runner->RunUntilStopped(); });
-  auto network_reader = std::make_unique<openscreen::platform::NetworkReader>();
-  openscreen::platform::UdpSocket::SetLifetimeObserver(network_reader.get());
-  std::thread network_reader_thread(
-      [&network_reader]() { network_reader->RunUntilStopped(); });
-
-  openscreen::BrowseDemo(task_runner.get(), labels[0], labels[1],
+  openscreen::platform::NetworkReaderThread network_reader_thread;
+  openscreen::platform::UdpSocket::SetLifetimeObserver(
+      network_reader_thread.network_reader());
+
+  openscreen::BrowseDemo(task_runner_thread.task_runner(), labels[0], labels[1],
                          service_instance);
 
-  network_reader->RequestStopSoon();
-  task_runner->RequestStopSoon();
-  network_reader_thread.join();
-  task_runner_thread.join();
   openscreen::g_services = nullptr;
   return 0;
 }