Replace RunInSeries() and RunInParallel() with CallbackSeries helper class.

media/base/callback_util.cc

Index: media/base/callback_util.cc
diff --git a/media/base/callback_util.cc b/media/base/callback_util.cc
index 6ab7818c8b193103e8750febb5192ecfccba80cb..4703c29fc7744f41ddcfb7068c1ced8ac1814d33 100644
--- a/media/base/callback_util.cc
+++ b/media/base/callback_util.cc
@@ -5,130 +5,87 @@
 #include "media/base/callback_util.h"
 
 #include "base/bind.h"
-#include "base/synchronization/lock.h"
-#include "base/memory/ref_counted.h"
+#include "base/callback_helpers.h"
 #include "base/message_loop.h"
 #include "base/message_loop_proxy.h"
 
 namespace media {
 
-// Executes the given closure if and only if the closure returned by
-// GetClosure() has been executed exactly |count| times.
-//
-// |done_cb| will be executed on the same thread that created the CountingCB.
-class CountingCB : public base::RefCountedThreadSafe<CountingCB> {
- public:
-  CountingCB(int count, const base::Closure& done_cb)
-      : message_loop_(base::MessageLoopProxy::current()),
-        count_(count),
-        done_cb_(done_cb) {
-  }
-
-  // Returns a closure bound to this object.
-  base::Closure GetClosure() {
-    return base::Bind(&CountingCB::OnCallback, this);
-  }
-
- protected:
-  friend class base::RefCountedThreadSafe<CountingCB>;
-  virtual ~CountingCB() {}
-
- private:
-  void OnCallback() {
-    {
-      base::AutoLock l(lock_);
-      count_--;
-      DCHECK_GE(count_, 0) << "CountingCB executed too many times";
-      if (count_ != 0)
-        return;
-    }
-
-    if (!message_loop_->BelongsToCurrentThread()) {
-      message_loop_->PostTask(FROM_HERE, done_cb_);
-      return;
-    }
-
-    done_cb_.Run();
-  }
-
-  scoped_refptr<base::MessageLoopProxy> message_loop_;
-  base::Lock lock_;
-  int count_;
-  base::Closure done_cb_;
-
-  DISALLOW_COPY_AND_ASSIGN(CountingCB);
-};
+// Converts a bound function accepting a Closure into a bound function
+// accepting a PipelineStatusCB. Since closures have no way of reporting a
+// status |status_cb| is executed with PIPELINE_OK.
+void RunBoundClosure(

[Ami GONE FROM CHROMIUM, 2012/08/03 20:31:43]

static

+    const SerialCallbackRunner::BoundClosure& bound_closure,
+    const PipelineStatusCB& status_cb) {
+  bound_closure.Run(base::Bind(status_cb, PIPELINE_OK));
+}
 
-static void OnSeriesCallback(
-    scoped_refptr<base::MessageLoopProxy> message_loop,
-    scoped_ptr<std::queue<ClosureFunc> > closures,
-    const base::Closure& done_cb) {
+// Runs |status_cb| with |last_status| on |message_loop|, trampolining if
+// necessary.
+static void RunOnMessageLoop(
+    const scoped_refptr<base::MessageLoopProxy>& message_loop,
+    const PipelineStatusCB& status_cb,
+    PipelineStatus last_status) {
   if (!message_loop->BelongsToCurrentThread()) {
     message_loop->PostTask(FROM_HERE, base::Bind(
-        &OnSeriesCallback, message_loop, base::Passed(&closures), done_cb));
+        &RunOnMessageLoop, message_loop, status_cb, last_status));
     return;
   }
+  status_cb.Run(last_status);
+}
 
-  if (closures->empty()) {
-    done_cb.Run();
-    return;
-  }
+SerialCallbackRunner::Queue::Queue() {}
+SerialCallbackRunner::Queue::~Queue() {}
 
-  ClosureFunc cb = closures->front();
-  closures->pop();
-  cb.Run(base::Bind(
-      &OnSeriesCallback, message_loop, base::Passed(&closures), done_cb));
+void SerialCallbackRunner::Queue::Push(
+    const BoundClosure& bound_closure) {
+  bound_fns_.push(base::Bind(&RunBoundClosure, bound_closure));
 }
 
