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..0b9e1176f5ba78e0535da6c2ff633bb4b729d6db 100644
--- a/media/base/callback_util.cc
+++ b/media/base/callback_util.cc
@@ -5,130 +5,59 @@
 #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);
-};
-
-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));
-    return;
-  }
-
-  if (closures->empty()) {
-    done_cb.Run();
+        &RunOnMessageLoop, message_loop, status_cb, last_status));
     return;
   }
-
-  ClosureFunc cb = closures->front();
-  closures->pop();
-  cb.Run(base::Bind(
-      &OnSeriesCallback, message_loop, base::Passed(&closures), done_cb));
-}
-
-void RunInSeries(scoped_ptr<std::queue<ClosureFunc> > closures,
-                 const base::Closure& done_cb) {
-  OnSeriesCallback(base::MessageLoopProxy::current(),
-                   closures.Pass(), done_cb);
+  status_cb.Run(last_status);
 }
 
-static void OnStatusCallback(
-    scoped_refptr<base::MessageLoopProxy> message_loop,
+SerialCallbackRunner::SerialCallbackRunner(
     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;
-  }
-
-  PipelineStatusCBFunc status_cb = status_cbs->front();
-  status_cbs->pop();
-  status_cb.Run(base::Bind(
-      &OnStatusCallback, message_loop, base::Passed(&status_cbs), done_cb));
+    const PipelineStatusCB& done_cb)
+    : weak_this_(this),
+      message_loop_(base::MessageLoopProxy::current()),
+      status_cbs_(status_cbs.Pass()),
+      done_cb_(done_cb) {
 }
 
-void RunInSeriesWithStatus(
+SerialCallbackRunner::~SerialCallbackRunner() {}
+
+scoped_ptr<SerialCallbackRunner> SerialCallbackRunner::Run(
     scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs,
     const PipelineStatusCB& done_cb) {
-  OnStatusCallback(base::MessageLoopProxy::current(),
-                   status_cbs.Pass(), done_cb, PIPELINE_OK);
+  scoped_ptr<SerialCallbackRunner> callback_series(
+      new SerialCallbackRunner(status_cbs.Pass(), done_cb));
+  callback_series->RunNextInSeries(PIPELINE_OK);

[Ami GONE FROM CHROMIUM, 2012/08/03 19:29:01]

Post this instead of running inline per acolwell@'s observation.

[scherkus (not reviewing), 2012/08/03 20:19:04]

done in ctor as it has access to message_loop_ and weak_this_

+  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 (status_cbs_->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();
-  }
+  PipelineStatusCBFunc status_cb = status_cbs_->front();
+  status_cbs_->pop();
+  status_cb.Run(base::Bind(&RunOnMessageLoop, message_loop_, base::Bind(
+      &SerialCallbackRunner::RunNextInSeries, weak_this_.GetWeakPtr())));
 }
 
 }  // namespace media