Convert Filter::Seek() to use new callback system.

media/base/composite_filter.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.
 
 #include "media/base/composite_filter.h"
 
+#include "base/bind.h"
 #include "base/message_loop.h"
 #include "base/stl_util-inl.h"
 #include "media/base/callback.h"
 
 namespace media {
 
 class CompositeFilter::FilterHostImpl : public FilterHost {
  public:
   FilterHostImpl(CompositeFilter* parent, FilterHost* host);
 
@@ skipping 59 matching lines @@
   host_impl_.reset(new FilterHostImpl(this, host));
 }
 
 FilterHost* CompositeFilter::host() {
   return host_impl_.get() ? host_impl_->host() : NULL;
 }
 
 void CompositeFilter::Play(FilterCallback* play_callback) {
   DCHECK_EQ(message_loop_, MessageLoop::current());
   scoped_ptr<FilterCallback> callback(play_callback);
-  if (callback_.get()) {
+  if (IsOperationPending()) {
     SendErrorToHost(PIPELINE_ERROR_OPERATION_PENDING);
     callback->Run();
     return;
   } else if (state_ == kPlaying) {
     callback->Run();
     return;
   } else if (!host() || (state_ != kPaused && state_ != kCreated)) {
     SendErrorToHost(PIPELINE_ERROR_INVALID_STATE);
     callback->Run();
     return;
   }
 
   ChangeState(kPlayPending);
   callback_.reset(callback.release());
   StartSerialCallSequence();
 }
 
 void CompositeFilter::Pause(FilterCallback* pause_callback) {
   DCHECK_EQ(message_loop_, MessageLoop::current());
   scoped_ptr<FilterCallback> callback(pause_callback);
-  if (callback_.get()) {
+  if (IsOperationPending()) {
     SendErrorToHost(PIPELINE_ERROR_OPERATION_PENDING);
     callback->Run();
     return;
   } else if (state_ == kPaused) {
     callback->Run();
     return;
   } else if (!host() || state_ != kPlaying) {
     SendErrorToHost(PIPELINE_ERROR_INVALID_STATE);
     callback->Run();
     return;
   }
 
   ChangeState(kPausePending);
   callback_.reset(callback.release());
   StartSerialCallSequence();
 }
 
 void CompositeFilter::Flush(FilterCallback* flush_callback) {
   DCHECK_EQ(message_loop_, MessageLoop::current());
   scoped_ptr<FilterCallback> callback(flush_callback);
-  if (callback_.get()) {
+  if (IsOperationPending()) {
     SendErrorToHost(PIPELINE_ERROR_OPERATION_PENDING);
     callback->Run();
     return;
   } else if (!host() || (state_ != kCreated && state_ != kPaused)) {
     SendErrorToHost(PIPELINE_ERROR_INVALID_STATE);
     callback->Run();
     return;
   }
 
   ChangeState(kFlushPending);
@@ skipping 47 matching lines @@
 void CompositeFilter::SetPlaybackRate(float playback_rate) {
   DCHECK_EQ(message_loop_, MessageLoop::current());
   for (FilterVector::iterator iter = filters_.begin();
        iter != filters_.end();
        ++iter) {
     (*iter)->SetPlaybackRate(playback_rate);
   }
 }
 
 void CompositeFilter::Seek(base::TimeDelta time,
-                           FilterCallback* seek_callback) {
+                           const FilterStatusCB& seek_cb) {
   DCHECK_EQ(message_loop_, MessageLoop::current());
-  scoped_ptr<FilterCallback> callback(seek_callback);
-  if (callback_.get()) {
-    SendErrorToHost(PIPELINE_ERROR_OPERATION_PENDING);
-    callback->Run();
+
+  if (IsOperationPending()) {
+    seek_cb.Run(PIPELINE_ERROR_OPERATION_PENDING);
     return;
   } else if (!host() || (state_ != kPaused && state_ != kCreated)) {
-    SendErrorToHost(PIPELINE_ERROR_INVALID_STATE);
-    callback->Run();
+    seek_cb.Run(PIPELINE_ERROR_INVALID_STATE);
     return;
   }
 
   ChangeState(kSeekPending);
-  callback_.reset(callback.release());
+  status_cb_ = seek_cb;
   pending_seek_time_ = time;
   StartSerialCallSequence();
 }
 
 void CompositeFilter::OnAudioRendererDisabled() {
   DCHECK_EQ(message_loop_, MessageLoop::current());
   for (FilterVector::iterator iter = filters_.begin();
        iter != filters_.end();
        ++iter) {
     (*iter)->OnAudioRendererDisabled();
   }
 }
 
 void CompositeFilter::ChangeState(State new_state) {
   DCHECK_EQ(message_loop_, MessageLoop::current());
   state_ = new_state;
 }
 
 void CompositeFilter::StartSerialCallSequence() {
   DCHECK_EQ(message_loop_, MessageLoop::current());
   status_ = PIPELINE_OK;
+  sequence_index_ = 0;
 
   if (!filters_.empty()) {
-    sequence_index_ = 0;
     CallFilter(filters_[sequence_index_],
                NewThreadSafeCallback(&CompositeFilter::SerialCallback));
   } else {
-    sequence_index_ = 0;
     SerialCallback();
   }
 }
 
 void CompositeFilter::StartParallelCallSequence() {
   DCHECK_EQ(message_loop_, MessageLoop::current());
   status_ = PIPELINE_OK;
+  sequence_index_ = 0;
 
   if (!filters_.empty()) {
-    sequence_index_ = 0;
     for (size_t i = 0; i < filters_.size(); i++) {
       CallFilter(filters_[i],
                  NewThreadSafeCallback(&CompositeFilter::ParallelCallback));
     }
   } else {
-    sequence_index_ = 0;
     ParallelCallback();
   }
 }
 
 void CompositeFilter::CallFilter(scoped_refptr<Filter>& filter,
                                  FilterCallback* callback) {
   switch (state_) {
     case kPlayPending:
       filter->Play(callback);
       break;
     case kPausePending:
       filter->Pause(callback);
       break;
     case kFlushPending:
       filter->Flush(callback);
       break;
     case kStopPending:
       filter->Stop(callback);
       break;
     case kSeekPending:
-      filter->Seek(pending_seek_time_, callback);
+      filter->Seek(pending_seek_time_,
+                   base::Bind(&CompositeFilter::OnStatusCB, this, callback));
       break;
     default:
       delete callback;
       ChangeState(kError);
-      HandleError(PIPELINE_ERROR_INVALID_STATE);
+      DispatchPendingCallback(PIPELINE_ERROR_INVALID_STATE);
   }
 }
 
-void CompositeFilter::DispatchPendingCallback() {
+void CompositeFilter::DispatchPendingCallback(PipelineStatus status) {
+  DCHECK((status_cb_.is_null() && callback_.get()) ||
+         (!status_cb_.is_null() && !callback_.get()));
+
+  if (!status_cb_.is_null()) {
+    RunAndResetCB(&status_cb_, status);
+    return;
+  }
+
   if (callback_.get()) {
+    if (status != PIPELINE_OK)
+      SendErrorToHost(status);
+
     scoped_ptr<FilterCallback> callback(callback_.release());
     callback->Run();
   }
 }
 
 CompositeFilter::State CompositeFilter::GetNextState(State state) const {
   State ret = kInvalid;
   switch (state) {
     case kPlayPending:
       ret = kPlaying;
@@ skipping 28 matching lines @@
   }
 
   return ret;
 }
 
 void CompositeFilter::SerialCallback() {
   DCHECK_EQ(message_loop_, MessageLoop::current());
   if (status_ != PIPELINE_OK) {
     // We encountered an error. Terminate the sequence now.
     ChangeState(kError);
-    HandleError(status_);
+    DispatchPendingCallback(status_);
     return;
   }
 
   if (!filters_.empty())
     sequence_index_++;
 
   if (sequence_index_ == filters_.size()) {
     // All filters have been successfully called without error.
     OnCallSequenceDone();
   } else if (GetNextState(state_) == kStopPending) {
@@ skipping 10 matching lines @@
 void CompositeFilter::ParallelCallback() {
   DCHECK_EQ(message_loop_, MessageLoop::current());
 
   if (!filters_.empty())
     sequence_index_++;
 
   if (sequence_index_ == filters_.size()) {
     if (status_ != PIPELINE_OK) {
       // We encountered an error.
       ChangeState(kError);
-      HandleError(status_);
+      DispatchPendingCallback(status_);
       return;
     }
 
     OnCallSequenceDone();
   }
 }
 
 void CompositeFilter::OnCallSequenceDone() {
   State next_state = GetNextState(state_);
 
   if (next_state == kInvalid) {
     // We somehow got into an unexpected state.
     ChangeState(kError);
-    HandleError(PIPELINE_ERROR_INVALID_STATE);
+    DispatchPendingCallback(PIPELINE_ERROR_INVALID_STATE);
+    return;
   }
 
   ChangeState(next_state);
 
   if (state_ == kStopPending) {
     // Handle a deferred Stop().
     StartSerialCallSequence();
   } else {
     // Call the callback to indicate that the operation has completed.
-    DispatchPendingCallback();
+    DispatchPendingCallback(PIPELINE_OK);
   }
 }
 
 void CompositeFilter::SendErrorToHost(PipelineStatus error) {
   if (host_impl_.get())
     host_impl_.get()->host()->SetError(error);
 }
 
-void CompositeFilter::HandleError(PipelineStatus error) {
-  DCHECK_NE(error, PIPELINE_OK);
-  SendErrorToHost(error);
-  DispatchPendingCallback();
-}
-
 FilterCallback* CompositeFilter::NewThreadSafeCallback(
     void (CompositeFilter::*method)()) {
   return TaskToCallbackAdapter::NewCallback(
       NewRunnableFunction(&CompositeFilter::OnCallback,
                           message_loop_,
                           runnable_factory_->NewRunnableMethod(method)));
 }
 
 // This method is intentionally static so that no reference to the composite
 // is needed to call it. This method may be called by other threads and we
@@ skipping 11 matching lines @@
   }
 
   task->Run();
   delete task;
 }
 
 bool CompositeFilter::CanForwardError() {
   return (state_ == kCreated) || (state_ == kPlaying) || (state_ == kPaused);
 }
 
+bool CompositeFilter::IsOperationPending() const {
+  DCHECK(!(callback_.get() && !status_cb_.is_null()));
+
+  return callback_.get() || !status_cb_.is_null();
+}
+
+void CompositeFilter::OnStatusCB(FilterCallback* callback,
+                                 PipelineStatus status) {
+  if (status != PIPELINE_OK)
+    SetError(status);
+
+  callback->Run();
+  delete callback;
+}
+
 void CompositeFilter::SetError(PipelineStatus error) {
   // TODO(acolwell): Temporary hack to handle errors that occur
   // during filter initialization. In this case we just forward
   // the error to the host even if it is on the wrong thread. We
   // have to do this because if we defer the call, we can't be
   // sure the host will get the error before the "init done" callback
   // is executed. This will be cleaned up when filter init is refactored.
   if (state_ == kCreated) {
     SendErrorToHost(error);
     return;
@@ skipping 90 matching lines @@
 
 void CompositeFilter::FilterHostImpl::SetCurrentReadPosition(int64 offset) {
   host_->SetCurrentReadPosition(offset);
 }
 
 int64 CompositeFilter::FilterHostImpl::GetCurrentReadPosition() {
   return host_->GetCurrentReadPosition();
 }
 
 }  // namespace media