Revert "Splitting media filter's Initialize() into Create() + callback and Seek() + callback."

Review URL: http://codereview.chromium.org/149682


git-svn-id: svn://svn.chromium.org/chrome/trunk/src@20745 0039d316-1c4b-4281-b951-d872f2087c98

9 files changed, 96 insertions, 187 deletions

diff --git a/media/base/filter_host.h b/media/base/filter_host.h
index 982b3e5..4907df2a 100644
--- a/media/base/filter_host.h
+++ b/media/base/filter_host.h
@@ -24,6 +24,11 @@ namespace media {
 
 class FilterHost {
  public:
+  // Filters must call this method to indicate that their initialization is
+  // complete.  They may call this from within their Initialize() method or may
+  // choose call it after processing some data.
+  virtual void InitializationComplete() = 0;
+
   // Stops execution of the pipeline due to a fatal error.  Do not call this
   // method with PIPELINE_OK or PIPELINE_STOPPING (used internally by pipeline).
   virtual void Error(PipelineError error) = 0;
diff --git a/media/base/filter_host_impl.cc b/media/base/filter_host_impl.cc
index 4192afd..09f1791 100644
--- a/media/base/filter_host_impl.cc
+++ b/media/base/filter_host_impl.cc
@@ -6,6 +6,10 @@
 
 namespace media {
 
+void FilterHostImpl::InitializationComplete() {
+  pipeline_internal_->InitializationComplete(this);
+}
+
 void FilterHostImpl::Error(PipelineError error) {
   pipeline_internal_->Error(error);
 }
diff --git a/media/base/filter_host_impl.h b/media/base/filter_host_impl.h
index cb33a48..71e2417 100644
--- a/media/base/filter_host_impl.h
+++ b/media/base/filter_host_impl.h
@@ -16,6 +16,7 @@ namespace media {
 class FilterHostImpl : public FilterHost {
  public:
   // FilterHost interface.
+  virtual void InitializationComplete();
   virtual void Error(PipelineError error);
   virtual base::TimeDelta GetTime() const;
   virtual void SetTime(base::TimeDelta time);
diff --git a/media/base/filters.h b/media/base/filters.h
index 84df0966..b892961 100644
--- a/media/base/filters.h
+++ b/media/base/filters.h
@@ -53,8 +53,6 @@ enum FilterType {
   FILTER_VIDEO_RENDERER
 };
 
-// Used for completing asynchronous methods.
-typedef Callback0::Type FilterCallback;
 
 class MediaFilter : public base::RefCountedThreadSafe<MediaFilter> {
  public:
@@ -96,14 +94,9 @@ class MediaFilter : public base::RefCountedThreadSafe<MediaFilter> {
   // method if they need to respond to this call.
   virtual void SetPlaybackRate(float playback_rate) {}
 
-  // Carry out any actions required to seek to the given time, executing the
-  // callback upon completion.
-  virtual void Seek(base::TimeDelta time, FilterCallback* callback) {
-    scoped_ptr<FilterCallback> seek_callback(callback);
-    if (seek_callback.get()) {
-      seek_callback->Run();
-    }
-  }
+  // The pipeline is seeking to the specified time.  Filters may implement
+  // this method if they need to respond to this call.
+  virtual void Seek(base::TimeDelta time) {}
 
  protected:
   // Only allow scoped_refptr<> to delete filters.
@@ -135,9 +128,8 @@ class DataSource : public MediaFilter {
 
   static const size_t kReadError = static_cast<size_t>(-1);
 
-  // Initialize a DataSource for the given URL, executing the callback upon
-  // completion.
-  virtual void Initialize(const std::string& url, FilterCallback* callback) = 0;
+  // Initializes this filter, returns true if successful, false otherwise.
+  virtual bool Initialize(const std::string& url) = 0;
 
   // Returns the MediaFormat for this filter.
   virtual const MediaFormat& media_format() = 0;
@@ -174,10 +166,8 @@ class Demuxer : public MediaFilter {
             mime_type == mime_type::kApplicationOctetStream);
   }
 
-  // Initialize a Demuxer with the given DataSource, executing the callback upon
-  // completion.
-  virtual void Initialize(DataSource* data_source,
-                          FilterCallback* callback) = 0;
+  // Initializes this filter, returns true if successful, false otherwise.
+  virtual bool Initialize(DataSource* data_source) = 0;
 
   // Returns the number of streams available
   virtual size_t GetNumberOfStreams() = 0;
@@ -233,9 +223,8 @@ class VideoDecoder : public MediaFilter {
     return mime_type::kMajorTypeVideo;
   }
 
-  // Initialize a VideoDecoder with the given DemuxerStream, executing the
-  // callback upon completion.
-  virtual void Initialize(DemuxerStream* stream, FilterCallback* callback) = 0;
+  // Initializes this filter, returns true if successful, false otherwise.
+  virtual bool Initialize(DemuxerStream* demuxer_stream) = 0;
 
   // Returns the MediaFormat for this filter.
   virtual const MediaFormat& media_format() = 0;
@@ -257,9 +246,8 @@ class AudioDecoder : public MediaFilter {
     return mime_type::kMajorTypeAudio;
   }
 
-  // Initialize a AudioDecoder with the given DemuxerStream, executing the
-  // callback upon completion.
-  virtual void Initialize(DemuxerStream* stream, FilterCallback* callback) = 0;
+  // Initializes this filter, returns true if successful, false otherwise.
+  virtual bool Initialize(DemuxerStream* demuxer_stream) = 0;
 
   // Returns the MediaFormat for this filter.
   virtual const MediaFormat& media_format() = 0;
@@ -281,9 +269,8 @@ class VideoRenderer : public MediaFilter {
     return mime_type::kMajorTypeVideo;
   }
 
-  // Initialize a VideoRenderer with the given VideoDecoder, executing the
-  // callback upon completion.
-  virtual void Initialize(VideoDecoder* decoder, FilterCallback* callback) = 0;
+  // Initializes this filter, returns true if successful, false otherwise.
+  virtual bool Initialize(VideoDecoder* decoder) = 0;
 };
 
 
@@ -297,9 +284,8 @@ class AudioRenderer : public MediaFilter {
     return mime_type::kMajorTypeAudio;
   }
 
-  // Initialize a AudioRenderer with the given AudioDecoder, executing the
-  // callback upon completion.
-  virtual void Initialize(AudioDecoder* decoder, FilterCallback* callback) = 0;
+  // Initializes this filter, returns true if successful, false otherwise.
+  virtual bool Initialize(AudioDecoder* decoder) = 0;
 
   // Sets the output volume.
   virtual void SetVolume(float volume) = 0;
diff --git a/media/base/mock_filters.cc b/media/base/mock_filters.cc
deleted file mode 100644
index 7b854fc..0000000
--- a/media/base/mock_filters.cc
+++ /dev/null
@@ -1,18 +0,0 @@
-// Copyright (c) 2009 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/mock_filters.h"
-
-namespace media {
-
-void RunFilterCallback(::testing::Unused, FilterCallback* callback) {
-  callback->Run();
-  delete callback;
-}
-
-void DestroyFilterCallback(::testing::Unused, FilterCallback* callback) {
-  delete callback;
-}
-
-}  // namespace media
diff --git a/media/base/mock_filters.h b/media/base/mock_filters.h
index 6d8c8a9..bffba6f 100644
--- a/media/base/mock_filters.h
+++ b/media/base/mock_filters.h
@@ -40,52 +40,6 @@ class Destroyable : public MockClass {
   DISALLOW_COPY_AND_ASSIGN(Destroyable);
 };
 
-// Helper class used to test that callbacks are executed.  It is recommend you
-// combine this class with StrictMock<> to verify that the callback is executed.
-// You can reuse the same instance of a MockFilterCallback many times since
-// gmock will track the number of times the methods are executed.
-class MockFilterCallback {
- public:
-  MockFilterCallback() {}
-  virtual ~MockFilterCallback() {}
-
-  MOCK_METHOD0(OnCallbackDestroyed, void());
-  MOCK_METHOD0(OnFilterCallback, void());
-
-  // Helper method to create a new callback for this mock.  The callback will
-  // call OnFilterCallback() when executed and OnCallbackDestroyed() when
-  // destroyed.  Clients should use NiceMock<> or StrictMock<> depending on the
-  // test.
-  FilterCallback* NewCallback() {
-    return new CallbackImpl(this);
-  }
-
- private:
-  // Private implementation of CallbackRunner used to trigger expectations on
-  // MockFilterCallback.
-  class CallbackImpl : public CallbackRunner<Tuple0> {
-   public:
-    CallbackImpl(MockFilterCallback* mock_callback)
-        : mock_callback_(mock_callback) {
-    }
-
-    virtual ~CallbackImpl() {
-      mock_callback_->OnCallbackDestroyed();
-    }
-
-    virtual void RunWithParams(const Tuple0& params) {
-      mock_callback_->OnFilterCallback();
-    }
-
-   private:
-    MockFilterCallback* mock_callback_;
-
-    DISALLOW_COPY_AND_ASSIGN(CallbackImpl);
-  };
-
-  DISALLOW_COPY_AND_ASSIGN(MockFilterCallback);
-};
-
 class MockDataSource : public DataSource {
  public:
   MockDataSource() {}
@@ -93,11 +47,10 @@ class MockDataSource : public DataSource {
   // MediaFilter implementation.
   MOCK_METHOD0(Stop, void());
   MOCK_METHOD1(SetPlaybackRate, void(float playback_rate));
-  MOCK_METHOD2(Seek, void(base::TimeDelta time, FilterCallback* callback));
+  MOCK_METHOD1(Seek, void(base::TimeDelta time));
 
   // DataSource implementation.
-  MOCK_METHOD2(Initialize, void(const std::string& url,
-                                FilterCallback* callback));
+  MOCK_METHOD1(Initialize, bool(const std::string& url));
   const MediaFormat& media_format() { return media_format_; }
   MOCK_METHOD2(Read, size_t(uint8* data, size_t size));
   MOCK_METHOD1(GetPosition, bool(int64* position_out));
@@ -121,11 +74,10 @@ class MockDemuxer : public Demuxer {
   // MediaFilter implementation.
   MOCK_METHOD0(Stop, void());
   MOCK_METHOD1(SetPlaybackRate, void(float playback_rate));
-  MOCK_METHOD2(Seek, void(base::TimeDelta time, FilterCallback* callback));
+  MOCK_METHOD1(Seek, void(base::TimeDelta time));
 
   // Demuxer implementation.
-  MOCK_METHOD2(Initialize, void(DataSource* data_source,
-                                FilterCallback* callback));
+  MOCK_METHOD1(Initialize, bool(DataSource* data_source));
   MOCK_METHOD0(GetNumberOfStreams, size_t());
   MOCK_METHOD1(GetStream, scoped_refptr<DemuxerStream>(int stream_id));
 
@@ -164,21 +116,13 @@ class MockVideoDecoder : public VideoDecoder {
  public:
   MockVideoDecoder() {}
 
-  // Sets the essential media format keys for this decoder.
-  MockVideoDecoder(const std::string& mime_type, int width, int height) {
-    media_format_.SetAsString(MediaFormat::kMimeType, mime_type);
-    media_format_.SetAsInteger(MediaFormat::kWidth, width);
-    media_format_.SetAsInteger(MediaFormat::kHeight, height);
-  }
-
   // MediaFilter implementation.
   MOCK_METHOD0(Stop, void());
   MOCK_METHOD1(SetPlaybackRate, void(float playback_rate));
-  MOCK_METHOD2(Seek, void(base::TimeDelta time, FilterCallback* callback));
+  MOCK_METHOD1(Seek, void(base::TimeDelta time));
 
   // VideoDecoder implementation.
-  MOCK_METHOD2(Initialize, void(DemuxerStream* stream,
-                                FilterCallback* callback));
+  MOCK_METHOD1(Initialize, bool(DemuxerStream* demuxer_stream));
   const MediaFormat& media_format() { return media_format_; }
   MOCK_METHOD1(Read, void(Callback1<VideoFrame*>::Type* read_callback));
 
@@ -198,11 +142,10 @@ class MockAudioDecoder : public AudioDecoder {
   // MediaFilter implementation.
   MOCK_METHOD0(Stop, void());
   MOCK_METHOD1(SetPlaybackRate, void(float playback_rate));
-  MOCK_METHOD2(Seek, void(base::TimeDelta time, FilterCallback* callback));
+  MOCK_METHOD1(Seek, void(base::TimeDelta time));
 
   // AudioDecoder implementation.
-  MOCK_METHOD2(Initialize, void(DemuxerStream* stream,
-                                FilterCallback* callback));
+  MOCK_METHOD1(Initialize, bool(DemuxerStream* demuxer_stream));
   const MediaFormat& media_format() { return media_format_; }
   MOCK_METHOD1(Read, void(Callback1<Buffer*>::Type* read_callback));
 
@@ -222,11 +165,10 @@ class MockVideoRenderer : public VideoRenderer {
   // MediaFilter implementation.
   MOCK_METHOD0(Stop, void());
   MOCK_METHOD1(SetPlaybackRate, void(float playback_rate));
-  MOCK_METHOD2(Seek, void(base::TimeDelta time, FilterCallback* callback));
+  MOCK_METHOD1(Seek, void(base::TimeDelta time));
 
   // VideoRenderer implementation.
-  MOCK_METHOD2(Initialize, void(VideoDecoder* decoder,
-                                FilterCallback* callback));
+  MOCK_METHOD1(Initialize, bool(VideoDecoder* decoder));
 
  protected:
   virtual ~MockVideoRenderer() {}
@@ -242,11 +184,10 @@ class MockAudioRenderer : public AudioRenderer {
   // MediaFilter implementation.
   MOCK_METHOD0(Stop, void());
   MOCK_METHOD1(SetPlaybackRate, void(float playback_rate));
-  MOCK_METHOD2(Seek, void(base::TimeDelta time, FilterCallback* callback));
+  MOCK_METHOD1(Seek, void(base::TimeDelta time));
 
   // AudioRenderer implementation.
-  MOCK_METHOD2(Initialize, void(AudioDecoder* decoder,
-                                FilterCallback* callback));
+  MOCK_METHOD1(Initialize, bool(AudioDecoder* decoder));
   MOCK_METHOD1(SetVolume, void(float volume));
 
  protected:
@@ -322,15 +263,11 @@ class MockFilterFactory : public FilterFactory {
   DISALLOW_COPY_AND_ASSIGN(MockFilterFactory);
 };
 
-// Helper gmock function that immediately executes and destroys the
-// FilterCallback on behalf of the provided filter.  Can be used when mocking
-// the Initialize() and Seek() methods.
-void RunFilterCallback(::testing::Unused, FilterCallback* callback);
-
-// Helper gmock function that immediately destroys the FilterCallback on behalf
-// of the provided filter.  Can be used when mocking the Initialize() and Seek()
-// methods.
-void DestroyFilterCallback(::testing::Unused, FilterCallback* callback);
+// Helper gmock action that calls InitializationComplete() on behalf of the
+// provided filter.
+ACTION_P(InitializationComplete, filter) {
+  filter->host()->InitializationComplete();
+}
 
 // Helper gmock action that calls Error() on behalf of the provided filter.
 ACTION_P2(Error, filter, error) {
diff --git a/media/base/pipeline_impl.cc b/media/base/pipeline_impl.cc
index 7ef051d..f021ad3 100644
--- a/media/base/pipeline_impl.cc
+++ b/media/base/pipeline_impl.cc
@@ -328,6 +328,15 @@ void PipelineInternal::VolumeChanged(float volume) {
       NewRunnableMethod(this, &PipelineInternal::VolumeChangedTask, volume));
 }
 
+// Called from any thread.
+void PipelineInternal::InitializationComplete(FilterHostImpl* host) {
+  if (IsPipelineOk()) {
+    // Continue the initialize task by proceeding to the next stage.
+    message_loop_->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &PipelineInternal::InitializeTask));
+  }
+}
+
 // Called from any thread.  Updates the pipeline time.
 void PipelineInternal::SetTime(base::TimeDelta time) {
   // TODO(scherkus): why not post a task?
@@ -341,19 +350,6 @@ void PipelineInternal::Error(PipelineError error) {
       NewRunnableMethod(this, &PipelineInternal::ErrorTask, error));
 }
 
-// Called from any thread.
-void PipelineInternal::OnFilterInitialize() {
-  // Continue the initialize task by proceeding to the next stage.
-  message_loop_->PostTask(FROM_HERE,
-      NewRunnableMethod(this, &PipelineInternal::InitializeTask));
-}
-
-// Called from any thread.
-void PipelineInternal::OnFilterSeek() {
-  // TODO(scherkus): have PipelineInternal wait to receive replies from every
-  // filter before calling the client's |seek_callback_|.
-}
-
 void PipelineInternal::StartTask(FilterFactory* filter_factory,
                                  const std::string& url,
                                  PipelineCallback* start_callback) {
@@ -387,8 +383,8 @@ void PipelineInternal::StartTask(FilterFactory* filter_factory,
 void PipelineInternal::InitializeTask() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
 
-  // If we have received the stop or error signal, return immediately.
-  if (state_ == kStopped || state_ == kError)
+  // If we have received the stop signal, return immediately.
+  if (state_ == kStopped)
     return;
 
   DCHECK(state_ == kCreated || IsPipelineInitializing());
@@ -555,8 +551,7 @@ void PipelineInternal::SeekTask(base::TimeDelta time,
   for (FilterHostVector::iterator iter = filter_hosts_.begin();
        iter != filter_hosts_.end();
        ++iter) {
-    (*iter)->media_filter()->Seek(time,
-        NewCallback(this, &PipelineInternal::OnFilterSeek));
+    (*iter)->media_filter()->Seek(time);
   }
 
   // TODO(hclam): we should set the time when the above seek operations were all
@@ -607,8 +602,9 @@ void PipelineInternal::CreateFilter(FilterFactory* filter_factory,
   filter_hosts_.push_back(host.release());
 
   // Now initialize the filter.
-  filter->Initialize(source,
-      NewCallback(this, &PipelineInternal::OnFilterInitialize));
+  if (!filter->Initialize(source)) {
+    Error(PIPELINE_ERROR_INITIALIZATION_FAILED);
+  }
 }
 
 void PipelineInternal::CreateDataSource() {
diff --git a/media/base/pipeline_impl.h b/media/base/pipeline_impl.h
index e9f44d5..911d253 100644
--- a/media/base/pipeline_impl.h
+++ b/media/base/pipeline_impl.h
@@ -191,6 +191,11 @@ class PipelineInternal : public base::RefCountedThreadSafe<PipelineInternal> {
   // Methods called by a FilterHostImpl object.  These methods may be called
   // on any thread, either the pipeline's thread or any other.
 
+  // When a filter calls it's FilterHost, the filter host calls back to the
+  // pipeline thread.  If the pipeline thread is running a nested message loop
+  // then it will be exited.
+  void InitializationComplete(FilterHostImpl* host);
+
   // Sets the pipeline time and schedules a task to call back to any filters
   // that have registered a time update callback.
   void SetTime(base::TimeDelta time);
@@ -240,10 +245,6 @@ class PipelineInternal : public base::RefCountedThreadSafe<PipelineInternal> {
            state_ == kInitVideoRenderer;
   }
 
-  // Callback executed by filters upon completing initialization and seeking.
-  void OnFilterInitialize();
-  void OnFilterSeek();
-
   // The following "task" methods correspond to the public methods, but these
   // methods are run as the result of posting a task to the PipelineInternal's
   // message loop.
diff --git a/media/base/pipeline_impl_unittest.cc b/media/base/pipeline_impl_unittest.cc
index d6bd285..b673cfb 100644
--- a/media/base/pipeline_impl_unittest.cc
+++ b/media/base/pipeline_impl_unittest.cc
@@ -14,9 +14,7 @@
 #include "testing/gtest/include/gtest/gtest.h"
 
 using ::testing::DoAll;
-using ::testing::Invoke;
 using ::testing::Mock;
-using ::testing::NotNull;
 using ::testing::Return;
 using ::testing::StrictMock;
 
@@ -65,8 +63,9 @@ class PipelineImplTest : public ::testing::Test {
  protected:
   // Sets up expectations to allow the data source to initialize.
   void InitializeDataSource() {
-    EXPECT_CALL(*mocks_->data_source(), Initialize("", NotNull()))
-        .WillOnce(Invoke(&RunFilterCallback));
+    EXPECT_CALL(*mocks_->data_source(), Initialize(""))
+        .WillOnce(DoAll(InitializationComplete(mocks_->data_source()),
+                        Return(true)));
     EXPECT_CALL(*mocks_->data_source(), SetPlaybackRate(0.0f));
     EXPECT_CALL(*mocks_->data_source(), Stop());
   }
@@ -74,9 +73,9 @@ class PipelineImplTest : public ::testing::Test {
   // Sets up expectations to allow the demuxer to initialize.
   typedef std::vector<MockDemuxerStream*> MockDemuxerStreamVector;
   void InitializeDemuxer(MockDemuxerStreamVector* streams) {
-    EXPECT_CALL(*mocks_->demuxer(),
-                Initialize(mocks_->data_source(), NotNull()))
-        .WillOnce(Invoke(&RunFilterCallback));
+    EXPECT_CALL(*mocks_->demuxer(), Initialize(mocks_->data_source()))
+        .WillOnce(DoAll(InitializationComplete(mocks_->demuxer()),
+                        Return(true)));
     EXPECT_CALL(*mocks_->demuxer(), GetNumberOfStreams())
         .WillRepeatedly(Return(streams->size()));
     EXPECT_CALL(*mocks_->demuxer(), SetPlaybackRate(0.0f));
@@ -92,34 +91,36 @@ class PipelineImplTest : public ::testing::Test {
 
   // Sets up expectations to allow the video decoder to initialize.
   void InitializeVideoDecoder(MockDemuxerStream* stream) {
-    EXPECT_CALL(*mocks_->video_decoder(), Initialize(stream, NotNull()))
-        .WillOnce(Invoke(&RunFilterCallback));
+    EXPECT_CALL(*mocks_->video_decoder(), Initialize(stream))
+        .WillOnce(DoAll(InitializationComplete(mocks_->video_decoder()),
+                        Return(true)));
     EXPECT_CALL(*mocks_->video_decoder(), SetPlaybackRate(0.0f));
     EXPECT_CALL(*mocks_->video_decoder(), Stop());
   }
 
   // Sets up expectations to allow the audio decoder to initialize.
   void InitializeAudioDecoder(MockDemuxerStream* stream) {
-    EXPECT_CALL(*mocks_->audio_decoder(), Initialize(stream, NotNull()))
-        .WillOnce(Invoke(&RunFilterCallback));
+    EXPECT_CALL(*mocks_->audio_decoder(), Initialize(stream))
+        .WillOnce(DoAll(InitializationComplete(mocks_->audio_decoder()),
+                        Return(true)));
     EXPECT_CALL(*mocks_->audio_decoder(), SetPlaybackRate(0.0f));
     EXPECT_CALL(*mocks_->audio_decoder(), Stop());
   }
 
   // Sets up expectations to allow the video renderer to initialize.
   void InitializeVideoRenderer() {
-    EXPECT_CALL(*mocks_->video_renderer(),
-                Initialize(mocks_->video_decoder(), NotNull()))
-        .WillOnce(Invoke(&RunFilterCallback));
+    EXPECT_CALL(*mocks_->video_renderer(), Initialize(mocks_->video_decoder()))
+        .WillOnce(DoAll(InitializationComplete(mocks_->video_renderer()),
+                        Return(true)));
     EXPECT_CALL(*mocks_->video_renderer(), SetPlaybackRate(0.0f));
     EXPECT_CALL(*mocks_->video_renderer(), Stop());
   }
 
   // Sets up expectations to allow the audio renderer to initialize.
   void InitializeAudioRenderer() {
-    EXPECT_CALL(*mocks_->audio_renderer(),
-                Initialize(mocks_->audio_decoder(), NotNull()))
-        .WillOnce(Invoke(&RunFilterCallback));
+    EXPECT_CALL(*mocks_->audio_renderer(), Initialize(mocks_->audio_decoder()))
+        .WillOnce(DoAll(InitializationComplete(mocks_->audio_renderer()),
+                        Return(true)));
     EXPECT_CALL(*mocks_->audio_renderer(), SetPlaybackRate(0.0f));
     EXPECT_CALL(*mocks_->audio_renderer(), SetVolume(1.0f));
     EXPECT_CALL(*mocks_->audio_renderer(), Stop());
@@ -200,8 +201,8 @@ TEST_F(PipelineImplTest, NotStarted) {
 }
 
 TEST_F(PipelineImplTest, NeverInitializes) {
-  EXPECT_CALL(*mocks_->data_source(), Initialize("", NotNull()))
-      .WillOnce(Invoke(&DestroyFilterCallback));
+  EXPECT_CALL(*mocks_->data_source(), Initialize(""))
+      .WillOnce(Return(true));
   EXPECT_CALL(*mocks_->data_source(), Stop());
 
   // This test hangs during initialization by never calling
@@ -232,10 +233,10 @@ TEST_F(PipelineImplTest, RequiredFilterMissing) {
 }
 
 TEST_F(PipelineImplTest, URLNotFound) {
-  EXPECT_CALL(*mocks_->data_source(), Initialize("", NotNull()))
+  EXPECT_CALL(*mocks_->data_source(), Initialize(""))
       .WillOnce(DoAll(Error(mocks_->data_source(),
                             PIPELINE_ERROR_URL_NOT_FOUND),
-                      Invoke(&RunFilterCallback)));
+                      Return(false)));
   EXPECT_CALL(*mocks_->data_source(), Stop());
 
   InitializePipeline();
@@ -246,12 +247,14 @@ TEST_F(PipelineImplTest, URLNotFound) {
 TEST_F(PipelineImplTest, NoStreams) {
   // Manually set these expecations because SetPlaybackRate() is not called if
   // we cannot fully initialize the pipeline.
-  EXPECT_CALL(*mocks_->data_source(), Initialize("", NotNull()))
-      .WillOnce(Invoke(&RunFilterCallback));
+  EXPECT_CALL(*mocks_->data_source(), Initialize(""))
+      .WillOnce(DoAll(InitializationComplete(mocks_->data_source()),
+                      Return(true)));
   EXPECT_CALL(*mocks_->data_source(), Stop());
 
-  EXPECT_CALL(*mocks_->demuxer(), Initialize(mocks_->data_source(), NotNull()))
-      .WillOnce(Invoke(&RunFilterCallback));
+  EXPECT_CALL(*mocks_->demuxer(), Initialize(mocks_->data_source()))
+      .WillOnce(DoAll(InitializationComplete(mocks_->demuxer()),
+                      Return(true)));
   EXPECT_CALL(*mocks_->demuxer(), GetNumberOfStreams())
       .WillRepeatedly(Return(0));
   EXPECT_CALL(*mocks_->demuxer(), Stop());
@@ -338,18 +341,12 @@ TEST_F(PipelineImplTest, Seek) {
 
   // Every filter should receive a call to Seek().
   base::TimeDelta expected = base::TimeDelta::FromSeconds(2000);
-  EXPECT_CALL(*mocks_->data_source(), Seek(expected, NotNull()))
-      .WillOnce(Invoke(&RunFilterCallback));
-  EXPECT_CALL(*mocks_->demuxer(), Seek(expected, NotNull()))
-      .WillOnce(Invoke(&RunFilterCallback));
-  EXPECT_CALL(*mocks_->audio_decoder(), Seek(expected, NotNull()))
-      .WillOnce(Invoke(&RunFilterCallback));
-  EXPECT_CALL(*mocks_->audio_renderer(), Seek(expected, NotNull()))
-      .WillOnce(Invoke(&RunFilterCallback));
-  EXPECT_CALL(*mocks_->video_decoder(), Seek(expected, NotNull()))
-      .WillOnce(Invoke(&RunFilterCallback));
-  EXPECT_CALL(*mocks_->video_renderer(), Seek(expected, NotNull()))
-      .WillOnce(Invoke(&RunFilterCallback));
+  EXPECT_CALL(*mocks_->data_source(), Seek(expected));
+  EXPECT_CALL(*mocks_->demuxer(), Seek(expected));
+  EXPECT_CALL(*mocks_->audio_decoder(), Seek(expected));
+  EXPECT_CALL(*mocks_->audio_renderer(), Seek(expected));
+  EXPECT_CALL(*mocks_->video_decoder(), Seek(expected));
+  EXPECT_CALL(*mocks_->video_renderer(), Seek(expected));
 
   // We expect a successful seek callback.
   EXPECT_CALL(callbacks_, OnSeek());