Use SingleThreadedTaskQueue in DirectTransport

webrtc/test/single_threaded_task_queue.cc

+/*
+ *  Copyright (c) 2017 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/single_threaded_task_queue.h"
+
+#include <memory>
+
+#include "webrtc/rtc_base/checks.h"
+
+namespace webrtc {
+namespace test {
+
+SingleThreadedTaskQueue::SingleThreadedTaskQueue(const char* name)
+    : thread_(Run, this, name),
+      running_(true),
+      next_identifier_(0),
+      event_(false, false) {
+  thread_.Start();
+}
+
+SingleThreadedTaskQueue::~SingleThreadedTaskQueue() {
+  RTC_DCHECK_RUN_ON(&owner_thread_checker_);
+  {
+    rtc::CritScope lock(&cs_);
+    running_ = false;
+  }
+  event_.Set();
+  thread_.Stop();
+}
+
+SingleThreadedTaskQueue::TaskIdentifier SingleThreadedTaskQueue::PostTask(
+    Task task) {
+  rtc::CritScope lock(&cs_);
+  TaskIdentifier identifier = next_identifier_++;
+  tasks_.emplace_back(identifier, task);
+  if (tasks_.size() == 1) {
+    event_.Set();
+  }
+  return identifier;
+}
+
+void SingleThreadedTaskQueue::SendTask(Task task) {
+  rtc::Event done(true, false);
+  PostTask([&task, &done]() {
+    task();
+    done.Set();
+  });
+  done.Wait(rtc::Event::kForever);
+}
+
+bool SingleThreadedTaskQueue::CancelTask(TaskIdentifier task_id) {
+  rtc::CritScope lock(&cs_);
+  for (auto it = tasks_.cbegin(); it != tasks_.cend(); it++) {
+    if (it->first == task_id) {
+      tasks_.erase(it);
+      return true;
+    }
+  }
+  return false;
+}
+
+void SingleThreadedTaskQueue::Run(void* obj) {
+  static_cast<SingleThreadedTaskQueue*>(obj)->RunLoop();
+}
+
+void SingleThreadedTaskQueue::RunLoop() {
+  while (true) {
+    std::pair<TaskIdentifier, Task> task;
+    bool empty;
+    {
+      rtc::CritScope lock(&cs_);
+      if (!running_) {
+        return;
+      }
+      empty = tasks_.empty();
+      if (!empty) {
+        task = tasks_.front();
+        tasks_.pop_front();
+      }
+    }
+
+    if (empty) {
+      event_.Wait(rtc::Event::kForever);
+    } else {
+      task.second();
+    }
+  }
+}
+
+}  // namespace test
+}  // namespace webrtc