PipelineError is dead. Long live PipelineStatus!

media/base/pipeline_impl.cc

 // Copyright (c) 2011 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.
 //
 // TODO(scherkus): clean up PipelineImpl... too many crazy function names,
 // potential deadlocks, etc...
 
 #include "base/callback.h"
 #include "base/compiler_specific.h"
 #include "base/stl_util-inl.h"
 #include "base/synchronization/condition_variable.h"
 #include "media/base/clock.h"
 #include "media/base/filter_collection.h"
 #include "media/base/media_format.h"
 #include "media/base/pipeline_impl.h"
 
 namespace media {
 
+PipelineStatusNotification::PipelineStatusNotification()
+    : cv_(&lock_), status_(PIPELINE_OK), notified_(false) {
+  callback_.reset(::NewCallback(this, &PipelineStatusNotification::Notify));

[acolwell GONE FROM CHROMIUM, 2011/03/15 03:56:28]

Is :: necessary here?

[Ami GONE FROM CHROMIUM, 2011/03/15 17:37:18]

Not anymore (it used to be, back when I had PSN::Callback() named
PSN::NewCallback()).  Gone.

+}
+
+PipelineStatusNotification::~PipelineStatusNotification() {
+  DCHECK(notified_);
+}
+
+media::PipelineStatusCallback* PipelineStatusNotification::Callback() {
+  return callback_.release();
+}
+
+void PipelineStatusNotification::Notify(media::PipelineStatus status) {
+  base::AutoLock auto_lock(lock_);
+  DCHECK(!notified_);
+  notified_ = true;
+  status_ = status;
+  cv_.Signal();
+}
+
+void PipelineStatusNotification::Wait() {
+  base::AutoLock auto_lock(lock_);
+  while (!notified_)
+    cv_.Wait();
+}
+
+media::PipelineStatus PipelineStatusNotification::status() {
+  base::AutoLock auto_lock(lock_);
+  DCHECK(notified_);
+  return status_;
+}
+
 class PipelineImpl::PipelineInitState {
  public:
   scoped_refptr<Demuxer> demuxer_;
   scoped_refptr<AudioDecoder> audio_decoder_;
   scoped_refptr<VideoDecoder> video_decoder_;
   scoped_refptr<CompositeFilter> composite_;
 };
 
 PipelineImpl::PipelineImpl(MessageLoop* message_loop)
     : message_loop_(message_loop),
       clock_(new Clock(&base::Time::Now)),
       waiting_for_clock_update_(false),
       state_(kCreated),
       current_bytes_(0) {
   ResetState();
 }
 
 PipelineImpl::~PipelineImpl() {
   base::AutoLock auto_lock(lock_);
   DCHECK(!running_) << "Stop() must complete before destroying object";
   DCHECK(!stop_pending_);
   DCHECK(!seek_pending_);
 }
 
-void PipelineImpl::Init(PipelineCallback* ended_callback,
-                        PipelineCallback* error_callback,
-                        PipelineCallback* network_callback) {
+void PipelineImpl::Init(PipelineStatusCallback* ended_callback,
+                        PipelineStatusCallback* error_callback,
+                        PipelineStatusCallback* network_callback) {
   DCHECK(!IsRunning())
       << "Init() should be called before the pipeline has started";
   ended_callback_.reset(ended_callback);
   error_callback_.reset(error_callback);
   network_callback_.reset(network_callback);
 }
 
 // Creates the PipelineInternal and calls it's start method.
 bool PipelineImpl::Start(FilterCollection* collection,
                          const std::string& url,
-                         PipelineCallback* start_callback) {
+                         PipelineStatusCallback* start_callback) {
   base::AutoLock auto_lock(lock_);
-  scoped_ptr<PipelineCallback> callback(start_callback);
+  scoped_ptr<PipelineStatusCallback> callback(start_callback);
   scoped_ptr<FilterCollection> filter_collection(collection);
 
   if (running_) {
     VLOG(1) << "Media pipeline is already running";
     return false;
   }
 
   if (collection->IsEmpty()) {
     return false;
   }
 
   // Kick off initialization!
   running_ = true;
   message_loop_->PostTask(
       FROM_HERE,
       NewRunnableMethod(this,
                         &PipelineImpl::StartTask,
                         filter_collection.release(),
                         url,
                         callback.release()));
   return true;
 }
 
-void PipelineImpl::Stop(PipelineCallback* stop_callback) {
+void PipelineImpl::Stop(PipelineStatusCallback* stop_callback) {
   base::AutoLock auto_lock(lock_);
-  scoped_ptr<PipelineCallback> callback(stop_callback);
+  scoped_ptr<PipelineStatusCallback> callback(stop_callback);
   if (!running_) {
     VLOG(1) << "Media pipeline has already stopped";
     return;
   }
 
   // Stop the pipeline, which will set |running_| to false on behalf.
   message_loop_->PostTask(FROM_HERE,
       NewRunnableMethod(this, &PipelineImpl::StopTask, callback.release()));
 }
 
 void PipelineImpl::Seek(base::TimeDelta time,
-                        PipelineCallback* seek_callback) {
+                        PipelineStatusCallback* seek_callback) {
   base::AutoLock auto_lock(lock_);
-  scoped_ptr<PipelineCallback> callback(seek_callback);
+  scoped_ptr<PipelineStatusCallback> callback(seek_callback);
   if (!running_) {
     VLOG(1) << "Media pipeline must be running";
     return;
   }
 
   message_loop_->PostTask(FROM_HERE,
       NewRunnableMethod(this, &PipelineImpl::SeekTask, time,
                         callback.release()));
 }
 
@@ skipping 155 matching lines @@
 bool PipelineImpl::IsStreaming() const {
   base::AutoLock auto_lock(lock_);
   return streaming_;
 }
 
 bool PipelineImpl::IsLoaded() const {
   base::AutoLock auto_lock(lock_);
   return loaded_;
 }
 
-PipelineError PipelineImpl::GetError() const {
-  base::AutoLock auto_lock(lock_);
-  return error_;
-}
-
 PipelineStatistics PipelineImpl::GetStatistics() const {
   base::AutoLock auto_lock(lock_);
   return statistics_;
 }
 
 void PipelineImpl::SetClockForTesting(Clock* clock) {
   clock_.reset(clock);
 }
 
 void PipelineImpl::SetCurrentReadPosition(int64 offset) {
@@ skipping 26 matching lines @@
   duration_         = kZero;
   buffered_time_    = kZero;
   buffered_bytes_   = 0;
   streaming_        = false;
   loaded_           = false;
   total_bytes_      = 0;
   video_width_      = 0;
   video_height_     = 0;
   volume_           = 1.0f;
   playback_rate_    = 0.0f;
-  error_            = PIPELINE_OK;
+  status_           = PIPELINE_OK;
   has_audio_        = false;
   has_video_        = false;
   waiting_for_clock_update_ = false;
   audio_disabled_   = false;
   clock_->SetTime(kZero);
 }
 
 void PipelineImpl::set_state(State next_state) {
   state_ = next_state;
 }
 
 bool PipelineImpl::IsPipelineOk() {
-  return PIPELINE_OK == GetError();
+  base::AutoLock auto_lock(lock_);
+  return status_ == PIPELINE_OK;
 }
 
 bool PipelineImpl::IsPipelineStopped() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   return state_ == kStopped || state_ == kError;
 }
 
 bool PipelineImpl::IsPipelineTearingDown() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   return tearing_down_;
@@ skipping 12 matching lines @@
          kFlushing == state_ || kStarting == state_)
       << "Current state : " << state_;
   return true;
 }
 
 void PipelineImpl::FinishInitialization() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   // Execute the seek callback, if present.  Note that this might be the
   // initial callback passed into Start().
   if (seek_callback_.get()) {
-    seek_callback_->Run();
+    seek_callback_->Run(status_);
     seek_callback_.reset();
   }
 }
 
 // static
 bool PipelineImpl::TransientState(State state) {
   return state == kPausing ||
          state == kFlushing ||
          state == kSeeking ||
          state == kStarting ||
@@ skipping 16 matching lines @@
     return kStarting;
   } else if (current == kStarting) {
     return kStarted;
   } else if (current == kStopping) {
     return error_caused_teardown_ ? kError : kStopped;
   } else {
     return current;
   }
 }
 
