Move AudioRenderer out of Filter heirarchy.

media/base/callback_util.cc

+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "media/base/callback_util.h"
+
+#include "base/bind.h"
+#include "base/synchronization/lock.h"
+#include "base/memory/ref_counted.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

[scherkus (not reviewing), 2012/07/19 21:29:02]

I tried the ref-counting approach for fun but something I overlooked yesterday
was that relying on ref-counting assumed that all closures would release their
ref via Run() but refs may also get released if cbs are reset / deallocated,
which would lead to mistakenly executing done_cb!

it's still refcounted (to avoid use-after-free) but we now require the closure
to be executed an exact # of times + DCHECK if we execute too many times

+// 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:
+  explicit CountingCB(int count, const base::Closure& done_cb)

[acolwell GONE FROM CHROMIUM, 2012/07/19 23:09:14]

nit: remove explicit now.

[scherkus (not reviewing), 2012/07/19 23:46:47]

Done.

+      : message_loop_(base::MessageLoopProxy::current()),
+        count_(count),
+        completed_(0),
+        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_);
+      completed_++;
+      DCHECK_LE(completed_, count_) << "CountingCB executed too many times";
+      if (completed_ != count_)
+        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_;
+  int completed_;

[acolwell GONE FROM CHROMIUM, 2012/07/19 23:09:14]

nit: Consider decrementing count_ so you don't need this.

[scherkus (not reviewing), 2012/07/19 23:46:47]

Done.

+  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) {
+  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();
+    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);
+}
+
+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;
+  }
+
+  PipelineStatusCBFunc status_cb = status_cbs->front();
+  status_cbs->pop();
+  status_cb.Run(base::Bind(
+      &OnStatusCallback, message_loop, base::Passed(&status_cbs), done_cb));
+}
+
+void RunInSeriesWithStatus(
+    scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs,
+    const PipelineStatusCB& done_cb) {
+  OnStatusCallback(base::MessageLoopProxy::current(),
+                   status_cbs.Pass(), done_cb, PIPELINE_OK);
+}
+
+void RunInParallel(scoped_ptr<std::queue<ClosureFunc> > closures,
+                   const base::Closure& done_cb) {
+  if (closures->empty()) {
+    done_cb.Run();
+    return;
+  }
+
+  scoped_refptr<CountingCB> counting_cb =
+      new CountingCB(closures->size(), done_cb);
+  while (!closures->empty()) {
+    closures->front().Run(counting_cb->GetClosure());
+    closures->pop();
+  }
+}
+
+}  // namespace media