DemuxerFactory is born!

PipelineImpl no longer knows about DataSources, with DemuxerFactory taking a
DataSourceFactory when appropriate (i.e. for FFmpegDemuxerFactory).

This diff is relative to patchset 8 of issue 6480050, which must be submitted
first.

BUG=none
TEST=media_unittests pass; chrome plays videos; trybots passed: mac,linux,linux_clang,win_layout; trybots with only unrelated-looking failures: win,{mac,linux}_layout

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

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

10 files changed, 137 insertions, 165 deletions

diff --git a/media/base/filter_collection.cc b/media/base/filter_collection.cc
index c25fa7a..ed2339f 100644
--- a/media/base/filter_collection.cc
+++ b/media/base/filter_collection.cc
@@ -12,17 +12,13 @@ FilterCollection::FilterCollection() {}
 
 FilterCollection::~FilterCollection() {}
 
-void FilterCollection::SetDataSourceFactory(DataSourceFactory* factory) {
+void FilterCollection::SetDemuxerFactory(DemuxerFactory* factory) {
   DCHECK(factory);
-  data_source_factory_.reset(factory);
+  demuxer_factory_.reset(factory);
 }
 
-DataSourceFactory* FilterCollection::GetDataSourceFactory() {
-  return data_source_factory_.get();
-}
-
-void FilterCollection::AddDemuxer(Demuxer* filter) {
-  AddFilter(DEMUXER, filter);
+DemuxerFactory* FilterCollection::GetDemuxerFactory() {
+  return demuxer_factory_.get();
 }
 
 void FilterCollection::AddVideoDecoder(VideoDecoder* filter) {
@@ -49,10 +45,6 @@ void FilterCollection::Clear() {
   filters_.clear();
 }
 
-void FilterCollection::SelectDemuxer(scoped_refptr<Demuxer>* filter_out) {
-  SelectFilter<DEMUXER>(filter_out);
-}
-
 void FilterCollection::SelectVideoDecoder(
     scoped_refptr<VideoDecoder>* filter_out) {
   SelectFilter<VIDEO_DECODER>(filter_out);
diff --git a/media/base/filter_collection.h b/media/base/filter_collection.h
index 6fc445d..0c02566 100644
--- a/media/base/filter_collection.h
+++ b/media/base/filter_collection.h
@@ -20,13 +20,12 @@ class FilterCollection {
   FilterCollection();
   ~FilterCollection();
 
-  // DataSourceFactory accessor methods.
+  // DemuxerFactory accessor methods.
   // FilterCollection takes ownership of the factory here.
-  void SetDataSourceFactory(DataSourceFactory* factory);
-  DataSourceFactory* GetDataSourceFactory();
+  void SetDemuxerFactory(DemuxerFactory* factory);
+  DemuxerFactory* GetDemuxerFactory();
 
   // Adds a filter to the collection.
-  void AddDemuxer(Demuxer* filter);
   void AddVideoDecoder(VideoDecoder* filter);
   void AddAudioDecoder(AudioDecoder* filter);
   void AddVideoRenderer(VideoRenderer* filter);
@@ -41,7 +40,6 @@ class FilterCollection {
   // Selects a filter of the specified type from the collection.
   // If the required filter cannot be found, NULL is returned.
   // If a filter is returned it is removed from the collection.
-  void SelectDemuxer(scoped_refptr<Demuxer>* filter_out);
   void SelectVideoDecoder(scoped_refptr<VideoDecoder>* filter_out);
   void SelectAudioDecoder(scoped_refptr<AudioDecoder>* filter_out);
   void SelectVideoRenderer(scoped_refptr<VideoRenderer>* filter_out);
@@ -52,7 +50,6 @@ class FilterCollection {
   // the following types. This is used to mark, identify, and support
   // downcasting of different filter types stored in the filters_ list.
   enum FilterType {
-    DEMUXER,
     AUDIO_DECODER,
     VIDEO_DECODER,
     AUDIO_RENDERER,
@@ -63,7 +60,7 @@ class FilterCollection {
   typedef std::pair<FilterType, scoped_refptr<Filter> > FilterListElement;
   typedef std::list<FilterListElement> FilterList;
   FilterList filters_;
-  scoped_ptr<DataSourceFactory> data_source_factory_;
+  scoped_ptr<DemuxerFactory> demuxer_factory_;
 
   // Helper function that adds a filter to the filter list.
   void AddFilter(FilterType filter_type, Filter* filter);
diff --git a/media/base/filter_factories.cc b/media/base/filter_factories.cc
index bb5df5e..c074cc6 100644
--- a/media/base/filter_factories.cc
+++ b/media/base/filter_factories.cc
@@ -8,4 +8,6 @@ namespace media {
 
 DataSourceFactory::~DataSourceFactory() {}
 
+DemuxerFactory::~DemuxerFactory() {}
+
 }  // namespace media
diff --git a/media/base/filter_factories.h b/media/base/filter_factories.h
index c00adef..aad391f 100644
--- a/media/base/filter_factories.h
+++ b/media/base/filter_factories.h
@@ -30,6 +30,24 @@ class DataSourceFactory {
   virtual DataSourceFactory* Clone() const = 0;
 };
 
+class Demuxer;
+
+// Asynchronous factory interface for building Demuxer objects.
+class DemuxerFactory {
+ public:
+  // Ownership of the Demuxer is transferred through this callback.
+  typedef Callback2<PipelineError, Demuxer*>::Type BuildCallback;
+
+  virtual ~DemuxerFactory();
+
+  // Builds a Demuxer for |url| and returns it via |callback|.
+  virtual void Build(const std::string& url, BuildCallback* callback) = 0;
+
+  // Makes a copy of this factory.
+  // NOTE: Pending requests are not cloned.
+  virtual DemuxerFactory* Clone() const = 0;
+};
+
 }  // namespace media
 
 #endif  // MEDIA_BASE_FILTER_FACTORIES_H_
diff --git a/media/base/filters.h b/media/base/filters.h
index 89a7584..654da5a 100644
--- a/media/base/filters.h
+++ b/media/base/filters.h
@@ -132,11 +132,6 @@ class DataSource : public Filter {
 
 class Demuxer : public Filter {
  public:
-  // Initialize a Demuxer with the given DataSource, executing the callback upon
-  // completion.
-  virtual void Initialize(DataSource* data_source,
-                          FilterCallback* callback) = 0;
-
   // Returns the number of streams available
   virtual size_t GetNumberOfStreams() = 0;
 
diff --git a/media/base/mock_filters.cc b/media/base/mock_filters.cc
index 68e344f..fbf2bb1 100644
--- a/media/base/mock_filters.cc
+++ b/media/base/mock_filters.cc
@@ -35,51 +35,71 @@ void MockDataSource::SetTotalAndBufferedBytes(int64 total_bytes,
   buffered_bytes_ = buffered_bytes;
 }
 
-MockDataSourceFactory::MockDataSourceFactory(MockDataSource* data_source)
-    : data_source_(data_source),
-      error_(PIPELINE_OK) {
+MockDemuxerFactory::MockDemuxerFactory(MockDemuxer* demuxer)
+    : demuxer_(demuxer), error_(PIPELINE_OK) {
 }
 
-MockDataSourceFactory::~MockDataSourceFactory() {}
+MockDemuxerFactory::~MockDemuxerFactory() {}
 
-void MockDataSourceFactory::SetError(PipelineError error) {
+void MockDemuxerFactory::SetError(PipelineError error) {
   error_ = error;
 }
 
-void MockDataSourceFactory::RunBuildCallback(const std::string& url,
-                                             BuildCallback* callback) {
+void MockDemuxerFactory::RunBuildCallback(const std::string& url,
+                                          BuildCallback* callback) {
   scoped_ptr<BuildCallback> cb(callback);
 
-  if (!data_source_.get()) {
+  if (!demuxer_.get()) {
     cb->Run(PIPELINE_ERROR_REQUIRED_FILTER_MISSING,
-            static_cast<DataSource*>(NULL));
+            static_cast<Demuxer*>(NULL));
     return;
   }
 
-  scoped_refptr<MockDataSource> data_source = data_source_;
-  data_source_ = NULL;
+  scoped_refptr<MockDemuxer> demuxer = demuxer_;
+  demuxer_ = NULL;
 
   if (error_ == PIPELINE_OK) {
-    cb->Run(PIPELINE_OK, data_source.get());
+    cb->Run(PIPELINE_OK, demuxer.get());
     return;
   }
 
-  cb->Run(error_, static_cast<DataSource*>(NULL));
+  cb->Run(error_, static_cast<Demuxer*>(NULL));
 }
 
-void MockDataSourceFactory::DestroyBuildCallback(const std::string& url,
-                                                 BuildCallback* callback) {
+void MockDemuxerFactory::DestroyBuildCallback(const std::string& url,
+                                              BuildCallback* callback) {
   delete callback;
 }
 
-DataSourceFactory* MockDataSourceFactory::Clone() const {
-  return new MockDataSourceFactory(data_source_.get());
+DemuxerFactory* MockDemuxerFactory::Clone() const {
+  return new MockDemuxerFactory(demuxer_.get());
 }
 
-MockDemuxer::MockDemuxer() {}
+MockDemuxer::MockDemuxer()
+  : total_bytes_(-1), buffered_bytes_(-1), duration_() {}
 
 MockDemuxer::~MockDemuxer() {}
 
+void MockDemuxer::set_host(FilterHost* filter_host) {
+  Filter::set_host(filter_host);
+
+  if (total_bytes_ > 0)
+    host()->SetTotalBytes(total_bytes_);
+
+  if (buffered_bytes_ > 0)
+    host()->SetBufferedBytes(buffered_bytes_);
+
+  if (duration_.InMilliseconds() > 0)
+    host()->SetDuration(duration_);
+}
+
+void MockDemuxer::SetTotalAndBufferedBytesAndDuration(
+    int64 total_bytes, int64 buffered_bytes, const base::TimeDelta& duration) {
+  total_bytes_ = total_bytes;
+  buffered_bytes_ = buffered_bytes;
+  duration_ = duration;
+}
+
 MockDemuxerStream::MockDemuxerStream() {}
 
 MockDemuxerStream::~MockDemuxerStream() {}
@@ -101,8 +121,7 @@ MockAudioRenderer::MockAudioRenderer() {}
 MockAudioRenderer::~MockAudioRenderer() {}
 
 MockFilterCollection::MockFilterCollection()
-    : data_source_(new MockDataSource()),
-      demuxer_(new MockDemuxer()),
+    : demuxer_(new MockDemuxer()),
       video_decoder_(new MockVideoDecoder()),
       audio_decoder_(new MockAudioDecoder()),
       video_renderer_(new MockVideoRenderer()),
@@ -112,30 +131,27 @@ MockFilterCollection::MockFilterCollection()
 MockFilterCollection::~MockFilterCollection() {}
 
 FilterCollection* MockFilterCollection::filter_collection(
-    bool include_data_source,
+    bool include_demuxer,
     bool run_build_callback,
     PipelineError build_error) const {
   FilterCollection* collection = new FilterCollection();
 
-  MockDataSourceFactory* data_source_factory =
-      new MockDataSourceFactory(include_data_source ? data_source_ : NULL);
+  MockDemuxerFactory* demuxer_factory =
+      new MockDemuxerFactory(include_demuxer ? demuxer_ : NULL);
 
   if (build_error != PIPELINE_OK)
-    data_source_factory->SetError(build_error);
+    demuxer_factory->SetError(build_error);
 
   if (run_build_callback) {
-    ON_CALL(*data_source_factory, Build(_, NotNull()))
-        .WillByDefault(Invoke(data_source_factory,
-                              &MockDataSourceFactory::RunBuildCallback));
+    ON_CALL(*demuxer_factory, Build(_, NotNull())).WillByDefault(Invoke(
+        demuxer_factory, &MockDemuxerFactory::RunBuildCallback));
   } else {
-    ON_CALL(*data_source_factory, Build(_, NotNull()))
-        .WillByDefault(Invoke(data_source_factory,
-                              &MockDataSourceFactory::DestroyBuildCallback));
+    ON_CALL(*demuxer_factory, Build(_, NotNull())).WillByDefault(Invoke(
+        demuxer_factory, &MockDemuxerFactory::DestroyBuildCallback));
   }
-  EXPECT_CALL(*data_source_factory, Build(_, NotNull()));
+  EXPECT_CALL(*demuxer_factory, Build(_, NotNull()));
 
-  collection->SetDataSourceFactory(data_source_factory);
-  collection->AddDemuxer(demuxer_);
+  collection->SetDemuxerFactory(demuxer_factory);
   collection->AddVideoDecoder(video_decoder_);
   collection->AddAudioDecoder(audio_decoder_);
   collection->AddVideoRenderer(video_renderer_);
@@ -148,6 +164,12 @@ void RunFilterCallback(::testing::Unused, FilterCallback* callback) {
   delete callback;
 }
 
+void RunPipelineStatusCallback(
+    PipelineError status, PipelineStatusCallback* callback) {
+  callback->Run(status);
+  delete callback;
+}
+
 void RunFilterCallback3(::testing::Unused, FilterCallback* callback,
                         ::testing::Unused) {
   callback->Run();
diff --git a/media/base/mock_filters.h b/media/base/mock_filters.h
index cc121ba..d1ae03d 100644
--- a/media/base/mock_filters.h
+++ b/media/base/mock_filters.h
@@ -97,31 +97,11 @@ class MockDataSource : public DataSource {
   DISALLOW_COPY_AND_ASSIGN(MockDataSource);
 };
 
-class MockDataSourceFactory : public DataSourceFactory {
- public:
-  explicit MockDataSourceFactory(MockDataSource* data_source);
-  virtual ~MockDataSourceFactory();
-
-  void SetError(PipelineError error);
-  void RunBuildCallback(const std::string& url, BuildCallback* callback);
-  void DestroyBuildCallback(const std::string& url, BuildCallback* callback);
-
-  // DataSourceFactory methods.
-  MOCK_METHOD2(Build, void(const std::string& url, BuildCallback* callback));
-  virtual DataSourceFactory* Clone() const;
-
- private:
-  scoped_refptr<MockDataSource> data_source_;
-  PipelineError error_;
-
-  DISALLOW_COPY_AND_ASSIGN(MockDataSourceFactory);
-};
-
 class MockDemuxer : public Demuxer {
  public:
   MockDemuxer();
-
   // Filter implementation.
+  virtual void set_host(FilterHost* host);
   MOCK_METHOD1(Stop, void(FilterCallback* callback));
   MOCK_METHOD1(SetPlaybackRate, void(float playback_rate));
   MOCK_METHOD2(Seek, void(base::TimeDelta time, FilterCallback* callback));
@@ -133,13 +113,42 @@ class MockDemuxer : public Demuxer {
   MOCK_METHOD0(GetNumberOfStreams, size_t());
   MOCK_METHOD1(GetStream, scoped_refptr<DemuxerStream>(int stream_id));
 
+  // Sets the TotalBytes, BufferedBytes, & Duration values to be sent to host()
+  // when set_host() is called.
+  void SetTotalAndBufferedBytesAndDuration(
+      int64 total_bytes, int64 buffered_bytes, const base::TimeDelta& duration);
+
  protected:
   virtual ~MockDemuxer();
 
  private:
+  int64 total_bytes_;
+  int64 buffered_bytes_;
+  base::TimeDelta duration_;
+
   DISALLOW_COPY_AND_ASSIGN(MockDemuxer);
 };
 
+class MockDemuxerFactory : public DemuxerFactory {
+ public:
+  explicit MockDemuxerFactory(MockDemuxer* demuxer);
+  virtual ~MockDemuxerFactory();
+
+  void SetError(PipelineError error);
+  void RunBuildCallback(const std::string& url, BuildCallback* callback);
+  void DestroyBuildCallback(const std::string& url, BuildCallback* callback);
+
+  // DemuxerFactory methods.
+  MOCK_METHOD2(Build, void(const std::string& url, BuildCallback* callback));
+  virtual DemuxerFactory* Clone() const;
+
+ private:
+  scoped_refptr<MockDemuxer> demuxer_;
+  PipelineError error_;
+
+  DISALLOW_COPY_AND_ASSIGN(MockDemuxerFactory);
+};
+
 class MockDemuxerStream : public DemuxerStream {
  public:
   MockDemuxerStream();
@@ -274,7 +283,6 @@ class MockFilterCollection {
   virtual ~MockFilterCollection();
 
   // Mock accessors.
-  MockDataSource* data_source() const { return data_source_; }
   MockDemuxer* demuxer() const { return demuxer_; }
   MockVideoDecoder* video_decoder() const { return video_decoder_; }
   MockAudioDecoder* audio_decoder() const { return audio_decoder_; }
@@ -285,12 +293,11 @@ class MockFilterCollection {
     return filter_collection(true, true, PIPELINE_OK);
   }
 
-  FilterCollection* filter_collection(bool include_data_source,
+  FilterCollection* filter_collection(bool include_demuxer,
                                       bool run_build_callback,
                                       PipelineError build_error) const;
 
  private:
-  scoped_refptr<MockDataSource> data_source_;
   scoped_refptr<MockDemuxer> demuxer_;
   scoped_refptr<MockVideoDecoder> video_decoder_;
   scoped_refptr<MockAudioDecoder> audio_decoder_;
@@ -304,6 +311,8 @@ class MockFilterCollection {
 // FilterCallback on behalf of the provided filter.  Can be used when mocking
 // the Initialize() and Seek() methods.
 void RunFilterCallback(::testing::Unused, FilterCallback* callback);
+void RunPipelineStatusCallback(PipelineError status,
+                               PipelineStatusCallback* callback);
 void RunFilterCallback3(::testing::Unused, FilterCallback* callback,
                         ::testing::Unused);
 
diff --git a/media/base/pipeline_impl.cc b/media/base/pipeline_impl.cc
index 4f19fcc..958cd5c 100644
--- a/media/base/pipeline_impl.cc
+++ b/media/base/pipeline_impl.cc
@@ -549,23 +549,23 @@ void PipelineImpl::StartTask(FilterCollection* filter_collection,
   seek_callback_.reset(start_callback);
 
   // Kick off initialization.
-  set_state(kInitDataSource);
+  set_state(kInitDemuxer);
   pipeline_init_state_.reset(new PipelineInitState());
   pipeline_init_state_->composite_ = new CompositeFilter(message_loop_);
   pipeline_init_state_->composite_->set_host(this);
 
-  InitializeDataSource();
+  InitializeDemuxer();
 }
 
 // Main initialization method called on the pipeline thread.  This code attempts
 // to use the specified filter factory to build a pipeline.
 // Initialization step performed in this method depends on current state of this
 // object, indicated by |state_|.  After each step of initialization, this
-// object transits to the next stage.  It starts by creating a DataSource,
-// connects it to a Demuxer, and then connects the Demuxer's audio stream to an
-// AudioDecoder which is then connected to an AudioRenderer.  If the media has
-// video, then it connects a VideoDecoder to the Demuxer's video stream, and
-// then connects the VideoDecoder to a VideoRenderer.
+// object transits to the next stage.  It starts by creating a Demuxer, and then
+// connects the Demuxer's audio stream to an AudioDecoder which is then
+// connected to an AudioRenderer.  If the media has video, then it connects a
+// VideoDecoder to the Demuxer's video stream, and then connects the
+// VideoDecoder to a VideoRenderer.
 //
 // When all required filters have been created and have called their
 // FilterHost's InitializationComplete() method, the pipeline will update its
@@ -938,7 +938,6 @@ void PipelineImpl::TeardownStateTransitionTask() {
 
     case kCreated:
     case kError:
-    case kInitDataSource:
     case kInitDemuxer:
     case kInitAudioDecoder:
     case kInitAudioRenderer:
@@ -994,22 +993,22 @@ bool PipelineImpl::PrepareFilter(scoped_refptr<Filter> filter) {
   return ret;
 }
 
-void PipelineImpl::InitializeDataSource() {
+void PipelineImpl::InitializeDemuxer() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   DCHECK(IsPipelineOk());
 
-  filter_collection_->GetDataSourceFactory()->Build(url_,
-      NewCallback(this, &PipelineImpl::OnDataSourceBuilt));
+  filter_collection_->GetDemuxerFactory()->Build(url_,
+      NewCallback(this, &PipelineImpl::OnDemuxerBuilt));
 }
 
-void PipelineImpl::OnDataSourceBuilt(PipelineError error,
-                                     DataSource* data_source) {
+void PipelineImpl::OnDemuxerBuilt(PipelineError error,
+                                  Demuxer* demuxer) {
   if (MessageLoop::current() != message_loop_) {
     message_loop_->PostTask(FROM_HERE,
                             NewRunnableMethod(this,
-                                              &PipelineImpl::OnDataSourceBuilt,
+                                              &PipelineImpl::OnDemuxerBuilt,
                                               error,
-                                              make_scoped_refptr(data_source)));
+                                              make_scoped_refptr(demuxer)));
     return;
   }
 
@@ -1018,22 +1017,6 @@ void PipelineImpl::OnDataSourceBuilt(PipelineError error,
     return;
   }
 
-  PrepareFilter(data_source);
-
-  set_state(kInitDemuxer);
-  InitializeDemuxer(data_source);
-}
-
-void PipelineImpl::InitializeDemuxer(
-    const scoped_refptr<DataSource>& data_source) {
-  DCHECK_EQ(MessageLoop::current(), message_loop_);
-  DCHECK(IsPipelineOk());
-
-  scoped_refptr<Demuxer> demuxer;
-
-  CHECK(data_source);
-
-  filter_collection_->SelectDemuxer(&demuxer);
   if (!demuxer) {
     SetError(PIPELINE_ERROR_REQUIRED_FILTER_MISSING);
     return;
@@ -1043,8 +1026,7 @@ void PipelineImpl::InitializeDemuxer(
     return;
 
   pipeline_init_state_->demuxer_ = demuxer;
-  demuxer->Initialize(data_source,
-                      NewCallback(this, &PipelineImpl::OnFilterInitialize));
+  OnFilterInitialize();
 }
 
 bool PipelineImpl::InitializeAudioDecoder(
@@ -1189,7 +1171,6 @@ void PipelineImpl::TearDownPipeline() {
           NewRunnableMethod(this, &PipelineImpl::FinishDestroyingFiltersTask));
       break;
 
-    case kInitDataSource:
     case kInitDemuxer:
     case kInitAudioDecoder:
     case kInitAudioRenderer:
diff --git a/media/base/pipeline_impl.h b/media/base/pipeline_impl.h
index 0fe8c09..8f0e4ea 100644
--- a/media/base/pipeline_impl.h
+++ b/media/base/pipeline_impl.h
@@ -101,7 +101,6 @@ class PipelineImpl : public Pipeline, public FilterHost {
   // Pipeline states, as described above.
   enum State {
     kCreated,
-    kInitDataSource,
     kInitDemuxer,
     kInitAudioDecoder,
     kInitAudioRenderer,
@@ -245,10 +244,8 @@ class PipelineImpl : public Pipeline, public FilterHost {
 
   // The following initialize methods are used to select a specific type of
   // Filter object from FilterCollection and initialize it asynchronously.
-  void InitializeDataSource();
-  void OnDataSourceBuilt(PipelineError error, DataSource* data_source);
-
-  void InitializeDemuxer(const scoped_refptr<DataSource>& data_source);
+  void InitializeDemuxer();
+  void OnDemuxerBuilt(PipelineError error, Demuxer* demuxer);
 
   // Returns true if the asynchronous action of creating decoder has started.
   // Returns false if this method did nothing because the corresponding
diff --git a/media/base/pipeline_impl_unittest.cc b/media/base/pipeline_impl_unittest.cc
index 486b5e0..ce4974f 100644
--- a/media/base/pipeline_impl_unittest.cc
+++ b/media/base/pipeline_impl_unittest.cc
@@ -83,26 +83,12 @@ class PipelineImplTest : public ::testing::Test {
   }
 
  protected:
-  // Sets up expectations to allow the data source to initialize.
-  void InitializeDataSource() {
-    mocks_->data_source()->SetTotalAndBufferedBytes(kTotalBytes,
-                                                    kBufferedBytes);
-
-    EXPECT_CALL(*mocks_->data_source(), SetPlaybackRate(0.0f));
-    EXPECT_CALL(*mocks_->data_source(), Seek(base::TimeDelta(), NotNull()))
-        .WillOnce(Invoke(&RunFilterCallback));
-    EXPECT_CALL(*mocks_->data_source(), Stop(NotNull()))
-        .WillOnce(Invoke(&RunStopFilterCallback));
-  }
-
   // Sets up expectations to allow the demuxer to initialize.
   typedef std::vector<MockDemuxerStream*> MockDemuxerStreamVector;
   void InitializeDemuxer(MockDemuxerStreamVector* streams,
                          const base::TimeDelta& duration) {
-    EXPECT_CALL(*mocks_->demuxer(),
-                Initialize(mocks_->data_source(), NotNull()))
-        .WillOnce(DoAll(SetDuration(mocks_->data_source(), duration),
-                        Invoke(&RunFilterCallback)));
+    mocks_->demuxer()->SetTotalAndBufferedBytesAndDuration(
+        kTotalBytes, kBufferedBytes, duration);
     EXPECT_CALL(*mocks_->demuxer(), GetNumberOfStreams())
         .WillRepeatedly(Return(streams->size()));
     EXPECT_CALL(*mocks_->demuxer(), SetPlaybackRate(0.0f));
@@ -217,8 +203,6 @@ class PipelineImplTest : public ::testing::Test {
 
   void ExpectSeek(const base::TimeDelta& seek_time) {
     // Every filter should receive a call to Seek().
-    EXPECT_CALL(*mocks_->data_source(), Seek(seek_time, NotNull()))
-        .WillOnce(Invoke(&RunFilterCallback));
     EXPECT_CALL(*mocks_->demuxer(), Seek(seek_time, NotNull()))
         .WillOnce(Invoke(&RunFilterCallback));
 
@@ -247,9 +231,9 @@ class PipelineImplTest : public ::testing::Test {
                                 &CallbackHelper::OnSeek));
 
     // We expect the time to be updated only after the seek has completed.
-    EXPECT_TRUE(seek_time != pipeline_->GetCurrentTime());
+    EXPECT_NE(seek_time, pipeline_->GetCurrentTime());
     message_loop_.RunAllPending();
-    EXPECT_TRUE(seek_time == pipeline_->GetCurrentTime());
+    EXPECT_EQ(seek_time, pipeline_->GetCurrentTime());
   }
 
   // Fixture members.
@@ -366,11 +350,6 @@ TEST_F(PipelineImplTest, URLNotFound) {
 TEST_F(PipelineImplTest, NoStreams) {
   // Manually set these expectations because SetPlaybackRate() is not called if
   // we cannot fully initialize the pipeline.
-  EXPECT_CALL(*mocks_->data_source(), Stop(NotNull()))
-      .WillOnce(Invoke(&RunStopFilterCallback));
-
-  EXPECT_CALL(*mocks_->demuxer(), Initialize(mocks_->data_source(), NotNull()))
-      .WillOnce(Invoke(&RunFilterCallback));
   EXPECT_CALL(*mocks_->demuxer(), GetNumberOfStreams())
       .WillRepeatedly(Return(0));
   EXPECT_CALL(*mocks_->demuxer(), Stop(NotNull()))
@@ -387,7 +366,6 @@ TEST_F(PipelineImplTest, AudioStream) {
   MockDemuxerStreamVector streams;
   streams.push_back(audio_stream());
 
-  InitializeDataSource();
   InitializeDemuxer(&streams, base::TimeDelta());
   InitializeAudioDecoder(audio_stream());
   InitializeAudioRenderer();
@@ -404,7 +382,6 @@ TEST_F(PipelineImplTest, VideoStream) {
   MockDemuxerStreamVector streams;
   streams.push_back(video_stream());
 
-  InitializeDataSource();
   InitializeDemuxer(&streams, base::TimeDelta());
   InitializeVideoDecoder(video_stream());
   InitializeVideoRenderer();
@@ -423,7 +400,6 @@ TEST_F(PipelineImplTest, AudioVideoStream) {
   streams.push_back(audio_stream());
   streams.push_back(video_stream());
 
-  InitializeDataSource();
   InitializeDemuxer(&streams, base::TimeDelta());
   InitializeAudioDecoder(audio_stream());
   InitializeAudioRenderer();
@@ -444,7 +420,6 @@ TEST_F(PipelineImplTest, Seek) {
   streams.push_back(audio_stream());
   streams.push_back(video_stream());
 
-  InitializeDataSource();
   InitializeDemuxer(&streams, base::TimeDelta::FromSeconds(3000));
   InitializeAudioDecoder(audio_stream());
   InitializeAudioRenderer();
@@ -465,7 +440,6 @@ TEST_F(PipelineImplTest, SetVolume) {
   MockDemuxerStreamVector streams;
   streams.push_back(audio_stream());
 
-  InitializeDataSource();
   InitializeDemuxer(&streams, base::TimeDelta());
   InitializeAudioDecoder(audio_stream());
   InitializeAudioRenderer();
@@ -484,7 +458,6 @@ TEST_F(PipelineImplTest, Properties) {
   MockDemuxerStreamVector streams;
   streams.push_back(video_stream());
 
-  InitializeDataSource();
   const base::TimeDelta kDuration = base::TimeDelta::FromSeconds(100);
   InitializeDemuxer(&streams, kDuration);
   InitializeVideoDecoder(video_stream());
@@ -509,7 +482,6 @@ TEST_F(PipelineImplTest, GetBufferedTime) {
   MockDemuxerStreamVector streams;
   streams.push_back(video_stream());
 
-  InitializeDataSource();
   const base::TimeDelta kDuration = base::TimeDelta::FromSeconds(100);
   InitializeDemuxer(&streams, kDuration);
   InitializeVideoDecoder(video_stream());
@@ -579,7 +551,6 @@ TEST_F(PipelineImplTest, DisableAudioRenderer) {
   streams.push_back(audio_stream());
   streams.push_back(video_stream());
 
-  InitializeDataSource();
   InitializeDemuxer(&streams, base::TimeDelta());
   InitializeAudioDecoder(audio_stream());
   InitializeAudioRenderer();
@@ -594,8 +565,6 @@ TEST_F(PipelineImplTest, DisableAudioRenderer) {
 
   EXPECT_CALL(*mocks_->audio_renderer(), SetPlaybackRate(1.0f))
       .WillOnce(DisableAudioRenderer(mocks_->audio_renderer()));
-  EXPECT_CALL(*mocks_->data_source(),
-              OnAudioRendererDisabled());
   EXPECT_CALL(*mocks_->demuxer(),
               OnAudioRendererDisabled());
   EXPECT_CALL(*mocks_->audio_decoder(),
@@ -624,15 +593,12 @@ TEST_F(PipelineImplTest, DisableAudioRendererDuringInit) {
   streams.push_back(audio_stream());
   streams.push_back(video_stream());
 
-  InitializeDataSource();
   InitializeDemuxer(&streams, base::TimeDelta());
   InitializeAudioDecoder(audio_stream());
   InitializeAudioRenderer(true);
   InitializeVideoDecoder(video_stream());
   InitializeVideoRenderer();
 
-  EXPECT_CALL(*mocks_->data_source(),
-              OnAudioRendererDisabled());
   EXPECT_CALL(*mocks_->demuxer(),
               OnAudioRendererDisabled());
   EXPECT_CALL(*mocks_->audio_decoder(),
@@ -665,7 +631,6 @@ TEST_F(PipelineImplTest, EndedCallback) {
   streams.push_back(audio_stream());
   streams.push_back(video_stream());
 
-  InitializeDataSource();
   InitializeDemuxer(&streams, base::TimeDelta());
   InitializeAudioDecoder(audio_stream());
   InitializeAudioRenderer();
@@ -711,7 +676,6 @@ TEST_F(PipelineImplTest, AudioStreamShorterThanVideo) {
   streams.push_back(audio_stream());
   streams.push_back(video_stream());
 
-  InitializeDataSource();
   InitializeDemuxer(&streams, duration);
   InitializeAudioDecoder(audio_stream());
   InitializeAudioRenderer();
@@ -729,7 +693,6 @@ TEST_F(PipelineImplTest, AudioStreamShorterThanVideo) {
   EXPECT_EQ(0, host->GetTime().ToInternalValue());
 
   float playback_rate = 1.0f;
-  EXPECT_CALL(*mocks_->data_source(), SetPlaybackRate(playback_rate));
   EXPECT_CALL(*mocks_->demuxer(), SetPlaybackRate(playback_rate));
   EXPECT_CALL(*mocks_->video_decoder(), SetPlaybackRate(playback_rate));
   EXPECT_CALL(*mocks_->audio_decoder(), SetPlaybackRate(playback_rate));
@@ -775,14 +738,12 @@ TEST_F(PipelineImplTest, ErrorDuringSeek) {
   MockDemuxerStreamVector streams;
   streams.push_back(audio_stream());
 
-  InitializeDataSource();
   InitializeDemuxer(&streams, base::TimeDelta::FromSeconds(10));
   InitializeAudioDecoder(audio_stream());
   InitializeAudioRenderer();
   InitializePipeline();
 
   float playback_rate = 1.0f;
-  EXPECT_CALL(*mocks_->data_source(), SetPlaybackRate(playback_rate));
   EXPECT_CALL(*mocks_->demuxer(), SetPlaybackRate(playback_rate));
   EXPECT_CALL(*mocks_->audio_decoder(), SetPlaybackRate(playback_rate));
   EXPECT_CALL(*mocks_->audio_renderer(), SetPlaybackRate(playback_rate));
@@ -792,8 +753,6 @@ TEST_F(PipelineImplTest, ErrorDuringSeek) {
   InSequence s;
 
   base::TimeDelta seek_time = base::TimeDelta::FromSeconds(5);
-  EXPECT_CALL(*mocks_->data_source(), Seek(seek_time, NotNull()))
-      .WillOnce(Invoke(&RunFilterCallback));
 
   EXPECT_CALL(*mocks_->demuxer(), Seek(seek_time, NotNull()))
       .WillOnce(DoAll(SetError(mocks_->demuxer(),