-void PipelineImpl::SetError(PipelineError error) {
+void PipelineImpl::SetError(PipelineStatus error) {
   DCHECK(IsRunning());
   DCHECK(error != PIPELINE_OK) << "PIPELINE_OK isn't an error!";
   VLOG(1) << "Media pipeline error: " << error;
 
   message_loop_->PostTask(FROM_HERE,
      NewRunnableMethod(this, &PipelineImpl::ErrorChangedTask, error));
 }
 
 base::TimeDelta PipelineImpl::GetTime() const {
   DCHECK(IsRunning());
@@ skipping 115 matching lines @@
 void PipelineImpl::OnUpdateStatistics(const PipelineStatistics& stats) {
   base::AutoLock auto_lock(lock_);
   statistics_.audio_bytes_decoded += stats.audio_bytes_decoded;
   statistics_.video_bytes_decoded += stats.video_bytes_decoded;
   statistics_.video_frames_decoded += stats.video_frames_decoded;
   statistics_.video_frames_dropped += stats.video_frames_dropped;
 }
 
 void PipelineImpl::StartTask(FilterCollection* filter_collection,
                              const std::string& url,
-                             PipelineCallback* start_callback) {
+                             PipelineStatusCallback* start_callback) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   DCHECK_EQ(kCreated, state_);
   filter_collection_.reset(filter_collection);
   url_ = url;
   seek_callback_.reset(start_callback);
 
   // Kick off initialization.
   set_state(kInitDemuxer);
   pipeline_init_state_.reset(new PipelineInitState());
   pipeline_init_state_->composite_ = new CompositeFilter(message_loop_);
@@ skipping 16 matching lines @@
 // FilterHost's InitializationComplete() method, the pipeline will update its
 // state to kStarted and |init_callback_|, will be executed.
 //
 // TODO(hclam): InitializeTask() is now starting the pipeline asynchronously. It
 // works like a big state change table. If we no longer need to start filters
 // in order, we need to get rid of all the state change.
 void PipelineImpl::InitializeTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
   // If we have received the stop or error signal, return immediately.
-  if (IsPipelineStopPending() ||
-      IsPipelineStopped() ||
-      PIPELINE_OK != GetError()) {
+  if (IsPipelineStopPending() || IsPipelineStopped() || !IsPipelineOk())
     return;
-  }
 
   DCHECK(state_ == kInitDemuxer ||
          state_ == kInitAudioDecoder ||
          state_ == kInitAudioRenderer ||
          state_ == kInitVideoDecoder ||
          state_ == kInitVideoRenderer);
 
 
   // Demuxer created, create audio decoder.
   if (state_ == kInitDemuxer) {
@@ skipping 66 matching lines @@
   }
 }
 
 // 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.
-void PipelineImpl::StopTask(PipelineCallback* stop_callback) {
+void PipelineImpl::StopTask(PipelineStatusCallback* stop_callback) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   DCHECK(!IsPipelineStopPending());
   DCHECK_NE(state_, kStopped);
 
   if (state_ == kStopped) {
     // Already stopped so just run callback.
-    stop_callback->Run();
+    stop_callback->Run(status_);
     delete stop_callback;
     return;
   }
 
   if (IsPipelineTearingDown() && error_caused_teardown_) {
     // If we are stopping due to SetError(), stop normally instead of
     // going to error state and calling |error_callback_|. This converts
     // the teardown in progress from an error teardown into one that acts
     // like the error never occurred.
     base::AutoLock auto_lock(lock_);
-    error_ = PIPELINE_OK;
+    status_ = PIPELINE_OK;
     error_caused_teardown_ = false;
   }
 
   stop_callback_.reset(stop_callback);
 
   stop_pending_ = true;
   if (!IsPipelineSeeking() && !IsPipelineTearingDown()) {
     // We will tear down pipeline immediately when there is no seek operation
     // pending and no teardown in progress. This should include the case where
     // we are partially initialized.
     TearDownPipeline();
   }
 }
 
-void PipelineImpl::ErrorChangedTask(PipelineError error) {
+void PipelineImpl::ErrorChangedTask(PipelineStatus error) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   DCHECK_NE(PIPELINE_OK, error) << "PIPELINE_OK isn't an error!";
 
   // Suppress executing additional error logic. Note that if we are currently
   // performing a normal stop, then we return immediately and continue the
   // normal stop.
   if (IsPipelineStopped() || IsPipelineTearingDown()) {
     return;
   }
 
   base::AutoLock auto_lock(lock_);
-  error_ = error;
+  status_ = error;
 
   error_caused_teardown_ = true;
 
   // Posting TearDownPipeline() to message loop so that we can make sure
   // it runs after any pending callbacks that are already queued.
   // |tearing_down_| is set early here to make sure that pending callbacks
   // don't modify the state before TeadDownPipeline() can run.
   tearing_down_ = true;
   message_loop_->PostTask(FROM_HERE,
       NewRunnableMethod(this, &PipelineImpl::TearDownPipeline));
@@ skipping 16 matching lines @@
 
 void PipelineImpl::VolumeChangedTask(float volume) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
   if (audio_renderer_) {
     audio_renderer_->SetVolume(volume);
   }
 }
 
 void PipelineImpl::SeekTask(base::TimeDelta time,
-                            PipelineCallback* seek_callback) {
+                            PipelineStatusCallback* seek_callback) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   DCHECK(!IsPipelineStopPending());
 
   // Suppress seeking if we're not fully started.
   if (state_ != kStarted && state_ != kEnded) {
     // TODO(scherkus): should we run the callback?  I'm tempted to say the API
     // will only execute the first Seek() request.
     VLOG(1) << "Media pipeline has not started, ignoring seek to "
               << time.InMicroseconds();
     delete seek_callback;
@@ skipping 49 matching lines @@
     }
   }
 
   if (video_renderer_ && !video_renderer_->HasEnded()) {
     return;
   }
 
   // Transition to ended, executing the callback if present.
   set_state(kEnded);
   if (ended_callback_.get()) {
-    ended_callback_->Run();
+    ended_callback_->Run(status_);
   }
 }
 
 void PipelineImpl::NotifyNetworkEventTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   if (network_callback_.get()) {
-    network_callback_->Run();
+    network_callback_->Run(status_);
   }
 }
 
 void PipelineImpl::DisableAudioRendererTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
   base::AutoLock auto_lock(lock_);
   has_audio_ = false;
   audio_disabled_ = true;
 