-void RunInSeries(scoped_ptr<std::queue<ClosureFunc> > closures,
-                 const base::Closure& done_cb) {
-  OnSeriesCallback(base::MessageLoopProxy::current(),
-                   closures.Pass(), done_cb);
+void SerialCallbackRunner::Queue::Push(
+    const BoundPipelineStatusCB& bound_status_cb) {
+  bound_fns_.push(bound_status_cb);
 }
 
-static void OnStatusCallback(
-    scoped_refptr<base::MessageLoopProxy> message_loop,
-    scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs,
-    const PipelineStatusCB& done_cb,
-    PipelineStatus last_status) {
-  if (!message_loop->BelongsToCurrentThread()) {
-    message_loop->PostTask(FROM_HERE, base::Bind(
-        &OnStatusCallback, message_loop, base::Passed(&status_cbs), done_cb,
-        last_status));
-    return;
-  }
-
-  if (status_cbs->empty() || last_status != PIPELINE_OK) {
-    done_cb.Run(last_status);
-    return;
-  }
+SerialCallbackRunner::BoundPipelineStatusCB SerialCallbackRunner::Queue::Pop() {
+  BoundPipelineStatusCB bound_fn = bound_fns_.front();
+  bound_fns_.pop();
+  return bound_fn;
+}
 
-  PipelineStatusCBFunc status_cb = status_cbs->front();
-  status_cbs->pop();
-  status_cb.Run(base::Bind(
-      &OnStatusCallback, message_loop, base::Passed(&status_cbs), done_cb));
+SerialCallbackRunner::SerialCallbackRunner(
+    scoped_ptr<Queue> bound_fns, const PipelineStatusCB& done_cb)
+    : weak_this_(this),
+      message_loop_(base::MessageLoopProxy::current()),
+      bound_fns_(bound_fns.Pass()),
+      done_cb_(done_cb) {
+  message_loop_->PostTask(FROM_HERE, base::Bind(
+      &SerialCallbackRunner::RunNextInSeries, weak_this_.GetWeakPtr(),
+      PIPELINE_OK));
 }
 
-void RunInSeriesWithStatus(
-    scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs,
+SerialCallbackRunner::~SerialCallbackRunner() {}
+
+scoped_ptr<SerialCallbackRunner> SerialCallbackRunner::Run(
+    scoped_ptr<Queue> bound_fns,
     const PipelineStatusCB& done_cb) {
-  OnStatusCallback(base::MessageLoopProxy::current(),
-                   status_cbs.Pass(), done_cb, PIPELINE_OK);
+  scoped_ptr<SerialCallbackRunner> callback_series(
+      new SerialCallbackRunner(bound_fns.Pass(), done_cb));
+  return callback_series.Pass();
 }
 
-void RunInParallel(scoped_ptr<std::queue<ClosureFunc> > closures,
-                   const base::Closure& done_cb) {
-  if (closures->empty()) {
-    done_cb.Run();
+void SerialCallbackRunner::RunNextInSeries(PipelineStatus last_status) {
+  DCHECK(message_loop_->BelongsToCurrentThread());
+  DCHECK(!done_cb_.is_null());
+
+  if (bound_fns_->empty() || last_status != PIPELINE_OK) {
+    base::ResetAndReturn(&done_cb_).Run(last_status);
     return;
   }
 
-  scoped_refptr<CountingCB> counting_cb =
-      new CountingCB(closures->size(), done_cb);
-  while (!closures->empty()) {
-    closures->front().Run(counting_cb->GetClosure());
-    closures->pop();
-  }
+  BoundPipelineStatusCB bound_fn = bound_fns_->Pop();
+  bound_fn.Run(base::Bind(&RunOnMessageLoop, message_loop_, base::Bind(
+      &SerialCallbackRunner::RunNextInSeries, weak_this_.GetWeakPtr())));
 }
 
 }  // namespace media