Introduce the concept of media task runner.

chromecast/media/cma/base/balanced_media_task_runner_factory.cc

Index: chromecast/media/cma/base/balanced_media_task_runner_factory.cc
diff --git a/chromecast/media/cma/base/balanced_media_task_runner_factory.cc b/chromecast/media/cma/base/balanced_media_task_runner_factory.cc
new file mode 100644
index 0000000000000000000000000000000000000000..93132603cf3952e34c6fcec7f297e02b2b93e96f
--- /dev/null
+++ b/chromecast/media/cma/base/balanced_media_task_runner_factory.cc
@@ -0,0 +1,252 @@
+// Copyright 2014 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 "chromecast/media/cma/base/balanced_media_task_runner_factory.h"
+
+#include <map>
+
+#include "base/bind.h"
+#include "base/callback_helpers.h"
+#include "base/logging.h"
+#include "base/single_thread_task_runner.h"
+#include "chromecast/media/cma/base/media_task_runner.h"
+#include "media/base/buffers.h"
+
+namespace chromecast {
+namespace media {
+
+// MediaTaskRunnerWithNotification -
+// Media task runner which also behaves as a media task runner observer.
+class MediaTaskRunnerWithNotification : public MediaTaskRunner {
+ public:
+  // Wraps a MediaTaskRunner so that a third party can:
+  // - be notified when a PostMediaTask is performed on this media task runner.
+  //   |new_task_cb| is invoked in that case.
+  // - monitor the lifetime of the media task runner, i.e. check when the media
+  //   task runner is not needed anymore.
+  //   |shutdown_cb| is invoked in that case.
+  MediaTaskRunnerWithNotification(
+      const scoped_refptr<MediaTaskRunner>& media_task_runner,
+      const base::Closure& new_task_cb,
+      const base::Closure& shutdown_cb);
+
+  // MediaTaskRunner implementation.
+  virtual bool PostMediaTask(
+      const tracked_objects::Location& from_here,
+      const base::Closure& task,
+      base::TimeDelta timestamp) OVERRIDE;
+
+ private:
+  virtual ~MediaTaskRunnerWithNotification();
+
+  scoped_refptr<MediaTaskRunner> const media_task_runner_;
+
+  const base::Closure new_task_cb_;
+  const base::Closure shutdown_cb_;
+
+  DISALLOW_COPY_AND_ASSIGN(MediaTaskRunnerWithNotification);
+};
+
+MediaTaskRunnerWithNotification::MediaTaskRunnerWithNotification(
+    const scoped_refptr<MediaTaskRunner>& media_task_runner,
+    const base::Closure& new_task_cb,
+    const base::Closure& shutdown_cb)
+  : media_task_runner_(media_task_runner),
+    new_task_cb_(new_task_cb),
+    shutdown_cb_(shutdown_cb) {
+}
+
+MediaTaskRunnerWithNotification::~MediaTaskRunnerWithNotification() {
+  shutdown_cb_.Run();
+}
+
+bool MediaTaskRunnerWithNotification::PostMediaTask(
+    const tracked_objects::Location& from_here,
+    const base::Closure& task,
+    base::TimeDelta timestamp) {
+  bool may_run_in_future =
+      media_task_runner_->PostMediaTask(from_here, task, timestamp);
+  if (may_run_in_future)
+    new_task_cb_.Run();
+  return may_run_in_future;
+}
+
+
+// BalancedMediaTaskRunner -
+// Run media tasks whose timestamp is less or equal to a max timestamp.
+//
+// Restrictions of BalancedMediaTaskRunner:
+// - Can have at most one task in the queue.
+// - Tasks should be given by increasing timestamps.
+class BalancedMediaTaskRunner
+    : public MediaTaskRunner {
+ public:
+  explicit BalancedMediaTaskRunner(
+      const scoped_refptr<base::SingleThreadTaskRunner>& task_runner);
+
+  // Schedule tasks whose timestamp is less than or equal to |max_timestamp|.
+  void ScheduleWork(base::TimeDelta max_timestamp);
+
+  // Return the timestamp of the last media task.
+  // Return ::media::kNoTimestamp() if no media task has been posted.
+  base::TimeDelta GetMediaTimestamp() const;
+
+  // MediaTaskRunner implementation.
+  virtual bool PostMediaTask(
+      const tracked_objects::Location& from_here,
+      const base::Closure& task,
+      base::TimeDelta timestamp) OVERRIDE;
+
+ private:
+  virtual ~BalancedMediaTaskRunner();
+
+  scoped_refptr<base::SingleThreadTaskRunner> const task_runner_;
+
+  // Protects the following variables.
+  mutable base::Lock lock_;
+
+  // Possible pending media task.
+  tracked_objects::Location from_here_;
+  base::Closure pending_task_;
+
+  // Timestamp of the last posted task.
+  // Is initialized to ::media::kNoTimestamp().
+  base::TimeDelta last_timestamp_;
+
+  DISALLOW_COPY_AND_ASSIGN(BalancedMediaTaskRunner);
+};
+
+BalancedMediaTaskRunner::BalancedMediaTaskRunner(
+    const scoped_refptr<base::SingleThreadTaskRunner>& task_runner)
+  : task_runner_(task_runner),
+    last_timestamp_(::media::kNoTimestamp()) {
+}
+
+BalancedMediaTaskRunner::~BalancedMediaTaskRunner() {
+}
+
+void BalancedMediaTaskRunner::ScheduleWork(base::TimeDelta max_media_time) {
+  base::Closure task;
+  {
+    base::AutoLock auto_lock(lock_);
+    if (pending_task_.is_null())
+      return;
+
+    if (last_timestamp_ != ::media::kNoTimestamp() &&
+        last_timestamp_ >= max_media_time) {
+      return;
+    }
+
+    task = base::ResetAndReturn(&pending_task_);
+  }
+  task_runner_->PostTask(from_here_, task);
+}
+
+base::TimeDelta BalancedMediaTaskRunner::GetMediaTimestamp() const {
+  base::AutoLock auto_lock(lock_);
+  return last_timestamp_;
+}
+
+bool BalancedMediaTaskRunner::PostMediaTask(
+    const tracked_objects::Location& from_here,
+    const base::Closure& task,
+    base::TimeDelta timestamp) {
+  DCHECK(!task.is_null());
+
+  // Pass through for a task with no timestamp.
+  if (timestamp == ::media::kNoTimestamp()) {
+    return task_runner_->PostTask(from_here, task);
+  }
+
+  base::AutoLock auto_lock(lock_);
+
+  // Timestamps must be in order.
+  // Any task that does not meet that condition is simply discarded.
+  if (last_timestamp_ != ::media::kNoTimestamp() &&
+      timestamp < last_timestamp_) {
+    return false;
+  }
+
+  // Only support one pending task at a time.
+  DCHECK(pending_task_.is_null());
+  from_here_ = from_here;
+  pending_task_ = task;
+  last_timestamp_ = timestamp;
+
+  return true;
+}
+
+
+BalancedMediaTaskRunnerFactory::BalancedMediaTaskRunnerFactory(
+    base::TimeDelta max_delta)
+  : max_delta_(max_delta) {
+}
+
+BalancedMediaTaskRunnerFactory::~BalancedMediaTaskRunnerFactory() {
+}
+
+scoped_refptr<MediaTaskRunner>
+BalancedMediaTaskRunnerFactory::CreateMediaTaskRunner(
+    const scoped_refptr<base::SingleThreadTaskRunner>& task_runner) {
+  scoped_refptr<BalancedMediaTaskRunner> media_task_runner(
+      new BalancedMediaTaskRunner(task_runner));
+  scoped_refptr<MediaTaskRunnerWithNotification> media_task_runner_wrapper(
+      new MediaTaskRunnerWithNotification(
+          media_task_runner,
+          base::Bind(&BalancedMediaTaskRunnerFactory::OnNewTask, this),
+          base::Bind(
+              &BalancedMediaTaskRunnerFactory::UnregisterMediaTaskRunner,
+              this, media_task_runner)));
+  base::AutoLock auto_lock(lock_);
+  // Note that |media_task_runner| is inserted here and
+  // not |media_task_runner_wrapper|. Otherwise, we would always have one
+  // ref on |media_task_runner_wrapper| and would never get the release
+  // notification.
+  // When |media_task_runner_wrapper| is going away,
+  // BalancedMediaTaskRunnerFactory will receive a notification and will in
+  // turn remove |media_task_runner|.
+  task_runners_.insert(media_task_runner);
+  return media_task_runner_wrapper;
+}
+
+void BalancedMediaTaskRunnerFactory::OnNewTask() {
+  typedef
+      std::multimap<base::TimeDelta, scoped_refptr<BalancedMediaTaskRunner> >
+      TaskRunnerMap;
+  TaskRunnerMap runnable_task_runner;
+
+  base::AutoLock auto_lock(lock_);
+
+  // Get the minimum timestamp among all streams.
+  for (MediaTaskRunnerSet::const_iterator it = task_runners_.begin();
+       it != task_runners_.end(); ++it) {
+    base::TimeDelta timestamp((*it)->GetMediaTimestamp());
+    if (timestamp == ::media::kNoTimestamp())
+      continue;
+    runnable_task_runner.insert(
+        std::pair<base::TimeDelta, scoped_refptr<BalancedMediaTaskRunner> >(
+            timestamp, *it));
+  }
+
+  // If there is no media task, just returns.
+  if (runnable_task_runner.empty())
+    return;
+
+  // Run tasks which meet the balancing criteria.
+  base::TimeDelta min_timestamp(runnable_task_runner.begin()->first);
+  base::TimeDelta max_timestamp = min_timestamp + max_delta_;
+  for (TaskRunnerMap::iterator it = runnable_task_runner.begin();
+       it != runnable_task_runner.end(); ++it) {
+    (*it).second->ScheduleWork(max_timestamp);
+  }
+}
+
+void BalancedMediaTaskRunnerFactory::UnregisterMediaTaskRunner(
+      const scoped_refptr<BalancedMediaTaskRunner>& media_task_runner) {
+  base::AutoLock auto_lock(lock_);
+  task_runners_.erase(media_task_runner);
+}
+
+}  // namespace media
+}  // namespace chromecast