Use SingleThreadedTaskQueue in DirectTransport

webrtc/test/direct_transport.cc

 /*
  *  Copyright (c) 2013 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 "webrtc/test/direct_transport.h"
 
 #include "webrtc/call/call.h"
 #include "webrtc/system_wrappers/include/clock.h"
+#include "webrtc/test/single_threaded_task_queue.h"
 
 namespace webrtc {
 namespace test {
 
 DirectTransport::DirectTransport(
+    SingleThreadedTaskQueueForTesting* task_queue,
     Call* send_call,
     const std::map<uint8_t, MediaType>& payload_type_map)
-    : DirectTransport(FakeNetworkPipe::Config(), send_call, payload_type_map) {}
+    : DirectTransport(task_queue,
+                      FakeNetworkPipe::Config(),
+                      send_call,
+                      payload_type_map) {
+  RTC_DCHECK(task_queue);
+}
 
 DirectTransport::DirectTransport(
+    SingleThreadedTaskQueueForTesting* task_queue,
     const FakeNetworkPipe::Config& config,
     Call* send_call,
     const std::map<uint8_t, MediaType>& payload_type_map)
     : DirectTransport(
+          task_queue,
           config,
           send_call,
-          std::unique_ptr<Demuxer>(new DemuxerImpl(payload_type_map))) {}
+          std::unique_ptr<Demuxer>(new DemuxerImpl(payload_type_map))) {
+  RTC_DCHECK(task_queue);
+}
 
-DirectTransport::DirectTransport(const FakeNetworkPipe::Config& config,
+DirectTransport::DirectTransport(SingleThreadedTaskQueueForTesting* task_queue,
+                                 const FakeNetworkPipe::Config& config,
                                  Call* send_call,
                                  std::unique_ptr<Demuxer> demuxer)
     : send_call_(send_call),
-      packet_event_(false, false),
-      thread_(NetworkProcess, this, "NetworkProcess"),
       clock_(Clock::GetRealTimeClock()),
-      shutting_down_(false),
+      task_queue_(task_queue),
       fake_network_(clock_, config, std::move(demuxer)) {
-  thread_.Start();
   if (send_call_) {
     send_call_->SignalChannelNetworkState(MediaType::AUDIO, kNetworkUp);
     send_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
   }
+  SendPackets();
 }
 
-DirectTransport::~DirectTransport() { StopSending(); }
+DirectTransport::~DirectTransport() {
+  RTC_DCHECK_RUN_ON(&thread_checker_);
+  // Constructor updates |next_scheduled_task_|, so it's guaranteed to
+  // be initialized.
+  task_queue_->CancelTask(next_scheduled_task_);

[nisse-webrtc, 2017/08/18 11:08:29]

So we only need Cancel for tasks which were posted with a delay?

[eladalon, 2017/08/18 12:12:37]

The delay doesn't matter. What matters is that we can only delay tasks which
were posted (asynchronous), not ones that were sent (synchronous).
1. The reason is mostly that we don't really ever need to cancel the latter, and
since this class will only be used in tests, there's no reason to support more
than we need.
2. An additional reason is that an asynchronous task has an obvious entity to
provide the cancellation handle to - the caller of Post(). But a synchronous
task won't be able to do that until the task has already been executed, meaning
it's un-cancellable. Ways around this would be significant overkill, especially
given that it isn't used.

+}
 
 void DirectTransport::SetConfig(const FakeNetworkPipe::Config& config) {
   fake_network_.SetConfig(config);
 }
 
-void DirectTransport::StopSending() {
-  {
-    rtc::CritScope crit(&lock_);
-    shutting_down_ = true;
-  }
-
-  packet_event_.Set();
-  thread_.Stop();
-}
-
 void DirectTransport::SetReceiver(PacketReceiver* receiver) {
+  RTC_DCHECK_RUN_ON(&thread_checker_);
   fake_network_.SetReceiver(receiver);
 }
 
 bool DirectTransport::SendRtp(const uint8_t* data,
                               size_t length,
                               const PacketOptions& options) {
   if (send_call_) {
     rtc::SentPacket sent_packet(options.packet_id,
                                 clock_->TimeInMilliseconds());
     send_call_->OnSentPacket(sent_packet);
   }
   fake_network_.SendPacket(data, length);
-  packet_event_.Set();
   return true;
 }
 
 bool DirectTransport::SendRtcp(const uint8_t* data, size_t length) {
   fake_network_.SendPacket(data, length);
-  packet_event_.Set();
   return true;
 }
 
 int DirectTransport::GetAverageDelayMs() {
   return fake_network_.AverageDelay();
 }
 
 DirectTransport::ForceDemuxer::ForceDemuxer(MediaType media_type)
     : media_type_(media_type) {}
 
 void DirectTransport::ForceDemuxer::SetReceiver(PacketReceiver* receiver) {
   packet_receiver_ = receiver;
 }
 
 void DirectTransport::ForceDemuxer::DeliverPacket(
     const NetworkPacket* packet,
     const PacketTime& packet_time) {
   if (!packet_receiver_)
     return;
   packet_receiver_->DeliverPacket(media_type_, packet->data(),
                                   packet->data_length(), packet_time);
 }
 
-bool DirectTransport::NetworkProcess(void* transport) {
-  return static_cast<DirectTransport*>(transport)->SendPackets();
-}
+void DirectTransport::SendPackets() {
+  RTC_DCHECK_RUN_ON(&thread_checker_);
 
-bool DirectTransport::SendPackets() {
   fake_network_.Process();
-  int64_t wait_time_ms = fake_network_.TimeUntilNextProcess();
-  if (wait_time_ms > 0) {
-    packet_event_.Wait(static_cast<int>(wait_time_ms));
-  }
-  rtc::CritScope crit(&lock_);
-  return shutting_down_ ? false : true;
+
+  int64_t delay_ms = fake_network_.TimeUntilNextProcess();
+  next_scheduled_task_ = task_queue_->PostDelayedTask([this]() {
+    SendPackets();
+  }, delay_ms);
 }
 }  // namespace test
 }  // namespace webrtc