Add support for PendingTaskSafetyFlag to ToQueuedTask.

This keeps usage of ToQueuedTask consistent and avoids callers having
to add additional boiler plate when using the safety flag.

From this:

tq->PostTask(ToQueuedTask([safety = my_safety_flag_]() {
  if (!safety->alive())
    return;
  Foo();
});

to this:

tq->PostTask(ToQueuedTask(my_safety_flag_, []() {
  Foo();
});


Bug: none
Change-Id: I205af56a64dd9839eb845321083d533140d614ca
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/174262
Commit-Queue: Tommi <tommi@webrtc.org>
Reviewed-by: Karl Wiberg <kwiberg@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#31161}

rtc_base/task_utils/BUILD.gn

@@ -42,7 +42,10 @@ rtc_library("pending_task_safety_flag") {
 
 rtc_source_set("to_queued_task") {
   sources = [ "to_queued_task.h" ]
-  deps = [ "../../api/task_queue" ]
+  deps = [
+    ":pending_task_safety_flag",
+    "../../api/task_queue",
+  ]
 }
 
 if (rtc_include_tests) {

rtc_base/task_utils/to_queued_task.h

@@ -16,6 +16,7 @@
 #include <utility>
 
 #include "api/task_queue/queued_task.h"
+#include "rtc_base/task_utils/pending_task_safety_flag.h"
 
 namespace webrtc {
 namespace webrtc_new_closure_impl {
@@ -35,6 +36,25 @@ class ClosureTask : public QueuedTask {
   typename std::decay<Closure>::type closure_;
 };
 
+template <typename Closure>
+class SafetyClosureTask : public QueuedTask {
+ public:
+  explicit SafetyClosureTask(PendingTaskSafetyFlag::Pointer safety,
+                             Closure&& closure)
+      : closure_(std::forward<Closure>(closure)),
+        safety_flag_(std::move(safety)) {}
+
+ private:
+  bool Run() override {
+    if (safety_flag_->alive())
+      closure_();
+    return true;
+  }
+
+  typename std::decay<Closure>::type closure_;
+  PendingTaskSafetyFlag::Pointer safety_flag_;
+};
+
 // Extends ClosureTask to also allow specifying cleanup code.
 // This is useful when using lambdas if guaranteeing cleanup, even if a task
 // was dropped (queue is too full), is required.
@@ -60,6 +80,13 @@ std::unique_ptr<QueuedTask> ToQueuedTask(Closure&& closure) {
       std::forward<Closure>(closure));
 }
 
+template <typename Closure>
+std::unique_ptr<QueuedTask> ToQueuedTask(PendingTaskSafetyFlag::Pointer safety,
+                                         Closure&& closure) {
+  return std::make_unique<webrtc_new_closure_impl::SafetyClosureTask<Closure>>(
+      std::move(safety), std::forward<Closure>(closure));
+}
+
 template <typename Closure, typename Cleanup>
 std::unique_ptr<QueuedTask> ToQueuedTask(Closure&& closure, Cleanup&& cleanup) {
   return std::make_unique<

rtc_base/task_utils/to_queued_task_unittest.cc

@@ -126,5 +126,22 @@ TEST(ToQueuedTaskTest, AcceptsMoveOnlyCleanup) {
   RunTask(std::move(task));
 }
 
+TEST(ToQueuedTaskTest, PendingTaskSafetyFlag) {
+  PendingTaskSafetyFlag::Pointer flag(PendingTaskSafetyFlag::Create());
+
+  int count = 0;
+  // Create two identical tasks that increment the |count|.
+  auto task1 = ToQueuedTask(flag, [&count]() { ++count; });
+  auto task2 = ToQueuedTask(flag, [&count]() { ++count; });
+
+  EXPECT_EQ(0, count);
+  RunTask(std::move(task1));
+  EXPECT_EQ(1, count);
+  flag->SetNotAlive();
+  // Now task2 should actually not run.
+  RunTask(std::move(task2));
+  EXPECT_EQ(1, count);
+}
+
 }  // namespace
 }  // namespace webrtc