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

19 files changed, 350 insertions, 209 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(),
diff --git a/media/filters/ffmpeg_demuxer.cc b/media/filters/ffmpeg_demuxer.cc
index fe9f6ea..400ccf1 100644
--- a/media/filters/ffmpeg_demuxer.cc
+++ b/media/filters/ffmpeg_demuxer.cc
@@ -250,7 +250,9 @@ FFmpegDemuxer::FFmpegDemuxer(MessageLoop* message_loop)
       read_event_(false, false),
       read_has_failed_(false),
       last_read_bytes_(0),
-      read_position_(0) {
+      read_position_(0),
+      max_duration_(base::TimeDelta::FromMicroseconds(-1)),
+      deferred_status_(PIPELINE_OK) {
   DCHECK(message_loop_);
 }
 
@@ -313,8 +315,20 @@ void FFmpegDemuxer::OnAudioRendererDisabled() {
       NewRunnableMethod(this, &FFmpegDemuxer::DisableAudioStreamTask));
 }
 
+void FFmpegDemuxer::set_host(FilterHost* filter_host) {
+  Demuxer::set_host(filter_host);
+  if (data_source_)
+    data_source_->set_host(filter_host);
+  if (max_duration_.InMicroseconds() >= 0)
+    host()->SetDuration(max_duration_);
+  if (read_position_ > 0)
+    host()->SetCurrentReadPosition(read_position_);
+  if (deferred_status_ != PIPELINE_OK)
+    host()->SetError(deferred_status_);
+}
+
 void FFmpegDemuxer::Initialize(DataSource* data_source,
-                               FilterCallback* callback) {
+                               PipelineStatusCallback* callback) {
   message_loop_->PostTask(
       FROM_HERE,
       NewRunnableMethod(this,
@@ -356,7 +370,10 @@ int FFmpegDemuxer::Read(int size, uint8* data) {
   // let FFmpeg demuxer methods to run on.
   size_t last_read_bytes = WaitForRead();
   if (last_read_bytes == DataSource::kReadError) {
-    host()->SetError(PIPELINE_ERROR_READ);
+    if (host())
+      host()->SetError(PIPELINE_ERROR_READ);
+    else
+      deferred_status_ = PIPELINE_ERROR_READ;
 
     // Returns with a negative number to signal an error to FFmpeg.
     read_has_failed_ = true;
@@ -364,7 +381,8 @@ int FFmpegDemuxer::Read(int size, uint8* data) {
   }
   read_position_ += last_read_bytes;
 
-  host()->SetCurrentReadPosition(read_position_);
+  if (host())
+    host()->SetCurrentReadPosition(read_position_);
 
   return last_read_bytes;
 }
@@ -403,11 +421,13 @@ MessageLoop* FFmpegDemuxer::message_loop() {
 }
 
 void FFmpegDemuxer::InitializeTask(DataSource* data_source,
-                                   FilterCallback* callback) {
+                                   PipelineStatusCallback* callback) {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
-  scoped_ptr<FilterCallback> c(callback);
+  scoped_ptr<PipelineStatusCallback> callback_deleter(callback);
 
   data_source_ = data_source;
+  if (host())
+    data_source_->set_host(host());
 
   // Add ourself to Protocol list and get our unique key.
   std::string key = FFmpegGlue::GetInstance()->AddProtocol(this);
@@ -421,8 +441,7 @@ void FFmpegDemuxer::InitializeTask(DataSource* data_source,
   FFmpegGlue::GetInstance()->RemoveProtocol(this);
 
   if (result < 0) {
-    host()->SetError(DEMUXER_ERROR_COULD_NOT_OPEN);
-    callback->Run();
+    callback->Run(DEMUXER_ERROR_COULD_NOT_OPEN);
     return;
   }
 
@@ -432,8 +451,7 @@ void FFmpegDemuxer::InitializeTask(DataSource* data_source,
   // Fully initialize AVFormatContext by parsing the stream a little.
   result = av_find_stream_info(format_context_);
   if (result < 0) {
-    host()->SetError(DEMUXER_ERROR_COULD_NOT_PARSE);
-    callback->Run();
+    callback->Run(DEMUXER_ERROR_COULD_NOT_PARSE);
     return;
   }
 
@@ -464,8 +482,7 @@ void FFmpegDemuxer::InitializeTask(DataSource* data_source,
     }
   }
   if (streams_.empty()) {
-    host()->SetError(DEMUXER_ERROR_NO_SUPPORTED_STREAMS);
-    callback->Run();
+    callback->Run(DEMUXER_ERROR_NO_SUPPORTED_STREAMS);
     return;
   }
   if (format_context_->duration != static_cast<int64_t>(AV_NOPTS_VALUE)) {
@@ -483,8 +500,10 @@ void FFmpegDemuxer::InitializeTask(DataSource* data_source,
   }
 
   // Good to go: set the duration and notify we're done initializing.
-  host()->SetDuration(max_duration);
-  callback->Run();
+  if (host())
+    host()->SetDuration(max_duration);
+  max_duration_ = max_duration;
+  callback->Run(PIPELINE_OK);
 }
 
 void FFmpegDemuxer::SeekTask(base::TimeDelta time, FilterCallback* callback) {
@@ -572,7 +591,9 @@ void FFmpegDemuxer::StopTask(FilterCallback* callback) {
   for (iter = streams_.begin(); iter != streams_.end(); ++iter) {
     (*iter)->Stop();
   }
-  if (callback) {
+  if (data_source_) {
+    data_source_->Stop(callback);
+  } else {
     callback->Run();
     delete callback;
   }
diff --git a/media/filters/ffmpeg_demuxer.h b/media/filters/ffmpeg_demuxer.h
index 2ab78ae..e47be5e 100644
--- a/media/filters/ffmpeg_demuxer.h
+++ b/media/filters/ffmpeg_demuxer.h
@@ -30,6 +30,7 @@
 #include "base/synchronization/waitable_event.h"
 #include "media/base/buffers.h"
 #include "media/base/filters.h"
+#include "media/base/pipeline.h"
 #include "media/base/media_format.h"
 #include "media/filters/ffmpeg_glue.h"
 #include "media/filters/ffmpeg_interfaces.h"
@@ -127,13 +128,16 @@ class FFmpegDemuxer : public Demuxer,
   // Posts a task to perform additional demuxing.
   virtual void PostDemuxTask();
 
+  void Initialize(
+      DataSource* data_source, PipelineStatusCallback* callback);
+
   // Filter implementation.
   virtual void Stop(FilterCallback* callback);
   virtual void Seek(base::TimeDelta time, FilterCallback* callback);
   virtual void OnAudioRendererDisabled();
+  virtual void set_host(FilterHost* filter_host);
 
   // Demuxer implementation.
-  virtual void Initialize(DataSource* data_source, FilterCallback* callback);
   virtual size_t GetNumberOfStreams();
   virtual scoped_refptr<DemuxerStream> GetStream(int stream_id);
 
@@ -153,7 +157,8 @@ class FFmpegDemuxer : public Demuxer,
   FRIEND_TEST_ALL_PREFIXES(FFmpegDemuxerTest, ProtocolRead);
 
   // Carries out initialization on the demuxer thread.
-  void InitializeTask(DataSource* data_source, FilterCallback* callback);
+  void InitializeTask(
+      DataSource* data_source, PipelineStatusCallback* callback);
 
   // Carries out a seek on the demuxer thread.
   void SeekTask(base::TimeDelta time, FilterCallback* callback);
@@ -226,6 +231,11 @@ class FFmpegDemuxer : public Demuxer,
   size_t last_read_bytes_;
   int64 read_position_;
 
+  // Initialization can happen before set_host() is called, in which case we
+  // store these bits for deferred reporting to the FilterHost when we get one.
+  base::TimeDelta max_duration_;
+  PipelineError deferred_status_;
+
   DISALLOW_COPY_AND_ASSIGN(FFmpegDemuxer);
 };
 
diff --git a/media/filters/ffmpeg_demuxer_factory.cc b/media/filters/ffmpeg_demuxer_factory.cc
new file mode 100644
index 0000000..add2c54
--- /dev/null
+++ b/media/filters/ffmpeg_demuxer_factory.cc
@@ -0,0 +1,87 @@
+// 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 "base/message_loop.h"
+#include "media/filters/ffmpeg_demuxer.h"
+#include "media/filters/ffmpeg_demuxer_factory.h"
+
+namespace media {
+
+FFmpegDemuxerFactory::FFmpegDemuxerFactory(
+    DataSourceFactory* data_source_factory,
+    MessageLoop* loop)
+    : data_source_factory_(data_source_factory), loop_(loop) {}
+
+FFmpegDemuxerFactory::~FFmpegDemuxerFactory() {}
+
+// This and the next class are one-offs whose raison d'etre is the lack of
+// currying functionality in base/callback_old.h's machinery.  Once media/
+// PipelineStatusCallback and {DataSource,Demuxer}Factory::BuildCallback are
+// migrated to the new base/callback.h machinery these should be removed and
+// replaced with currying calls to base::Bind().
+class DemuxerCallbackAsPipelineStatusCallback : public PipelineStatusCallback {
+ public:
+  DemuxerCallbackAsPipelineStatusCallback(
+      DemuxerFactory::BuildCallback* cb,
+      Demuxer* demuxer)
+      : cb_(cb), demuxer_(demuxer) {
+    DCHECK(cb_.get() && demuxer_);
+  }
+
+  virtual ~DemuxerCallbackAsPipelineStatusCallback() {}
+
+  virtual void RunWithParams(const Tuple1<PipelineError>& params) {
+    cb_->Run(params.a, demuxer_);
+  }
+
+ private:
+  scoped_ptr<DemuxerFactory::BuildCallback> cb_;
+  scoped_refptr<Demuxer> demuxer_;
+
+  DISALLOW_IMPLICIT_CONSTRUCTORS(DemuxerCallbackAsPipelineStatusCallback);
+};
+
+// See comments on DemuxerCallbackAsPipelineStatusCallback above.
+class DemuxerCallbackAsDataSourceCallback :
+    public DataSourceFactory::BuildCallback {
+ public:
+  DemuxerCallbackAsDataSourceCallback(DemuxerFactory::BuildCallback* cb,
+                                      MessageLoop* loop)
+      : cb_(cb), loop_(loop) {
+    DCHECK(cb_.get() && loop_);
+  }
+
+  virtual ~DemuxerCallbackAsDataSourceCallback() {}
+
+  virtual void RunWithParams(const Tuple2<PipelineError, DataSource*>& params) {
+    PipelineError status = params.a;
+    DataSource* data_source = params.b;
+    if (status != PIPELINE_OK) {
+      cb_->Run(status, static_cast<Demuxer*>(NULL));
+      return;
+    }
+    DCHECK(data_source);
+    scoped_refptr<FFmpegDemuxer> demuxer = new FFmpegDemuxer(loop_);
+    demuxer->Initialize(
+        data_source,
+        new DemuxerCallbackAsPipelineStatusCallback(cb_.release(), demuxer));
+  }
+
+ private:
+  scoped_ptr<DemuxerFactory::BuildCallback> cb_;
+  MessageLoop* loop_;
+
+  DISALLOW_IMPLICIT_CONSTRUCTORS(DemuxerCallbackAsDataSourceCallback);
+};
+
+void FFmpegDemuxerFactory::Build(const std::string& url, BuildCallback* cb) {
+  data_source_factory_->Build(
+      url, new DemuxerCallbackAsDataSourceCallback(cb, loop_));
+}
+
+DemuxerFactory* FFmpegDemuxerFactory::Clone() const {
+  return new FFmpegDemuxerFactory(data_source_factory_->Clone(), loop_);
+}
+
+}  // namespace media
diff --git a/media/filters/ffmpeg_demuxer_factory.h b/media/filters/ffmpeg_demuxer_factory.h
new file mode 100644
index 0000000..8bc3d0d
--- /dev/null
+++ b/media/filters/ffmpeg_demuxer_factory.h
@@ -0,0 +1,37 @@
+// 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.
+
+// Implements the DemuxerFactory interface using FFmpegDemuxer.
+
+#ifndef MEDIA_FILTERS_FFMPEG_DEMUXER_FACTORY_H_
+#define MEDIA_FILTERS_FFMPEG_DEMUXER_FACTORY_H_
+
+#include "base/basictypes.h"
+#include "base/scoped_ptr.h"
+#include "media/base/filter_factories.h"
+
+class MessageLoop;
+
+namespace media {
+
+class FFmpegDemuxerFactory : public DemuxerFactory {
+ public:
+  // Takes ownership of |data_source_factory|, but not of |loop|.
+  FFmpegDemuxerFactory(DataSourceFactory* data_source_factory,
+                       MessageLoop* loop);
+  virtual ~FFmpegDemuxerFactory();
+
+  virtual void Build(const std::string& url, BuildCallback* cb);
+  virtual DemuxerFactory* Clone() const;
+
+ private:
+  scoped_ptr<DataSourceFactory> data_source_factory_;
+  MessageLoop* loop_;  // Unowned.
+
+  DISALLOW_COPY_AND_ASSIGN(FFmpegDemuxerFactory);
+};
+
+}  // namespace media
+
+#endif  // MEDIA_FILTERS_FFMPEG_DEMUXER_FACTORY_H_
diff --git a/media/filters/ffmpeg_demuxer_unittest.cc b/media/filters/ffmpeg_demuxer_unittest.cc
index 50ebb1e..96be01d 100644
--- a/media/filters/ffmpeg_demuxer_unittest.cc
+++ b/media/filters/ffmpeg_demuxer_unittest.cc
@@ -72,6 +72,9 @@ class FFmpegDemuxerTest : public testing::Test {
     demuxer_->set_host(&host_);
     data_source_ = new StrictMock<MockDataSource>();
 
+    EXPECT_CALL(*data_source_, Stop(NotNull()))
+        .WillRepeatedly(Invoke(&RunStopFilterCallback));
+
     // Initialize FFmpeg fixtures.
     memset(&format_context_, 0, sizeof(format_context_));
     memset(&input_format_, 0, sizeof(input_format_));
@@ -137,7 +140,8 @@ class FFmpegDemuxerTest : public testing::Test {
         base::TimeDelta::FromMicroseconds(kDurations[AV_STREAM_AUDIO]);
     EXPECT_CALL(host_, SetDuration(expected_duration));
 
-    demuxer_->Initialize(data_source_.get(), NewExpectedCallback());
+    demuxer_->Initialize(data_source_.get(),
+                         NewExpectedStatusCallback(PIPELINE_OK));
     message_loop_.RunAllPending();
   }
 
@@ -175,9 +179,9 @@ TEST_F(FFmpegDemuxerTest, Initialize_OpenFails) {
   // Simulate av_open_input_file() failing.
   EXPECT_CALL(mock_ffmpeg_, AVOpenInputFile(_, _, NULL, 0, NULL))
       .WillOnce(Return(-1));
-  EXPECT_CALL(host_, SetError(DEMUXER_ERROR_COULD_NOT_OPEN));
 
-  demuxer_->Initialize(data_source_.get(), NewExpectedCallback());
+  demuxer_->Initialize(data_source_.get(),
+                       NewExpectedStatusCallback(DEMUXER_ERROR_COULD_NOT_OPEN));
   message_loop_.RunAllPending();
 }
 
@@ -188,9 +192,10 @@ TEST_F(FFmpegDemuxerTest, Initialize_ParseFails) {
   EXPECT_CALL(mock_ffmpeg_, AVFindStreamInfo(&format_context_))
       .WillOnce(Return(AVERROR_IO));
   EXPECT_CALL(mock_ffmpeg_, AVCloseInputFile(&format_context_));
-  EXPECT_CALL(host_, SetError(DEMUXER_ERROR_COULD_NOT_PARSE));
 
-  demuxer_->Initialize(data_source_.get(), NewExpectedCallback());
+  demuxer_->Initialize(
+      data_source_.get(),
+      NewExpectedStatusCallback(DEMUXER_ERROR_COULD_NOT_PARSE));
   message_loop_.RunAllPending();
 }
 
@@ -200,10 +205,11 @@ TEST_F(FFmpegDemuxerTest, Initialize_NoStreams) {
     SCOPED_TRACE("");
     InitializeDemuxerMocks();
   }
-  EXPECT_CALL(host_, SetError(DEMUXER_ERROR_NO_SUPPORTED_STREAMS));
   format_context_.nb_streams = 0;
 
-  demuxer_->Initialize(data_source_.get(), NewExpectedCallback());
+  demuxer_->Initialize(
+      data_source_.get(),
+      NewExpectedStatusCallback(DEMUXER_ERROR_NO_SUPPORTED_STREAMS));
   message_loop_.RunAllPending();
 }
 
@@ -213,11 +219,12 @@ TEST_F(FFmpegDemuxerTest, Initialize_DataStreamOnly) {
     SCOPED_TRACE("");
     InitializeDemuxerMocks();
   }
-  EXPECT_CALL(host_, SetError(DEMUXER_ERROR_NO_SUPPORTED_STREAMS));
   EXPECT_EQ(format_context_.streams[0], &streams_[AV_STREAM_DATA]);
   format_context_.nb_streams = 1;
 
-  demuxer_->Initialize(data_source_.get(), NewExpectedCallback());
+  demuxer_->Initialize(
+      data_source_.get(),
+      NewExpectedStatusCallback(DEMUXER_ERROR_NO_SUPPORTED_STREAMS));
   message_loop_.RunAllPending();
 }
 
@@ -548,7 +555,11 @@ TEST_F(FFmpegDemuxerTest, Stop) {
   scoped_refptr<DemuxerStream> audio = demuxer_->GetStream(DS_STREAM_AUDIO);
   ASSERT_TRUE(audio);
 
-  // Stop the demuxer.
+  // Stop the demuxer, overriding the default expectation to assert that
+  // data_source_ really is Stop()'d.
+  EXPECT_CALL(*data_source_, Stop(_))
+      .WillOnce(Invoke(&RunStopFilterCallback))
+      .RetiresOnSaturation();
   demuxer_->Stop(NewExpectedCallback());
 
   // Expect all calls in sequence.
diff --git a/media/media.gyp b/media/media.gyp
index 933f513..311e24f 100644
--- a/media/media.gyp
+++ b/media/media.gyp
@@ -153,6 +153,8 @@
         'filters/ffmpeg_video_decoder.h',
         'filters/file_data_source.cc',
         'filters/file_data_source.h',
+        'filters/ffmpeg_demuxer_factory.cc',
+        'filters/ffmpeg_demuxer_factory.h',
         'filters/file_data_source_factory.cc',
         'filters/file_data_source_factory.h',
         'filters/null_audio_renderer.cc',
diff --git a/media/tools/player_wtl/movie.cc b/media/tools/player_wtl/movie.cc
index 371d178..042d1f4 100644
--- a/media/tools/player_wtl/movie.cc
+++ b/media/tools/player_wtl/movie.cc
@@ -12,7 +12,7 @@
 #include "media/base/pipeline_impl.h"
 #include "media/filters/audio_renderer_impl.h"
 #include "media/filters/ffmpeg_audio_decoder.h"
-#include "media/filters/ffmpeg_demuxer.h"
+#include "media/filters/ffmpeg_demuxer_factory.h"
 #include "media/filters/ffmpeg_video_decoder.h"
 #include "media/filters/file_data_source_factory.h"
 #include "media/filters/null_audio_renderer.h"
@@ -20,7 +20,7 @@
 
 using media::AudioRendererImpl;
 using media::FFmpegAudioDecoder;
-using media::FFmpegDemuxer;
+using media::FFmpegDemuxerFactory;
 using media::FFmpegVideoDecoder;
 using media::FileDataSourceFactory;
 using media::FilterCollection;
@@ -63,13 +63,16 @@ bool Movie::Open(const wchar_t* url, WtlVideoRenderer* video_renderer) {
 
   message_loop_factory_.reset(new media::MessageLoopFactoryImpl());
 
+  MessageLoop* pipeline_loop =
+      message_loop_factory_->GetMessageLoop("PipelineThread");
+  pipeline_ = new PipelineImpl(pipeline_loop);
+
   // Create filter collection.
   scoped_ptr<FilterCollection> collection(new FilterCollection());
-  collection->SetDataSourceFactory(new FileDataSourceFactory());
+  collection->SetDemuxerFactory(new FFmpegDemuxerFactory(
+      new FileDataSourceFactory(), pipeline_loop));
   collection->AddAudioDecoder(new FFmpegAudioDecoder(
       message_loop_factory_->GetMessageLoop("AudioDecoderThread")));
-  collection->AddDemuxer(new FFmpegDemuxer(
-      message_loop_factory_->GetMessageLoop("DemuxThread")));
   collection->AddVideoDecoder(new FFmpegVideoDecoder(
       message_loop_factory_->GetMessageLoop("VideoDecoderThread"), NULL));
 
@@ -80,9 +83,6 @@ bool Movie::Open(const wchar_t* url, WtlVideoRenderer* video_renderer) {
   }
   collection->AddVideoRenderer(video_renderer);
 
-  pipeline_ = new PipelineImpl(
-      message_loop_factory_->GetMessageLoop("PipelineThread"));
-
   // Create and start our pipeline.
   pipeline_->Start(collection.release(), WideToUTF8(std::wstring(url)), NULL);
   while (true) {
diff --git a/media/tools/player_x11/player_x11.cc b/media/tools/player_x11/player_x11.cc
index 68c56ae..84e08e6 100644
--- a/media/tools/player_x11/player_x11.cc
+++ b/media/tools/player_x11/player_x11.cc
@@ -21,7 +21,7 @@
 #include "media/base/pipeline_impl.h"
 #include "media/filters/audio_renderer_impl.h"
 #include "media/filters/ffmpeg_audio_decoder.h"
-#include "media/filters/ffmpeg_demuxer.h"
+#include "media/filters/ffmpeg_demuxer_factory.h"
 #include "media/filters/ffmpeg_video_decoder.h"
 #include "media/filters/file_data_source_factory.h"
 #include "media/filters/null_audio_renderer.h"
@@ -103,9 +103,8 @@ bool InitPipeline(MessageLoop* message_loop,
   // Create our filter factories.
   scoped_ptr<media::FilterCollection> collection(
       new media::FilterCollection());
-  collection->SetDataSourceFactory(new media::FileDataSourceFactory());
-  collection->AddDemuxer(new media::FFmpegDemuxer(
-      message_loop_factory->GetMessageLoop("DemuxThread")));
+  collection->SetDemuxerFactory(new media::FFmpegDemuxerFactory(
+      new media::FileDataSourceFactory(), message_loop));
   collection->AddAudioDecoder(new media::FFmpegAudioDecoder(
       message_loop_factory->GetMessageLoop("AudioDecoderThread")));
   if (CommandLine::ForCurrentProcess()->HasSwitch(
@@ -127,7 +126,7 @@ bool InitPipeline(MessageLoop* message_loop,
   else
     collection->AddAudioRenderer(new media::NullAudioRenderer());
 
-  // Creates the pipeline and start it.
+  // Create and start the pipeline.
   *pipeline = new media::PipelineImpl(message_loop);
   (*pipeline)->Start(collection.release(), filename, NULL);
 
@@ -137,6 +136,7 @@ bool InitPipeline(MessageLoop* message_loop,
     if ((*pipeline)->IsInitialized())
       break;
     if ((*pipeline)->GetError() != media::PIPELINE_OK) {
+      std::cout << "InitPipeline: " << (*pipeline)->GetError() << std::endl;
       (*pipeline)->Stop(NULL);
       return false;
     }
diff --git a/webkit/glue/webmediaplayer_impl.cc b/webkit/glue/webmediaplayer_impl.cc
index 2d5e6a6..098cd61 100644
--- a/webkit/glue/webmediaplayer_impl.cc
+++ b/webkit/glue/webmediaplayer_impl.cc
@@ -16,7 +16,7 @@
 #include "media/base/pipeline_impl.h"
 #include "media/base/video_frame.h"
 #include "media/filters/ffmpeg_audio_decoder.h"
-#include "media/filters/ffmpeg_demuxer.h"
+#include "media/filters/ffmpeg_demuxer_factory.h"
 #include "media/filters/ffmpeg_video_decoder.h"
 #include "media/filters/null_audio_renderer.h"
 #include "skia/ext/platform_canvas.h"
@@ -335,11 +335,12 @@ bool WebMediaPlayerImpl::Initialize(
     data_source_factory->AddFactory(simple_data_source_factory.release());
   }
 
-  filter_collection_->SetDataSourceFactory(data_source_factory.release());
+  scoped_ptr<media::DemuxerFactory> demuxer_factory(
+      new media::FFmpegDemuxerFactory(data_source_factory.release(),
+                                      pipeline_message_loop));
+  filter_collection_->SetDemuxerFactory(demuxer_factory.release());
 
   // Add in the default filter factories.
-  filter_collection_->AddDemuxer(new media::FFmpegDemuxer(
-      message_loop_factory_->GetMessageLoop("DemuxThread")));
   filter_collection_->AddAudioDecoder(new media::FFmpegAudioDecoder(
       message_loop_factory_->GetMessageLoop("AudioDecoderThread")));
   filter_collection_->AddVideoDecoder(new media::FFmpegVideoDecoder(