@@ skipping 116 matching lines @@
 void PipelineImpl::FinishDestroyingFiltersTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   DCHECK(IsPipelineStopped());
 
   // Clear renderer references.
   audio_renderer_ = NULL;
   video_renderer_ = NULL;
 
   pipeline_filter_ = NULL;
 
-  if (error_caused_teardown_ && GetError() != PIPELINE_OK &&
-      error_callback_.get()) {
-    error_callback_->Run();
-  }
+  if (error_caused_teardown_ && !IsPipelineOk() && error_callback_.get())
+    error_callback_->Run(status_);
 
   if (stop_pending_) {
     stop_pending_ = false;
     ResetState();
-    scoped_ptr<PipelineCallback> stop_callback(stop_callback_.release());
+    scoped_ptr<PipelineStatusCallback> stop_callback(stop_callback_.release());
     // Notify the client that stopping has finished.
     if (stop_callback.get()) {
-      stop_callback->Run();
+      stop_callback->Run(status_);
     }
   }
 
   tearing_down_ = false;
   error_caused_teardown_ = false;
 }
 
 bool PipelineImpl::PrepareFilter(scoped_refptr<Filter> filter) {
   bool ret = pipeline_init_state_->composite_->AddFilter(filter.get());
 
   if (!ret) {
     SetError(PIPELINE_ERROR_INITIALIZATION_FAILED);
   }
   return ret;
 }
 
 void PipelineImpl::InitializeDemuxer() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   DCHECK(IsPipelineOk());
 
   filter_collection_->GetDemuxerFactory()->Build(url_,
       NewCallback(this, &PipelineImpl::OnDemuxerBuilt));
 }
 
-void PipelineImpl::OnDemuxerBuilt(PipelineError error,
-                                  Demuxer* demuxer) {
+void PipelineImpl::OnDemuxerBuilt(PipelineStatus status, Demuxer* demuxer) {
   if (MessageLoop::current() != message_loop_) {
     message_loop_->PostTask(FROM_HERE,
                             NewRunnableMethod(this,
                                               &PipelineImpl::OnDemuxerBuilt,
-                                              error,
+                                              status,
                                               make_scoped_refptr(demuxer)));
     return;
   }
 
-  if (error != PIPELINE_OK) {
-    SetError(error);
+  if (status != PIPELINE_OK) {
+    SetError(status);
     return;
   }
 
   if (!demuxer) {
     SetError(PIPELINE_ERROR_REQUIRED_FILTER_MISSING);
     return;
   }
 
   if (!PrepareFilter(demuxer))
     return;
@@ skipping 186 matching lines @@
     case kStopping:
     case kStopped:
       NOTREACHED() << "Unexpected state for teardown: " << state_;
       break;
     // default: intentionally left out to force new states to cause compiler
     // errors.
   };
 }
 
 }  // namespace media