Move AudioRenderer out of Filter heirarchy.

media/base/pipeline.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/pipeline.h"
 
 #include <algorithm>
 
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/callback_helpers.h"
 #include "base/compiler_specific.h"
 #include "base/metrics/histogram.h"
 #include "base/message_loop.h"
 #include "base/stl_util.h"
 #include "base/string_util.h"
 #include "base/synchronization/condition_variable.h"
 #include "media/base/audio_decoder.h"
 #include "media/base/audio_renderer.h"
+#include "media/base/callback_util.h"
 #include "media/base/clock.h"
 #include "media/base/composite_filter.h"
 #include "media/base/filter_collection.h"
 #include "media/base/media_log.h"
 #include "media/base/video_decoder.h"
 #include "media/base/video_renderer.h"
 
 using base::TimeDelta;
 
 namespace media {
@@ skipping 27 matching lines @@
 
 media::PipelineStatus PipelineStatusNotification::status() {
   base::AutoLock auto_lock(lock_);
   DCHECK(notified_);
   return status_;
 }
 
 struct Pipeline::PipelineInitState {
   scoped_refptr<AudioDecoder> audio_decoder;
   scoped_refptr<VideoDecoder> video_decoder;
-  scoped_refptr<AudioRenderer> audio_renderer;
   scoped_refptr<VideoRenderer> video_renderer;
   scoped_refptr<CompositeFilter> composite;
 };
 
 Pipeline::Pipeline(MessageLoop* message_loop, MediaLog* media_log)
     : message_loop_(message_loop->message_loop_proxy()),
       media_log_(media_log),
       clock_(new Clock(&base::Time::Now)),
       waiting_for_clock_update_(false),
       state_(kCreated),
@@ skipping 232 matching lines @@
   DCHECK(message_loop_->BelongsToCurrentThread());
   if (!seek_pending_)
     return false;
   DCHECK(kSeeking == state_ || kPausing == state_ ||
          kFlushing == state_ || kStarting == state_)
       << "Current state : " << state_;
   return true;
 }
 
 void Pipeline::ReportStatus(const PipelineStatusCB& cb, PipelineStatus status) {
+  DCHECK(message_loop_->BelongsToCurrentThread());
   if (cb.is_null())
     return;
   cb.Run(status);
   // Prevent double-reporting of errors to clients.
   if (status != PIPELINE_OK)
     error_cb_.Reset();
 }
 
 void Pipeline::FinishInitialization() {
   DCHECK(message_loop_->BelongsToCurrentThread());
@@ skipping 53 matching lines @@
 TimeDelta Pipeline::GetTime() const {
   DCHECK(IsRunning());
   return GetCurrentTime();
 }
 
 TimeDelta Pipeline::GetDuration() const {
   DCHECK(IsRunning());
   return GetMediaDuration();
 }
 
+void Pipeline::OnAudioDisabled() {
+  DCHECK(IsRunning());
+  message_loop_->PostTask(FROM_HERE, base::Bind(
+      &Pipeline::AudioDisabledTask, this));
+  media_log_->AddEvent(
+      media_log_->CreateEvent(MediaLogEvent::AUDIO_RENDERER_DISABLED));
+}
+
 void Pipeline::OnAudioTimeUpdate(TimeDelta time, TimeDelta max_time) {
   DCHECK(time <= max_time);
   DCHECK(IsRunning());
   base::AutoLock auto_lock(lock_);
 
   if (!has_audio_)
     return;
   if (waiting_for_clock_update_ && time < clock_->Elapsed())
     return;
 
@@ skipping 50 matching lines @@
   // Since the byte->time calculation is approximate, fudge the beginning &
   // ending areas to look better.
   TimeDelta epsilon = clock_->Duration() / 100;
   if (time_offset < epsilon)
     return TimeDelta();
   if (time_offset + epsilon > clock_->Duration())
     return clock_->Duration();
   return time_offset;
 }
 
+void Pipeline::DoPause(const base::Closure& done_cb) {
+  DCHECK(message_loop_->BelongsToCurrentThread());
+  scoped_ptr<std::queue<ClosureFunc> > closures(new std::queue<ClosureFunc>);
+
+  if (audio_renderer_)
+    closures->push(base::Bind(&AudioRenderer::Pause, audio_renderer_));
+
+  if (pipeline_filter_)
+    closures->push(base::Bind(&Filter::Pause, pipeline_filter_));
+
+  RunInSeries(closures.Pass(), done_cb);
+}
+
+void Pipeline::DoFlush(const base::Closure& done_cb) {
+  DCHECK(message_loop_->BelongsToCurrentThread());
+  scoped_ptr<std::queue<ClosureFunc> > closures(new std::queue<ClosureFunc>);
+
+  if (audio_renderer_)
+    closures->push(base::Bind(&AudioRenderer::Flush, audio_renderer_));
+
+  if (pipeline_filter_)
+    closures->push(base::Bind(&Filter::Flush, pipeline_filter_));
+
+  RunInParallel(closures.Pass(), done_cb);
+}
+
+void Pipeline::DoPlay(const base::Closure& done_cb) {
+  DCHECK(message_loop_->BelongsToCurrentThread());
+  scoped_ptr<std::queue<ClosureFunc> > closures(new std::queue<ClosureFunc>);
+
+  if (audio_renderer_)
+    closures->push(base::Bind(&AudioRenderer::Play, audio_renderer_));
+
+  if (pipeline_filter_)
+    closures->push(base::Bind(&Filter::Play, pipeline_filter_));
+
+  RunInSeries(closures.Pass(), done_cb);
+}
+
+void Pipeline::DoStop(const base::Closure& done_cb) {
+  DCHECK(message_loop_->BelongsToCurrentThread());
+  scoped_ptr<std::queue<ClosureFunc> > closures(new std::queue<ClosureFunc>);
+
+  if (demuxer_)
+    closures->push(base::Bind(&Demuxer::Stop, demuxer_));
+
+  if (audio_renderer_)
+    closures->push(base::Bind(&AudioRenderer::Stop, audio_renderer_));
+
+  if (pipeline_filter_)
+    closures->push(base::Bind(&Filter::Stop, pipeline_filter_));
+
+  RunInSeries(closures.Pass(), done_cb);
+}
+
 void Pipeline::AddBufferedByteRange(int64 start, int64 end) {
   DCHECK(IsRunning());
   base::AutoLock auto_lock(lock_);
   buffered_byte_ranges_.Add(start, end);
   did_loading_progress_ = true;
 }
 
 void Pipeline::AddBufferedTimeRange(base::TimeDelta start,
                                     base::TimeDelta end) {
   DCHECK(IsRunning());
@@ skipping 11 matching lines @@
   natural_size_ = size;
 }
 
 void Pipeline::NotifyEnded() {
   DCHECK(IsRunning());
   message_loop_->PostTask(FROM_HERE, base::Bind(
       &Pipeline::NotifyEndedTask, this));
   media_log_->AddEvent(media_log_->CreateEvent(MediaLogEvent::ENDED));
 }
 
-void Pipeline::DisableAudioRenderer() {
-  DCHECK(IsRunning());
-
-  // Disable renderer on the message loop.
-  message_loop_->PostTask(FROM_HERE, base::Bind(
-      &Pipeline::DisableAudioRendererTask, this));
-  media_log_->AddEvent(
-      media_log_->CreateEvent(MediaLogEvent::AUDIO_RENDERER_DISABLED));
-}
-
 // Called from any thread.
 void Pipeline::OnFilterInitialize(PipelineStatus status) {
   // Continue the initialize task by proceeding to the next stage.
   message_loop_->PostTask(FROM_HERE, base::Bind(
       &Pipeline::InitializeTask, this, status));
 }
 
 // Called from any thread.
 void Pipeline::OnFilterStateTransition() {
   message_loop_->PostTask(FROM_HERE, base::Bind(
@@ skipping 137 matching lines @@
     pipeline_filter_ = pipeline_init_state_->composite;
 
     // Clear init state since we're done initializing.
     pipeline_init_state_.reset();
 
     // Initialization was successful, we are now considered paused, so it's safe
     // to set the initial playback rate and volume.
     PlaybackRateChangedTask(GetPlaybackRate());
     VolumeChangedTask(GetVolume());
 
-    // Fire a seek request to get the renderers to preroll. We don't need to
-    // tell the demuxer to seek since it should already be at the start.
+    // Fire a seek request to get the renderers to preroll. We can skip a seek
+    // here as the demuxer should be at the start of the stream.
     seek_pending_ = true;
     SetState(kSeeking);
     seek_timestamp_ = demuxer_->GetStartTime();
-    OnDemuxerSeekDone(seek_timestamp_, PIPELINE_OK);
+    DoSeek(seek_timestamp_, true,
+           base::Bind(&Pipeline::OnFilterStateTransitionWithStatus, this));
   }
 }
 
 // This method is called as a result of the client calling Pipeline::Stop() or
 // as the result of an error condition.
 // We stop the filters in the reverse order.
 //
 // TODO(scherkus): beware!  this can get posted multiple times since we post
 // Stop() tasks even if we've already stopped.  Perhaps this should no-op for
 // additional calls, however most of this logic will be changing.
@@ skipping 68 matching lines @@
 
   {
     base::AutoLock auto_lock(lock_);
     clock_->SetPlaybackRate(playback_rate);
   }
 
   // Notify |pipeline_filter_| if it has been initialized. If initialization
   // hasn't completed yet, the playback rate will be set when initialization
   // completes.
   if (pipeline_filter_) {
-    DCHECK(demuxer_);
     demuxer_->SetPlaybackRate(playback_rate);
     pipeline_filter_->SetPlaybackRate(playback_rate);
+
+    if (audio_renderer_)
+      audio_renderer_->SetPlaybackRate(playback_rate_);
   }
 }
 
 void Pipeline::VolumeChangedTask(float volume) {
   DCHECK(message_loop_->BelongsToCurrentThread());
   if (!running_ || tearing_down_)
     return;
 
   if (audio_renderer_)
     audio_renderer_->SetVolume(volume);
@@ skipping 25 matching lines @@
   SetState(kPausing);
   seek_timestamp_ = std::max(time, demuxer_->GetStartTime());
   seek_cb_ = seek_cb;
 
   // Kick off seeking!
   {
     base::AutoLock auto_lock(lock_);
     if (clock_->IsPlaying())
       clock_->Pause();
   }
-  pipeline_filter_->Pause(
-      base::Bind(&Pipeline::OnFilterStateTransition, this));
+  DoPause(base::Bind(&Pipeline::OnFilterStateTransition, this));
 }
 
 void Pipeline::NotifyEndedTask() {
   DCHECK(message_loop_->BelongsToCurrentThread());
 
   // We can only end if we were actually playing.
   if (state_ != kStarted) {
     return;
   }
 
@@ skipping 19 matching lines @@
   // Transition to ended, executing the callback if present.
   SetState(kEnded);
   {
     base::AutoLock auto_lock(lock_);
     clock_->EndOfStream();
   }
 
   ReportStatus(ended_cb_, status_);
 }
 
-void Pipeline::DisableAudioRendererTask() {
+void Pipeline::AudioDisabledTask() {
   DCHECK(message_loop_->BelongsToCurrentThread());
 
   base::AutoLock auto_lock(lock_);
   has_audio_ = false;
   audio_disabled_ = true;
 
   // Notify our demuxer that we're no longer rendering audio.
   demuxer_->OnAudioRendererDisabled();
 
   // Start clock since there is no more audio to
@@ skipping 26 matching lines @@
   // to the next state if needed.
   SetState(FindNextState(state_));
   if (state_ == kSeeking) {
     base::AutoLock auto_lock(lock_);
     clock_->SetTime(seek_timestamp_, seek_timestamp_);
   }
 
   // Carry out the action for the current state.
   if (TransientState(state_)) {
     if (state_ == kPausing) {
-      pipeline_filter_->Pause(
-          base::Bind(&Pipeline::OnFilterStateTransition, this));
+      DoPause(base::Bind(&Pipeline::OnFilterStateTransition, this));
     } else if (state_ == kFlushing) {
-      pipeline_filter_->Flush(
-          base::Bind(&Pipeline::OnFilterStateTransition, this));
+      DoFlush(base::Bind(&Pipeline::OnFilterStateTransition, this));
     } else if (state_ == kSeeking) {
-      DoSeek(seek_timestamp_);
+      DoSeek(seek_timestamp_, false,
+             base::Bind(&Pipeline::OnFilterStateTransitionWithStatus, this));
     } else if (state_ == kStarting) {
-      pipeline_filter_->Play(
-          base::Bind(&Pipeline::OnFilterStateTransition, this));
+      DoPlay(base::Bind(&Pipeline::OnFilterStateTransition, this));
     } else if (state_ == kStopping) {
       DoStop(base::Bind(&Pipeline::OnFilterStateTransition, this));
     } else {
       NOTREACHED() << "Unexpected state: " << state_;
     }
   } else if (state_ == kStarted) {
     FinishInitialization();
 
     // Finally, complete the seek.
     seek_pending_ = false;
@@ skipping 25 matching lines @@
 
 void Pipeline::TeardownStateTransitionTask() {
   DCHECK(IsPipelineTearingDown());
   switch (state_) {
     case kStopping:
       SetState(error_caused_teardown_ ? kError : kStopped);
       FinishDestroyingFiltersTask();
       break;
     case kPausing:
       SetState(kFlushing);
-      pipeline_filter_->Flush(
-          base::Bind(&Pipeline::OnTeardownStateTransition, this));
+      DoFlush(base::Bind(&Pipeline::OnTeardownStateTransition, this));
       break;
     case kFlushing:
       SetState(kStopping);
       DoStop(base::Bind(&Pipeline::OnTeardownStateTransition, this));
       break;
 
     case kCreated:
     case kError:
     case kInitDemuxer:
     case kInitAudioDecoder:
@@ skipping 127 matching lines @@
 }
 
 bool Pipeline::InitializeAudioRenderer(
     const scoped_refptr<AudioDecoder>& decoder) {
   DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(IsPipelineOk());
 
   if (!decoder)
     return false;
 
-  filter_collection_->SelectAudioRenderer(
-      &pipeline_init_state_->audio_renderer);
-  if (!pipeline_init_state_->audio_renderer) {
+  filter_collection_->SelectAudioRenderer(&audio_renderer_);
+  if (!audio_renderer_) {
     SetError(PIPELINE_ERROR_REQUIRED_FILTER_MISSING);
     return false;
   }
 
-  pipeline_init_state_->composite->AddFilter(
-      pipeline_init_state_->audio_renderer);
-
-  pipeline_init_state_->audio_renderer->Initialize(
+  audio_renderer_->Initialize(
       decoder,
       base::Bind(&Pipeline::OnFilterInitialize, this),
       base::Bind(&Pipeline::OnAudioUnderflow, this),
-      base::Bind(&Pipeline::OnAudioTimeUpdate, this));
-
-  audio_renderer_ = pipeline_init_state_->audio_renderer;
+      base::Bind(&Pipeline::OnAudioTimeUpdate, this),
+      base::Bind(&Pipeline::NotifyEnded, this),
+      base::Bind(&Pipeline::OnAudioDisabled, this),
+      base::Bind(&Pipeline::SetError, this));
   return true;
 }
 
 bool Pipeline::InitializeVideoRenderer(
     const scoped_refptr<VideoDecoder>& decoder) {
   DCHECK(message_loop_->BelongsToCurrentThread());
   DCHECK(IsPipelineOk());
 
   if (!decoder)
     return false;
@@ skipping 65 matching lines @@
       if (seek_pending_) {
         seek_pending_ = false;
         FinishInitialization();
       }
 
       break;
 
     case kStarted:
     case kEnded:
       SetState(kPausing);
-      pipeline_filter_->Pause(
-          base::Bind(&Pipeline::OnTeardownStateTransition, this));
+      DoPause(base::Bind(&Pipeline::OnTeardownStateTransition, this));
       break;
 
     case kStopping:
     case kStopped:
       NOTREACHED() << "Unexpected state for teardown: " << state_;
       break;
     // default: intentionally left out to force new states to cause compiler
     // errors.
   };
 }
 
-void Pipeline::DoStop(const base::Closure& callback) {
-  if (demuxer_) {
-    demuxer_->Stop(base::Bind(
-        &Pipeline::OnDemuxerStopDone, this, callback));
-    return;
-  }
+void Pipeline::DoSeek(base::TimeDelta seek_timestamp,
+                      bool skip_demuxer_seek,
+                      const PipelineStatusCB& done_cb) {
+  DCHECK(message_loop_->BelongsToCurrentThread());
+  scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs(
+      new std::queue<PipelineStatusCBFunc>());
 
-  OnDemuxerStopDone(callback);
-}
+  if (!skip_demuxer_seek)
+    status_cbs->push(base::Bind(&Demuxer::Seek, demuxer_, seek_timestamp));
 
-void Pipeline::OnDemuxerStopDone(const base::Closure& callback) {
-  if (!message_loop_->BelongsToCurrentThread()) {
-    message_loop_->PostTask(FROM_HERE, base::Bind(
-        &Pipeline::OnDemuxerStopDone, this, callback));
-    return;
-  }
+  if (audio_renderer_)
+    status_cbs->push(base::Bind(
+        &AudioRenderer::Seek, audio_renderer_, seek_timestamp));
 
-  if (pipeline_filter_) {
-    pipeline_filter_->Stop(callback);
-    return;
-  }
+  if (pipeline_filter_)
+    status_cbs->push(base::Bind(
+        &Filter::Seek, pipeline_filter_, seek_timestamp));
 
-  callback.Run();
-}
-
-void Pipeline::DoSeek(TimeDelta seek_timestamp) {
-  demuxer_->Seek(seek_timestamp, base::Bind(
-      &Pipeline::OnDemuxerSeekDone, this, seek_timestamp));
-}
-
-void Pipeline::OnDemuxerSeekDone(TimeDelta seek_timestamp,
-                                 PipelineStatus status) {
-  if (!message_loop_->BelongsToCurrentThread()) {
-    message_loop_->PostTask(FROM_HERE, base::Bind(
-        &Pipeline::OnDemuxerSeekDone, this, seek_timestamp, status));
-    return;
-  }
-
-  PipelineStatusCB done_cb =
-      base::Bind(&Pipeline::OnFilterStateTransitionWithStatus, this);
-
-  if (status == PIPELINE_OK && pipeline_filter_) {
-    pipeline_filter_->Seek(seek_timestamp, done_cb);
-    return;
-  }
-
-  ReportStatus(done_cb, status);
+  RunInSeriesWithStatus(status_cbs.Pass(), base::Bind(
+      &Pipeline::ReportStatus, this, done_cb));
 }
 
 void Pipeline::OnAudioUnderflow() {
   if (!message_loop_->BelongsToCurrentThread()) {
     message_loop_->PostTask(FROM_HERE, base::Bind(
         &Pipeline::OnAudioUnderflow, this));
     return;
   }
 
   if (state_ != kStarted)
     return;
 
   if (audio_renderer_)
     audio_renderer_->ResumeAfterUnderflow(true);
 }
 
 void Pipeline::StartClockIfWaitingForTimeUpdate_Locked() {
   lock_.AssertAcquired();
   if (!waiting_for_clock_update_)
     return;
 
   waiting_for_clock_update_ = false;
   clock_->Play();
 }
 
 }  // namespace media