Refactoring code to use factories to create DataSource objects.

BUG=72485
TEST=None for now. Existing unit tests cover this code.

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

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

28 files changed, 843 insertions, 177 deletions

diff --git a/media/base/async_filter_factory_base.cc b/media/base/async_filter_factory_base.cc
new file mode 100644
index 0000000..bbc5160
--- /dev/null
+++ b/media/base/async_filter_factory_base.cc
@@ -0,0 +1,112 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "media/base/async_filter_factory_base.h"
+
+#include "base/logging.h"
+#include "base/stl_util-inl.h"
+
+namespace media {
+
+AsyncDataSourceFactoryBase::AsyncDataSourceFactoryBase() {}
+
+AsyncDataSourceFactoryBase::~AsyncDataSourceFactoryBase() {
+  base::AutoLock auto_lock(lock_);
+  STLDeleteElements(&outstanding_requests_);
+}
+
+void AsyncDataSourceFactoryBase::Build(const std::string& url,
+                                       BuildCallback* callback) {
+  DCHECK(callback);
+  BuildRequest* request = NULL;
+  {
+    base::AutoLock auto_lock(lock_);
+
+    if (url.empty()) {
+      RunAndDestroyCallback(PIPELINE_ERROR_URL_NOT_FOUND, callback);
+      return;
+    }
+
+    if (!AllowRequests()) {
+      RunAndDestroyCallback(DATASOURCE_ERROR_URL_NOT_SUPPORTED, callback);
+      return;
+    }
+
+    request = CreateRequest(url, callback);
+
+    if (!request) {
+      RunAndDestroyCallback(DATASOURCE_ERROR_URL_NOT_SUPPORTED, callback);
+      return;
+    }
+
+    outstanding_requests_.insert(request);
+  }
+
+  request->Start(NewCallback(this,
+                             &AsyncDataSourceFactoryBase::BuildRequestDone));
+}
+
+void AsyncDataSourceFactoryBase::RunAndDestroyCallback(
+    PipelineError error,
+    BuildCallback* callback) const {
+  DCHECK_NE(error, PIPELINE_OK);
+  DCHECK(callback);
+
+  callback->Run(error, static_cast<DataSource*>(NULL));
+  delete callback;
+}
+
+void AsyncDataSourceFactoryBase::BuildRequestDone(BuildRequest* request) {
+  base::AutoLock auto_lock(lock_);
+  outstanding_requests_.erase(request);
+  delete request;
+}
+
+AsyncDataSourceFactoryBase::BuildRequest::BuildRequest(const std::string& url,
+                                                       BuildCallback* callback)
+    : url_(url),
+      callback_(callback) {
+}
+
+AsyncDataSourceFactoryBase::BuildRequest::~BuildRequest() {}
+
+void AsyncDataSourceFactoryBase::BuildRequest::Start(
+    RequestDoneCallback* done_callback) {
+  DCHECK(done_callback);
+  DCHECK(!done_callback_.get());
+
+  done_callback_.reset(done_callback);
+  DoStart();
+  // Don't do anything after this line since the object could
+  // have been deleted at this point if the request was completed
+  // inside the call.
+}
+
+void AsyncDataSourceFactoryBase::BuildRequest::RequestComplete(
+    PipelineError error,
+    DataSource* data_source) {
+  DCHECK(callback_.get());
+  DCHECK(done_callback_.get());
+
+  // Transfer ownership to local variables just in case the
+  // request object gets deleted by one of the callbacks.
+  scoped_ptr<RequestDoneCallback> done_callback(done_callback_.release());
+  scoped_ptr<BuildCallback> callback(callback_.release());
+
+  // Notify factory that this request has completed. We do this before
+  // calling |callback| so the factory doesn't consider this request
+  // pending if |callback| happens to destroy the factory.
+  //
+  // NOTE: This BuildRequest object is destroyed inside this callback so
+  //       no modifications should be made to this object after this call.
+  done_callback->Run(this);
+
+  callback->Run(error, data_source);
+}
+
+const std::string& AsyncDataSourceFactoryBase::BuildRequest::url() const {
+  return url_;
+}
+
+}  // namespace media
diff --git a/media/base/async_filter_factory_base.h b/media/base/async_filter_factory_base.h
new file mode 100644
index 0000000..75617f4
--- /dev/null
+++ b/media/base/async_filter_factory_base.h
@@ -0,0 +1,129 @@
+// 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.
+
+#ifndef MEDIA_BASE_ASYNC_FILTER_FACTORY_BASE_H_
+#define MEDIA_BASE_ASYNC_FILTER_FACTORY_BASE_H_
+
+#include <set>
+
+#include "base/scoped_ptr.h"
+#include "base/synchronization/lock.h"
+#include "media/base/filter_factories.h"
+
+namespace media {
+
+// This is a helper base class for DataSourceFactory implementation that
+// actually require asynchronous operations to build a data source. It
+// provides a common framework for dealing with an asychronous build.
+// Factories are expected to derive from this object and implement
+// AllowRequests(), CreateRequest(), and an custom implementation that derives
+// from BuildRequest.
+//
+// AllowRequests() just checks to see if the factory is in a state where it can
+// accept Build() requests. If it returns false, this base class will signal an
+// error through the BuildCallback and not call any of the other code. If
+// AllowRequests() returns true then this class will continue with the build
+// process by calling CreateRequest().
+//
+// CreateRequest() allows the derived implementation to create an instance of
+// their custom BuildRequest implementation that is specific to its asynchronous
+// build process. The custom BuildRequest should contain all of the state that
+// needs to be maintained for a particular build request. This state could
+// consist of partially initialized objects that require asynchronous operations
+// to complete before the build completes. This implementation MUST derive from
+// AsyncDataSourceFactoryBase::BuildRequest.
+//
+// Once CreateRequest() returns a BuildRequest implementation, this class adds
+// the object to its request list and then calls Start() on the BuildRequest
+// instance. BuildRequest::Start() manages storing the |done_callback| passed to
+// it and then call DoStart() on the derived object. DoStart() is where this
+// framework expects any neccesary asynchronous operations to be initiated.
+//
+// Once all asynchronous operations are completed and a fully initialized
+// DataSource object has been created, the BuildRequest instance should call
+// the RequestComplete() method. This call signals the end of the request and
+// the BuildRequest should be in a state where it can be deleted from inside
+// this call. If an error occurs during the build process, RequestComplete()
+// can also be called to signal the error.
+class AsyncDataSourceFactoryBase : public DataSourceFactory {
+ public:
+  AsyncDataSourceFactoryBase();
+  virtual ~AsyncDataSourceFactoryBase();
+
+  // DataSourceFactory method.
+  // Derived classes should not overload this Build() method. AllowRequests() &
+  // CreateRequest() should be implemented instead.
+  virtual void Build(const std::string& url, BuildCallback* callback);
+
+  // DataSourceFactory method.
+  // Clone() must be implemented by derived classes.
+  // NOTE: Nothing in this base class needs to be cloned because this class
+  // only keeps track of pending requests, which are not part of the cloning
+  // process.
+  virtual DataSourceFactory* Clone() const = 0;
+
+ protected:
+  class BuildRequest {
+   public:
+    BuildRequest(const std::string& url, BuildCallback* callback);
+    virtual ~BuildRequest();
+
+    typedef Callback1<BuildRequest*>::Type RequestDoneCallback;
+    // Starts the build request.
+    void Start(RequestDoneCallback* done_callback);
+
+    // Derived objects call this method to indicate that the build request
+    // has completed. If the build was successful |error| should be set to
+    // PIPELINE_OK and |data_source| should contain the DataSource object
+    // that was built by this request. Ownership of |data_source| is being
+    // passed in this call. If an error occurs during the build process, this
+    // method should be called with |error| set to an appropriate status code
+    // and |data_source| set to NULL.
+    //
+    // The derived object should be in a state where it can be deleted from
+    // within this call. This class as well AsyncDataSourceFactoryBase use this
+    // method to cleanup state associated with this request.
+    void RequestComplete(media::PipelineError error, DataSource* data_source);
+
+   protected:
+    // Implemented by the derived object to start the build. Called by Start().
+    virtual void DoStart() = 0;
+
+    // Gets the requested URL.
+    const std::string& url() const;
+
+   private:
+    std::string url_;
+    scoped_ptr<BuildCallback> callback_;
+    scoped_ptr<RequestDoneCallback> done_callback_;
+
+    DISALLOW_COPY_AND_ASSIGN(BuildRequest);
+  };
+
+  // Implemented by derived class. Called by Build() to check if the
+  // factory is in a state where it can accept requests.
+  virtual bool AllowRequests() const = 0;
+
+  // Implemented by derived class. Called by Build() to allow derived objects
+  // to create their own custom BuildRequest implementations.
+  virtual BuildRequest* CreateRequest(const std::string& url,
+                                      BuildCallback* callback) = 0;
+
+ private:
+  void RunAndDestroyCallback(PipelineError error,
+                             BuildCallback* callback) const;
+
+  typedef Callback1<BuildRequest*>::Type RequestDoneCallback;
+  void BuildRequestDone(BuildRequest* request);
+
+  base::Lock lock_;
+  typedef std::set<BuildRequest*> RequestSet;
+  RequestSet outstanding_requests_;
+
+  DISALLOW_COPY_AND_ASSIGN(AsyncDataSourceFactoryBase);
+};
+
+}  // namespace media
+
+#endif  // MEDIA_BASE_ASYNC_FILTER_FACTORY_BASE_H_
diff --git a/media/base/composite_data_source_factory.cc b/media/base/composite_data_source_factory.cc
new file mode 100644
index 0000000..84b5259
--- /dev/null
+++ b/media/base/composite_data_source_factory.cc
@@ -0,0 +1,108 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "media/base/composite_data_source_factory.h"
+
+#include "base/callback.h"
+#include "base/logging.h"
+#include "base/scoped_ptr.h"
+#include "base/stl_util-inl.h"
+
+namespace media {
+
+class CompositeDataSourceFactory::BuildRequest
+    : public AsyncDataSourceFactoryBase::BuildRequest {
+ public:
+  BuildRequest(const std::string& url, BuildCallback* callback,
+               const FactoryList& factories);
+  ~BuildRequest();
+
+ protected:
+  // AsyncDataSourceFactoryBase::BuildRequest method.
+  virtual void DoStart();
+
+ private:
+  void CallNextFactory();
+  void OnBuildDone(PipelineError error, DataSource* data_source);
+
+  FactoryList factories_;
+};
+
+CompositeDataSourceFactory::CompositeDataSourceFactory() {}
+
+CompositeDataSourceFactory::~CompositeDataSourceFactory() {
+  STLDeleteElements(&factories_);
+}
+
+void CompositeDataSourceFactory::AddFactory(DataSourceFactory* factory) {
+  DCHECK(factory);
+  factories_.push_back(factory);
+}
+
+DataSourceFactory* CompositeDataSourceFactory::Clone() const {
+  CompositeDataSourceFactory* new_factory = new CompositeDataSourceFactory();
+
+  for (FactoryList::const_iterator itr = factories_.begin();
+       itr != factories_.end();
+       ++itr) {
+    new_factory->AddFactory((*itr)->Clone());
+  }
+
+  return new_factory;
+}
+
+bool CompositeDataSourceFactory::AllowRequests() const {
+  return !factories_.empty();
+}
+
+AsyncDataSourceFactoryBase::BuildRequest*
+CompositeDataSourceFactory::CreateRequest(const std::string& url,
+                                          BuildCallback* callback) {
+  return new BuildRequest(url, callback, factories_);
+}
+
+CompositeDataSourceFactory::BuildRequest::BuildRequest(
+    const std::string& url,
+    BuildCallback* callback,
+    const FactoryList& factories)
+    : AsyncDataSourceFactoryBase::BuildRequest(url, callback),
+      factories_(factories){
+  DCHECK(!factories.empty());
+}
+
+CompositeDataSourceFactory::BuildRequest::~BuildRequest() {}
+
+void CompositeDataSourceFactory::BuildRequest::DoStart() {
+  CallNextFactory();
+}
+
+void CompositeDataSourceFactory::BuildRequest::CallNextFactory() {
+  DCHECK(!factories_.empty());
+
+  DataSourceFactory* factory = factories_.front();
+  factories_.pop_front();
+
+  factory->Build(url(), NewCallback(this, &BuildRequest::OnBuildDone));
+}
+
+void CompositeDataSourceFactory::BuildRequest::OnBuildDone(
+    PipelineError error,
+    DataSource* data_source) {
+
+  if (error == PIPELINE_OK) {
+    DCHECK(data_source);
+    RequestComplete(error, data_source);
+    return;
+  }
+
+  DCHECK(!data_source);
+  if ((error == DATASOURCE_ERROR_URL_NOT_SUPPORTED) && !factories_.empty()) {
+    CallNextFactory();
+    return;
+  }
+
+  RequestComplete(error, data_source);
+}
+
+}  // namespace media
diff --git a/media/base/composite_data_source_factory.h b/media/base/composite_data_source_factory.h
new file mode 100644
index 0000000..799960d
--- /dev/null
+++ b/media/base/composite_data_source_factory.h
@@ -0,0 +1,44 @@
+// 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.
+
+#ifndef MEDIA_BASE_COMPOSITE_DATA_SOURCE_FACTORY_H_
+#define MEDIA_BASE_COMPOSITE_DATA_SOURCE_FACTORY_H_
+
+#include <list>
+#include <set>
+
+#include "base/synchronization/lock.h"
+#include "media/base/async_filter_factory_base.h"
+
+namespace media {
+
+class CompositeDataSourceFactory : public AsyncDataSourceFactoryBase {
+ public:
+  CompositeDataSourceFactory();
+  virtual ~CompositeDataSourceFactory();
+
+  // Add factory to this composite. Ownership is transferred here.
+  void AddFactory(DataSourceFactory* factory);
+
+  // DataSourceFactory method.
+  virtual DataSourceFactory* Clone() const;
+
+ protected:
+  // AsyncDataSourceFactoryBase methods.
+  virtual bool AllowRequests() const;
+  virtual AsyncDataSourceFactoryBase::BuildRequest* CreateRequest(
+      const std::string& url, BuildCallback* callback);
+
+ private:
+  class BuildRequest;
+
+  typedef std::list<DataSourceFactory*> FactoryList;
+  FactoryList factories_;
+
+  DISALLOW_COPY_AND_ASSIGN(CompositeDataSourceFactory);
+};
+
+}  // namespace media
+
+#endif  // MEDIA_BASE_COMPOSITE_DATA_SOURCE_FACTORY_H_
diff --git a/media/base/filter_collection.cc b/media/base/filter_collection.cc
index b236b02..c25fa7a 100644
--- a/media/base/filter_collection.cc
+++ b/media/base/filter_collection.cc
@@ -1,17 +1,24 @@
-// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "media/base/filter_collection.h"
 
+#include "base/logging.h"
+
 namespace media {
 
 FilterCollection::FilterCollection() {}
 
 FilterCollection::~FilterCollection() {}
 
-void FilterCollection::AddDataSource(DataSource* filter) {
-  AddFilter(DATA_SOURCE, filter);
+void FilterCollection::SetDataSourceFactory(DataSourceFactory* factory) {
+  DCHECK(factory);
+  data_source_factory_.reset(factory);
+}
+
+DataSourceFactory* FilterCollection::GetDataSourceFactory() {
+  return data_source_factory_.get();
 }
 
 void FilterCollection::AddDemuxer(Demuxer* filter) {
@@ -42,11 +49,6 @@ void FilterCollection::Clear() {
   filters_.clear();
 }
 
-void FilterCollection::SelectDataSource(
-    scoped_refptr<DataSource>* filter_out) {
-  SelectFilter<DATA_SOURCE>(filter_out);
-}
-
 void FilterCollection::SelectDemuxer(scoped_refptr<Demuxer>* filter_out) {
   SelectFilter<DEMUXER>(filter_out);
 }
diff --git a/media/base/filter_collection.h b/media/base/filter_collection.h
index 8dd5b04..6fc445d 100644
--- a/media/base/filter_collection.h
+++ b/media/base/filter_collection.h
@@ -1,4 +1,4 @@
-// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// 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.
 
@@ -9,6 +9,7 @@
 
 #include "base/ref_counted.h"
 #include "media/base/filters.h"
+#include "media/base/filter_factories.h"
 
 namespace media {
 
@@ -19,8 +20,12 @@ class FilterCollection {
   FilterCollection();
   ~FilterCollection();
 
+  // DataSourceFactory accessor methods.
+  // FilterCollection takes ownership of the factory here.
+  void SetDataSourceFactory(DataSourceFactory* factory);
+  DataSourceFactory* GetDataSourceFactory();
+
   // Adds a filter to the collection.
-  void AddDataSource(DataSource* filter);
   void AddDemuxer(Demuxer* filter);
   void AddVideoDecoder(VideoDecoder* filter);
   void AddAudioDecoder(AudioDecoder* filter);
@@ -36,7 +41,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 SelectDataSource(scoped_refptr<DataSource>* filter_out);
   void SelectDemuxer(scoped_refptr<Demuxer>* filter_out);
   void SelectVideoDecoder(scoped_refptr<VideoDecoder>* filter_out);
   void SelectAudioDecoder(scoped_refptr<AudioDecoder>* filter_out);
@@ -48,7 +52,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 {
-    DATA_SOURCE,
     DEMUXER,
     AUDIO_DECODER,
     VIDEO_DECODER,
@@ -60,6 +63,7 @@ class FilterCollection {
   typedef std::pair<FilterType, scoped_refptr<Filter> > FilterListElement;
   typedef std::list<FilterListElement> FilterList;
   FilterList filters_;
+  scoped_ptr<DataSourceFactory> data_source_factory_;
 
   // Helper function that adds a filter to the filter list.
   void AddFilter(FilterType filter_type, Filter* filter);
diff --git a/media/base/filter_collection_unittest.cc b/media/base/filter_collection_unittest.cc
index fef7e7f7..70624a1 100644
--- a/media/base/filter_collection_unittest.cc
+++ b/media/base/filter_collection_unittest.cc
@@ -1,4 +1,4 @@
-// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// 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.
 
@@ -23,7 +23,7 @@ class FilterCollectionTest : public ::testing::Test {
 TEST_F(FilterCollectionTest, TestIsEmptyAndClear) {
   EXPECT_TRUE(collection_.IsEmpty());
 
-  collection_.AddDataSource(mock_filters_.data_source());
+  collection_.AddAudioDecoder(mock_filters_.audio_decoder());
 
   EXPECT_FALSE(collection_.IsEmpty());
 
@@ -34,25 +34,25 @@ TEST_F(FilterCollectionTest, TestIsEmptyAndClear) {
 
 TEST_F(FilterCollectionTest, SelectXXXMethods) {
   scoped_refptr<AudioDecoder> audio_decoder;
-  scoped_refptr<DataSource> data_source;
+  scoped_refptr<VideoDecoder> video_decoder;
 
-  collection_.AddDataSource(mock_filters_.data_source());
+  collection_.AddVideoDecoder(mock_filters_.video_decoder());
   EXPECT_FALSE(collection_.IsEmpty());
 
-  // Verify that the data source will not be returned if we
+  // Verify that the video decoder will not be returned if we
   // ask for a different type.
   collection_.SelectAudioDecoder(&audio_decoder);
   EXPECT_FALSE(audio_decoder);
   EXPECT_FALSE(collection_.IsEmpty());
 
-  // Verify that we can actually retrieve the data source
+  // Verify that we can actually retrieve the video decoder
   // and that it is removed from the collection.
-  collection_.SelectDataSource(&data_source);
-  EXPECT_TRUE(data_source);
+  collection_.SelectVideoDecoder(&video_decoder);
+  EXPECT_TRUE(video_decoder);
   EXPECT_TRUE(collection_.IsEmpty());
 
-  // Add a data source and audio decoder.
-  collection_.AddDataSource(mock_filters_.data_source());
+  // Add a video decoder and audio decoder.
+  collection_.AddVideoDecoder(mock_filters_.video_decoder());
   collection_.AddAudioDecoder(mock_filters_.audio_decoder());
 
   // Verify that we can select the audio decoder.
@@ -64,36 +64,36 @@ TEST_F(FilterCollectionTest, SelectXXXMethods) {
   collection_.SelectAudioDecoder(&audio_decoder);
   EXPECT_FALSE(audio_decoder);
 
-  // Verify that we can select the data source and that doing so will
+  // Verify that we can select the video decoder and that doing so will
   // empty the collection again.
-  collection_.SelectDataSource(&data_source);
+  collection_.SelectVideoDecoder(&video_decoder);
   EXPECT_TRUE(collection_.IsEmpty());
 }
 
 TEST_F(FilterCollectionTest, MultipleFiltersOfSameType) {
-  scoped_refptr<DataSource> data_source_a(new MockDataSource());
-  scoped_refptr<DataSource> data_source_b(new MockDataSource());
+  scoped_refptr<AudioDecoder> audio_decoder_a(new MockAudioDecoder());
+  scoped_refptr<AudioDecoder> audio_decoder_b(new MockAudioDecoder());
 
-  scoped_refptr<DataSource> data_source;
+  scoped_refptr<AudioDecoder> audio_decoder;
 
-  collection_.AddDataSource(data_source_a.get());
-  collection_.AddDataSource(data_source_b.get());
+  collection_.AddAudioDecoder(audio_decoder_a.get());
+  collection_.AddAudioDecoder(audio_decoder_b.get());
 
-  // Verify that first SelectDataSource() returns data_source_a.
-  collection_.SelectDataSource(&data_source);
-  EXPECT_TRUE(data_source);
-  EXPECT_EQ(data_source, data_source_a);
+  // Verify that first SelectAudioDecoder() returns audio_decoder_a.
+  collection_.SelectAudioDecoder(&audio_decoder);
+  EXPECT_TRUE(audio_decoder);
+  EXPECT_EQ(audio_decoder, audio_decoder_a);
   EXPECT_FALSE(collection_.IsEmpty());
 
-  // Verify that second SelectDataSource() returns data_source_b.
-  collection_.SelectDataSource(&data_source);
-  EXPECT_TRUE(data_source);
-  EXPECT_EQ(data_source, data_source_b);
+  // Verify that second SelectAudioDecoder() returns audio_decoder_b.
+  collection_.SelectAudioDecoder(&audio_decoder);
+  EXPECT_TRUE(audio_decoder);
+  EXPECT_EQ(audio_decoder, audio_decoder_b);
   EXPECT_TRUE(collection_.IsEmpty());
 
-  // Verify that third SelectDataSource() returns nothing.
-  collection_.SelectDataSource(&data_source);
-  EXPECT_FALSE(data_source);
+  // Verify that third SelectAudioDecoder() returns nothing.
+  collection_.SelectAudioDecoder(&audio_decoder);
+  EXPECT_FALSE(audio_decoder);
 }
 
 }  // namespace media
diff --git a/media/base/filter_factories.cc b/media/base/filter_factories.cc
new file mode 100644
index 0000000..bb5df5e
--- /dev/null
+++ b/media/base/filter_factories.cc
@@ -0,0 +1,11 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "media/base/filter_factories.h"
+
+namespace media {
+
+DataSourceFactory::~DataSourceFactory() {}
+
+}  // namespace media
diff --git a/media/base/filter_factories.h b/media/base/filter_factories.h
new file mode 100644
index 0000000..c00adef
--- /dev/null
+++ b/media/base/filter_factories.h
@@ -0,0 +1,35 @@
+// 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.
+
+#ifndef MEDIA_BASE_FILTER_FACTORIES_H_
+#define MEDIA_BASE_FILTER_FACTORIES_H_
+
+#include<string>
+
+#include "base/callback.h"
+#include "media/base/pipeline_status.h"
+
+namespace media {
+
+class DataSource;
+
+// Asynchronous factory interface for building DataSource objects.
+class DataSourceFactory {
+ public:
+  // Ownership of the DataSource is transferred through this callback.
+  typedef Callback2<PipelineError, DataSource*>::Type BuildCallback;
+
+  virtual ~DataSourceFactory();
+
+  // Builds a DataSource 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 DataSourceFactory* Clone() const = 0;
+};
+
+}  // namespace media
+
+#endif  // MEDIA_BASE_FILTER_FACTORIES_H_
diff --git a/media/base/filters.cc b/media/base/filters.cc
index 1328dd1..fc5f537 100644
--- a/media/base/filters.cc
+++ b/media/base/filters.cc
@@ -66,10 +66,6 @@ void Filter::Seek(base::TimeDelta time, FilterCallback* callback) {
 void Filter::OnAudioRendererDisabled() {
 }
 
-bool DataSource::IsUrlSupported(const std::string& url) {
-  return true;
-}
-
 void* DemuxerStream::QueryInterface(const char* interface_id) {
   return NULL;
 }
diff --git a/media/base/filters.h b/media/base/filters.h
index 11bf526..89a7584 100644
--- a/media/base/filters.h
+++ b/media/base/filters.h
@@ -112,12 +112,6 @@ class DataSource : public Filter {
   typedef Callback1<size_t>::Type ReadCallback;
   static const size_t kReadError = static_cast<size_t>(-1);
 
-  virtual bool IsUrlSupported(const std::string& url);
-
-  // Initialize a DataSource for the given URL, executing the callback upon
-  // completion.
-  virtual void Initialize(const std::string& url, FilterCallback* callback) = 0;
-
   // Reads |size| bytes from |position| into |data|. And when the read is done
   // or failed, |read_callback| is called with the number of bytes read or
   // kReadError in case of error.
diff --git a/media/base/mock_callback.cc b/media/base/mock_callback.cc
index e948368..03ed896 100644
--- a/media/base/mock_callback.cc
+++ b/media/base/mock_callback.cc
@@ -20,10 +20,35 @@ void MockCallback::ExpectRunAndDelete() {
   EXPECT_CALL(*this, Destructor());
 }
 
+MockStatusCallback::MockStatusCallback() {}
+
+MockStatusCallback::~MockStatusCallback() {
+  Destructor();
+}
+
+// Required by GMock to allow the RunWithParams() expectation
+// in ExpectRunAndDelete() to compile.
+bool operator==(const Tuple1<PipelineError>& lhs,
+                const Tuple1<PipelineError>& rhs) {
+  return lhs.a == rhs.a;
+}
+
+void MockStatusCallback::ExpectRunAndDelete(PipelineError error) {
+  EXPECT_CALL(*this, RunWithParams(Tuple1<PipelineError>(error)));
+  EXPECT_CALL(*this, Destructor());
+}
+
 MockCallback* NewExpectedCallback() {
   StrictMock<MockCallback>* callback = new StrictMock<MockCallback>();
   callback->ExpectRunAndDelete();
   return callback;
 }
 
+MockStatusCallback* NewExpectedStatusCallback(PipelineError error) {
+  StrictMock<MockStatusCallback>* callback =
+      new StrictMock<MockStatusCallback>();
+  callback->ExpectRunAndDelete(error);
+  return callback;
+}
+
 }  // namespace media
diff --git a/media/base/mock_callback.h b/media/base/mock_callback.h
index 2aef84a..ddf1bef 100644
--- a/media/base/mock_callback.h
+++ b/media/base/mock_callback.h
@@ -6,6 +6,7 @@
 #define MEDIA_BASE_MOCK_CALLBACK_H_
 
 #include "base/callback.h"
+#include "media/base/pipeline_status.h"
 #include "testing/gmock/include/gmock/gmock.h"
 
 namespace media {
@@ -48,9 +49,30 @@ class MockCallback : public CallbackRunner<Tuple0> {
   DISALLOW_COPY_AND_ASSIGN(MockCallback);
 };
 
-// Convenience function that automatically creates and sets an expectation for
+// Helper class similar to MockCallback but is used where a
+// PipelineStatusCallback is needed.
+class MockStatusCallback : public CallbackRunner<Tuple1<PipelineError> > {
+ public:
+  MockStatusCallback();
+  virtual ~MockStatusCallback();
+
+  MOCK_METHOD1(RunWithParams, void(const Tuple1<PipelineError>& params));
+
+  // Can be used to verify the object is destroyed.
+  MOCK_METHOD0(Destructor, void());
+
+  // Convenience function to set expectations for the callback to execute and
+  // deleted.
+  void ExpectRunAndDelete(PipelineError error);
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(MockStatusCallback);
+};
+
+// Convenience functions that automatically create and set an expectation for
 // the callback to run.
 MockCallback* NewExpectedCallback();
+MockStatusCallback* NewExpectedStatusCallback(PipelineError error);
 
 }  // namespace media
 
diff --git a/media/base/mock_filters.cc b/media/base/mock_filters.cc
index 1f8a78b..68e344f 100644
--- a/media/base/mock_filters.cc
+++ b/media/base/mock_filters.cc
@@ -4,12 +4,78 @@
 
 #include "media/base/mock_filters.h"
 
+#include "media/base/filter_host.h"
+
+using ::testing::_;
+using ::testing::Invoke;
+using ::testing::NotNull;
+
 namespace media {
 
-MockDataSource::MockDataSource() {}
+MockDataSource::MockDataSource()
+    : total_bytes_(-1),
+      buffered_bytes_(-1) {
+}
 
 MockDataSource::~MockDataSource() {}
 
+void MockDataSource::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_);
+}
+
+void MockDataSource::SetTotalAndBufferedBytes(int64 total_bytes,
+                                              int64 buffered_bytes) {
+  total_bytes_ = total_bytes;
+  buffered_bytes_ = buffered_bytes;
+}
+
+MockDataSourceFactory::MockDataSourceFactory(MockDataSource* data_source)
+    : data_source_(data_source),
+      error_(PIPELINE_OK) {
+}
+
+MockDataSourceFactory::~MockDataSourceFactory() {}
+
+void MockDataSourceFactory::SetError(PipelineError error) {
+  error_ = error;
+}
+
+void MockDataSourceFactory::RunBuildCallback(const std::string& url,
+                                             BuildCallback* callback) {
+  scoped_ptr<BuildCallback> cb(callback);
+
+  if (!data_source_.get()) {
+    cb->Run(PIPELINE_ERROR_REQUIRED_FILTER_MISSING,
+            static_cast<DataSource*>(NULL));
+    return;
+  }
+
+  scoped_refptr<MockDataSource> data_source = data_source_;
+  data_source_ = NULL;
+
+  if (error_ == PIPELINE_OK) {
+    cb->Run(PIPELINE_OK, data_source.get());
+    return;
+  }
+
+  cb->Run(error_, static_cast<DataSource*>(NULL));
+}
+
+void MockDataSourceFactory::DestroyBuildCallback(const std::string& url,
+                                                 BuildCallback* callback) {
+  delete callback;
+}
+
+DataSourceFactory* MockDataSourceFactory::Clone() const {
+  return new MockDataSourceFactory(data_source_.get());
+}
+
 MockDemuxer::MockDemuxer() {}
 
 MockDemuxer::~MockDemuxer() {}
@@ -46,12 +112,29 @@ MockFilterCollection::MockFilterCollection()
 MockFilterCollection::~MockFilterCollection() {}
 
 FilterCollection* MockFilterCollection::filter_collection(
-    bool include_data_source) const {
+    bool include_data_source,
+    bool run_build_callback,
+    PipelineError build_error) const {
   FilterCollection* collection = new FilterCollection();
 
-  if (include_data_source) {
-    collection->AddDataSource(data_source_);
+  MockDataSourceFactory* data_source_factory =
+      new MockDataSourceFactory(include_data_source ? data_source_ : NULL);
+
+  if (build_error != PIPELINE_OK)
+    data_source_factory->SetError(build_error);
+
+  if (run_build_callback) {
+    ON_CALL(*data_source_factory, Build(_, NotNull()))
+        .WillByDefault(Invoke(data_source_factory,
+                              &MockDataSourceFactory::RunBuildCallback));
+  } else {
+    ON_CALL(*data_source_factory, Build(_, NotNull()))
+        .WillByDefault(Invoke(data_source_factory,
+                              &MockDataSourceFactory::DestroyBuildCallback));
   }
+  EXPECT_CALL(*data_source_factory, Build(_, NotNull()));
+
+  collection->SetDataSourceFactory(data_source_factory);
   collection->AddDemuxer(demuxer_);
   collection->AddVideoDecoder(video_decoder_);
   collection->AddAudioDecoder(audio_decoder_);
diff --git a/media/base/mock_filters.h b/media/base/mock_filters.h
index dd391ac..cc121ba 100644
--- a/media/base/mock_filters.h
+++ b/media/base/mock_filters.h
@@ -69,28 +69,54 @@ class MockDataSource : public DataSource {
   MockDataSource();
 
   // 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));
   MOCK_METHOD0(OnAudioRendererDisabled, void());
 
   // DataSource implementation.
-  MOCK_METHOD1(IsUrlSupported, bool(const std::string& url));
-  MOCK_METHOD2(Initialize, void(const std::string& url,
-                                FilterCallback* callback));
   MOCK_METHOD0(media_format, const MediaFormat&());
   MOCK_METHOD4(Read, void(int64 position, size_t size, uint8* data,
                           DataSource::ReadCallback* callback));
   MOCK_METHOD1(GetSize, bool(int64* size_out));
   MOCK_METHOD0(IsStreaming, bool());
 
+  // Sets the TotalBytes & BufferedBytes values to be sent to host() when
+  // the set_host() is called.
+  void SetTotalAndBufferedBytes(int64 total_bytes, int64 buffered_bytes);
+
  protected:
   virtual ~MockDataSource();
 
  private:
+  int64 total_bytes_;
+  int64 buffered_bytes_;
+
   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();
@@ -256,10 +282,12 @@ class MockFilterCollection {
   MockAudioRenderer* audio_renderer() const { return audio_renderer_; }
 
   FilterCollection* filter_collection() const {
-    return filter_collection(true);
+    return filter_collection(true, true, PIPELINE_OK);
   }
 
-  FilterCollection* filter_collection(bool include_data_source) const;
+  FilterCollection* filter_collection(bool include_data_source,
+                                      bool run_build_callback,
+                                      PipelineError build_error) const;
 
  private:
   scoped_refptr<MockDataSource> data_source_;
@@ -300,18 +328,6 @@ ACTION_P2(SetDuration, filter, duration) {
   filter->host()->SetDuration(duration);
 }
 
-// Helper gmock action that calls SetTotalBytes() on behalf of the provided
-// filter.
-ACTION_P2(SetTotalBytes, filter, bytes) {
-  filter->host()->SetTotalBytes(bytes);
-}
-
-// Helper gmock action that calls SetBufferedBytes() on behalf of the provided
-// filter.
-ACTION_P2(SetBufferedBytes, filter, bytes) {
-  filter->host()->SetBufferedBytes(bytes);
-}
-
 // Helper gmock action that calls DisableAudioRenderer() on behalf of the
 // provided filter.
 ACTION_P(DisableAudioRenderer, filter) {
diff --git a/media/base/pipeline.h b/media/base/pipeline.h
index 8e7a8db..802903d 100644
--- a/media/base/pipeline.h
+++ b/media/base/pipeline.h
@@ -12,6 +12,7 @@
 #include <string>
 
 #include "base/callback.h"
+#include "media/base/pipeline_status.h"
 
 namespace base {
 class TimeDelta;
@@ -19,29 +20,6 @@ class TimeDelta;
 
 namespace media {
 
-// Error definitions for pipeline.  All codes indicate an error except:
-// PIPELINE_OK indicates the pipeline is running normally.
-enum PipelineError {
-  PIPELINE_OK,
-  PIPELINE_ERROR_URL_NOT_FOUND,
-  PIPELINE_ERROR_NETWORK,
-  PIPELINE_ERROR_DECODE,
-  PIPELINE_ERROR_ABORT,
-  PIPELINE_ERROR_INITIALIZATION_FAILED,
-  PIPELINE_ERROR_REQUIRED_FILTER_MISSING,
-  PIPELINE_ERROR_OUT_OF_MEMORY,
-  PIPELINE_ERROR_COULD_NOT_RENDER,
-  PIPELINE_ERROR_READ,
-  PIPELINE_ERROR_AUDIO_HARDWARE,
-  PIPELINE_ERROR_OPERATION_PENDING,
-  PIPELINE_ERROR_INVALID_STATE,
-  // Demuxer related errors.
-  DEMUXER_ERROR_COULD_NOT_OPEN,
-  DEMUXER_ERROR_COULD_NOT_PARSE,
-  DEMUXER_ERROR_NO_SUPPORTED_STREAMS,
-  DEMUXER_ERROR_COULD_NOT_CREATE_THREAD,
-};
-
 struct PipelineStatistics {
   PipelineStatistics() :
       audio_bytes_decoded(0),
diff --git a/media/base/pipeline_impl.cc b/media/base/pipeline_impl.cc
index d4009c8..4f19fcc 100644
--- a/media/base/pipeline_impl.cc
+++ b/media/base/pipeline_impl.cc
@@ -18,7 +18,6 @@ namespace media {
 
 class PipelineImpl::PipelineInitState {
  public:
-  scoped_refptr<DataSource> data_source_;
   scoped_refptr<Demuxer> demuxer_;
   scoped_refptr<AudioDecoder> audio_decoder_;
   scoped_refptr<VideoDecoder> video_decoder_;
@@ -550,7 +549,12 @@ void PipelineImpl::StartTask(FilterCollection* filter_collection,
   seek_callback_.reset(start_callback);
 
   // Kick off initialization.
-  InitializeTask();
+  set_state(kInitDataSource);
+  pipeline_init_state_.reset(new PipelineInitState());
+  pipeline_init_state_->composite_ = new CompositeFilter(message_loop_);
+  pipeline_init_state_->composite_->set_host(this);
+
+  InitializeDataSource();
 }
 
 // Main initialization method called on the pipeline thread.  This code attempts
@@ -580,31 +584,12 @@ void PipelineImpl::InitializeTask() {
     return;
   }
 
-  DCHECK(state_ == kCreated ||
-         state_ == kInitDataSource ||
-         state_ == kInitDemuxer ||
+  DCHECK(state_ == kInitDemuxer ||
          state_ == kInitAudioDecoder ||
          state_ == kInitAudioRenderer ||
          state_ == kInitVideoDecoder ||
          state_ == kInitVideoRenderer);
 
-  // Just created, create data source.
-  if (state_ == kCreated) {
-    set_state(kInitDataSource);
-    pipeline_init_state_.reset(new PipelineInitState());
-    pipeline_init_state_->composite_ = new CompositeFilter(message_loop_);
-    pipeline_init_state_->composite_->set_host(this);
-
-    InitializeDataSource();
-    return;
-  }
-
-  // Data source created, create demuxer.
-  if (state_ == kInitDataSource) {
-    set_state(kInitDemuxer);
-    InitializeDemuxer(pipeline_init_state_->data_source_);
-    return;
-  }
 
   // Demuxer created, create audio decoder.
   if (state_ == kInitDemuxer) {
@@ -1013,24 +998,30 @@ void PipelineImpl::InitializeDataSource() {
   DCHECK_EQ(MessageLoop::current(), message_loop_);
   DCHECK(IsPipelineOk());
 
-  scoped_refptr<DataSource> data_source;
-  while (true) {
-    filter_collection_->SelectDataSource(&data_source);
-    if (!data_source || data_source->IsUrlSupported(url_))
-      break;
-  }
+  filter_collection_->GetDataSourceFactory()->Build(url_,
+      NewCallback(this, &PipelineImpl::OnDataSourceBuilt));
+}
 
-  if (!data_source) {
-    SetError(PIPELINE_ERROR_REQUIRED_FILTER_MISSING);
+void PipelineImpl::OnDataSourceBuilt(PipelineError error,
+                                     DataSource* data_source) {
+  if (MessageLoop::current() != message_loop_) {
+    message_loop_->PostTask(FROM_HERE,
+                            NewRunnableMethod(this,
+                                              &PipelineImpl::OnDataSourceBuilt,
+                                              error,
+                                              make_scoped_refptr(data_source)));
     return;
   }
 
-  if (!PrepareFilter(data_source))
+  if (error != PIPELINE_OK) {
+    SetError(error);
     return;
+  }
+
+  PrepareFilter(data_source);
 
-  pipeline_init_state_->data_source_ = data_source;
-  data_source->Initialize(
-      url_, NewCallback(this, &PipelineImpl::OnFilterInitialize));
+  set_state(kInitDemuxer);
+  InitializeDemuxer(data_source);
 }
 
 void PipelineImpl::InitializeDemuxer(
@@ -1207,6 +1198,7 @@ void PipelineImpl::TearDownPipeline() {
       // Make it look like initialization was successful.
       pipeline_filter_ = pipeline_init_state_->composite_;
       pipeline_init_state_.reset();
+      filter_collection_.reset();
 
       set_state(kStopping);
       pipeline_filter_->Stop(
diff --git a/media/base/pipeline_impl.h b/media/base/pipeline_impl.h
index fa4b5d1..0fe8c09 100644
--- a/media/base/pipeline_impl.h
+++ b/media/base/pipeline_impl.h
@@ -246,6 +246,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);
 
   // Returns true if the asynchronous action of creating decoder has started.
diff --git a/media/base/pipeline_impl_unittest.cc b/media/base/pipeline_impl_unittest.cc
index b583879..486b5e0 100644
--- a/media/base/pipeline_impl_unittest.cc
+++ b/media/base/pipeline_impl_unittest.cc
@@ -85,13 +85,10 @@ 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(DoAll(SetTotalBytes(mocks_->data_source(), kTotalBytes),
-                        SetBufferedBytes(mocks_->data_source(), kBufferedBytes),
-                        Invoke(&RunFilterCallback)));
+    mocks_->data_source()->SetTotalAndBufferedBytes(kTotalBytes,
+                                                    kBufferedBytes);
+
     EXPECT_CALL(*mocks_->data_source(), SetPlaybackRate(0.0f));
-    EXPECT_CALL(*mocks_->data_source(), IsUrlSupported(_))
-        .WillOnce(Return(true));
     EXPECT_CALL(*mocks_->data_source(), Seek(base::TimeDelta(), NotNull()))
         .WillOnce(Invoke(&RunFilterCallback));
     EXPECT_CALL(*mocks_->data_source(), Stop(NotNull()))
@@ -189,9 +186,14 @@ class PipelineImplTest : public ::testing::Test {
   // Sets up expectations on the callback and initializes the pipeline.  Called
   // after tests have set expectations any filters they wish to use.
   void InitializePipeline() {
+    InitializePipeline(PIPELINE_OK);
+  }
+
+  void InitializePipeline(PipelineError factory_error) {
     // Expect an initialization callback.
     EXPECT_CALL(callbacks_, OnStart());
-    pipeline_->Start(mocks_->filter_collection(), "",
+    pipeline_->Start(mocks_->filter_collection(true, true, factory_error),
+                     "",
                      NewCallback(reinterpret_cast<CallbackHelper*>(&callbacks_),
                                  &CallbackHelper::OnStart));
     message_loop_.RunAllPending();
@@ -312,17 +314,10 @@ TEST_F(PipelineImplTest, NotStarted) {
 }
 
 TEST_F(PipelineImplTest, NeverInitializes) {
-  EXPECT_CALL(*mocks_->data_source(), Initialize("", NotNull()))
-      .WillOnce(Invoke(&DestroyFilterCallback));
-  EXPECT_CALL(*mocks_->data_source(), IsUrlSupported(_))
-      .WillOnce(Return(true));
-  EXPECT_CALL(*mocks_->data_source(), Stop(NotNull()))
-      .WillOnce(Invoke(&RunStopFilterCallback));
-
   // This test hangs during initialization by never calling
   // InitializationComplete().  StrictMock<> will ensure that the callback is
   // never executed.
-  pipeline_->Start(mocks_->filter_collection(), "",
+  pipeline_->Start(mocks_->filter_collection(false, false, PIPELINE_OK), "",
                    NewCallback(reinterpret_cast<CallbackHelper*>(&callbacks_),
                                &CallbackHelper::OnStart));
   message_loop_.RunAllPending();
@@ -346,7 +341,9 @@ TEST_F(PipelineImplTest, RequiredFilterMissing) {
   EXPECT_CALL(callbacks_, OnStart());
 
   // Create a filter collection with missing filter.
-  FilterCollection* collection = mocks_->filter_collection(false);
+  FilterCollection* collection =
+      mocks_->filter_collection(false, true,
+                                PIPELINE_ERROR_REQUIRED_FILTER_MISSING);
   pipeline_->Start(collection, "",
                    NewCallback(reinterpret_cast<CallbackHelper*>(&callbacks_),
                                &CallbackHelper::OnStart));
@@ -358,17 +355,10 @@ TEST_F(PipelineImplTest, RequiredFilterMissing) {
 }
 
 TEST_F(PipelineImplTest, URLNotFound) {
-  EXPECT_CALL(*mocks_->data_source(), Initialize("", NotNull()))
-      .WillOnce(DoAll(SetError(mocks_->data_source(),
-                               PIPELINE_ERROR_URL_NOT_FOUND),
-                      Invoke(&RunFilterCallback)));
-  EXPECT_CALL(*mocks_->data_source(), IsUrlSupported(_))
-      .WillOnce(Return(true));
+
   EXPECT_CALL(callbacks_, OnError());
-  EXPECT_CALL(*mocks_->data_source(), Stop(NotNull()))
-      .WillOnce(Invoke(&RunStopFilterCallback));
 
-  InitializePipeline();
+  InitializePipeline(PIPELINE_ERROR_URL_NOT_FOUND);
   EXPECT_FALSE(pipeline_->IsInitialized());
   EXPECT_EQ(PIPELINE_ERROR_URL_NOT_FOUND, pipeline_->GetError());
 }
@@ -376,10 +366,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(), Initialize("", NotNull()))
-      .WillOnce(Invoke(&RunFilterCallback));
-  EXPECT_CALL(*mocks_->data_source(), IsUrlSupported(_))
-      .WillOnce(Return(true));
   EXPECT_CALL(*mocks_->data_source(), Stop(NotNull()))
       .WillOnce(Invoke(&RunStopFilterCallback));
 
diff --git a/media/base/pipeline_status.h b/media/base/pipeline_status.h
new file mode 100644
index 0000000..787708e
--- /dev/null
+++ b/media/base/pipeline_status.h
@@ -0,0 +1,41 @@
+// 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.
+
+#ifndef MEDIA_BASE_PIPELINE_STATUS_H_
+#define MEDIA_BASE_PIPELINE_STATUS_H_
+
+#include "base/callback.h"
+
+namespace media {
+
+// Error definitions for pipeline.  All codes indicate an error except:
+// PIPELINE_OK indicates the pipeline is running normally.
+enum PipelineError {
+  PIPELINE_OK,
+  PIPELINE_ERROR_URL_NOT_FOUND,
+  PIPELINE_ERROR_NETWORK,
+  PIPELINE_ERROR_DECODE,
+  PIPELINE_ERROR_ABORT,
+  PIPELINE_ERROR_INITIALIZATION_FAILED,
+  PIPELINE_ERROR_REQUIRED_FILTER_MISSING,
+  PIPELINE_ERROR_OUT_OF_MEMORY,
+  PIPELINE_ERROR_COULD_NOT_RENDER,
+  PIPELINE_ERROR_READ,
+  PIPELINE_ERROR_AUDIO_HARDWARE,
+  PIPELINE_ERROR_OPERATION_PENDING,
+  PIPELINE_ERROR_INVALID_STATE,
+  // Demuxer related errors.
+  DEMUXER_ERROR_COULD_NOT_OPEN,
+  DEMUXER_ERROR_COULD_NOT_PARSE,
+  DEMUXER_ERROR_NO_SUPPORTED_STREAMS,
+  DEMUXER_ERROR_COULD_NOT_CREATE_THREAD,
+  // DataSourceFactory errors.
+  DATASOURCE_ERROR_URL_NOT_SUPPORTED,
+};
+
+typedef Callback1<media::PipelineError>::Type PipelineStatusCallback;
+
+}  // namespace media
+
+#endif  // MEDIA_BASE_PIPELINE_STATUS_H_
diff --git a/media/filters/file_data_source.cc b/media/filters/file_data_source.cc
index 9fd93a5..8a1f055 100644
--- a/media/filters/file_data_source.cc
+++ b/media/filters/file_data_source.cc
@@ -23,10 +23,8 @@ FileDataSource::~FileDataSource() {
   DCHECK(!file_);
 }
 
-void FileDataSource::Initialize(const std::string& url,
-                                FilterCallback* callback) {
+PipelineError FileDataSource::Initialize(const std::string& url) {
   DCHECK(!file_);
-  scoped_ptr<FilterCallback> c(callback);
 #if defined(OS_WIN)
   FilePath file_path(UTF8ToWide(url));
 #else
@@ -37,14 +35,24 @@ void FileDataSource::Initialize(const std::string& url,
   }
   if (!file_) {
     file_size_ = 0;
-    host()->SetError(PIPELINE_ERROR_URL_NOT_FOUND);
-    callback->Run();
-    return;
+    return PIPELINE_ERROR_URL_NOT_FOUND;
   }
   media_format_.SetAsString(MediaFormat::kURL, url);
-  host()->SetTotalBytes(file_size_);
-  host()->SetBufferedBytes(file_size_);
-  callback->Run();
+
+  if (host()) {
+    host()->SetTotalBytes(file_size_);
+    host()->SetBufferedBytes(file_size_);
+  }
+
+  return PIPELINE_OK;
+}
+
+void FileDataSource::set_host(FilterHost* filter_host) {
+  DataSource::set_host(filter_host);
+  if (file_) {
+    host()->SetTotalBytes(file_size_);
+    host()->SetBufferedBytes(file_size_);
+  }
 }
 
 void FileDataSource::Stop(FilterCallback* callback) {
diff --git a/media/filters/file_data_source.h b/media/filters/file_data_source.h
index 7be9a5a..5ae3956 100644
--- a/media/filters/file_data_source.h
+++ b/media/filters/file_data_source.h
@@ -1,4 +1,4 @@
-// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// 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.
 
@@ -20,11 +20,13 @@ class FileDataSource : public DataSource {
   FileDataSource();
   virtual ~FileDataSource();
 
+  PipelineError Initialize(const std::string& url);
+
   // Implementation of Filter.
+  virtual void set_host(FilterHost* filter_host);
   virtual void Stop(FilterCallback* callback);
 
   // Implementation of DataSource.
-  virtual void Initialize(const std::string& url, FilterCallback* callback);
   virtual const MediaFormat& media_format();
   virtual void Read(int64 position, size_t size, uint8* data,
                     ReadCallback* read_callback);
diff --git a/media/filters/file_data_source_factory.cc b/media/filters/file_data_source_factory.cc
new file mode 100644
index 0000000..80642fb
--- /dev/null
+++ b/media/filters/file_data_source_factory.cc
@@ -0,0 +1,40 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "media/filters/file_data_source_factory.h"
+
+#include "base/logging.h"
+#include "media/filters/file_data_source.h"
+
+namespace media {
+
+FileDataSourceFactory::FileDataSourceFactory() {}
+
+FileDataSourceFactory::~FileDataSourceFactory() {}
+
+void FileDataSourceFactory::Build(const std::string& url,
+                                  BuildCallback* callback) {
+  DCHECK(callback);
+
+  if (url.empty()) {
+    callback->Run(media::PIPELINE_ERROR_URL_NOT_FOUND,
+                  static_cast<media::DataSource*>(NULL));
+    delete callback;
+    return;
+  }
+
+  scoped_refptr<FileDataSource> file_data_source = new FileDataSource();
+
+  PipelineError error = file_data_source->Initialize(url);
+  DataSource* data_source =
+      (error == PIPELINE_OK) ? file_data_source.get() : NULL;
+  callback->Run(error, data_source);
+  delete callback;
+}
+
+DataSourceFactory* FileDataSourceFactory::Clone() const {
+  return new FileDataSourceFactory();
+}
+
+}  // namespace media
diff --git a/media/filters/file_data_source_factory.h b/media/filters/file_data_source_factory.h
new file mode 100644
index 0000000..18c0587
--- /dev/null
+++ b/media/filters/file_data_source_factory.h
@@ -0,0 +1,27 @@
+// 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.
+
+#ifndef MEDIA_FILTERS_FILE_DATA_SOURCE_FACTORY_H_
+#define MEDIA_FILTERS_FILE_DATA_SOURCE_FACTORY_H_
+
+#include "media/base/filter_factories.h"
+
+namespace media {
+
+class FileDataSourceFactory : public DataSourceFactory {
+ public:
+  FileDataSourceFactory();
+  virtual ~FileDataSourceFactory();
+
+  // DataSourceFactory methods.
+  virtual void Build(const std::string& url, BuildCallback* callback);
+  virtual DataSourceFactory* Clone() const;
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(FileDataSourceFactory);
+};
+
+}  // namespace media
+
+#endif  // MEDIA_FILTERS_FILE_DATA_SOURCE_FACTORY_H_
diff --git a/media/filters/file_data_source_unittest.cc b/media/filters/file_data_source_unittest.cc
index 1770ac4..1ae111e 100644
--- a/media/filters/file_data_source_unittest.cc
+++ b/media/filters/file_data_source_unittest.cc
@@ -56,7 +56,7 @@ TEST(FileDataSourceTest, OpenFile) {
 
   scoped_refptr<FileDataSource> filter(new FileDataSource());
   filter->set_host(&host);
-  filter->Initialize(TestFileURL(), NewExpectedCallback());
+  EXPECT_EQ(PIPELINE_OK, filter->Initialize(TestFileURL()));
 
   filter->Stop(NewExpectedCallback());
 }
@@ -71,7 +71,7 @@ TEST(FileDataSourceTest, ReadData) {
   scoped_refptr<FileDataSource> filter(new FileDataSource());
 
   filter->set_host(&host);
-  filter->Initialize(TestFileURL(), NewExpectedCallback());
+  EXPECT_EQ(PIPELINE_OK, filter->Initialize(TestFileURL()));
 
   EXPECT_TRUE(filter->GetSize(&size));
   EXPECT_EQ(10, size);
diff --git a/media/media.gyp b/media/media.gyp
index ebccbf2..933f513 100644
--- a/media/media.gyp
+++ b/media/media.gyp
@@ -72,12 +72,16 @@
         'audio/win/wavein_input_win.h',
         'audio/win/waveout_output_win.cc',
         'audio/win/waveout_output_win.h',
+        'base/async_filter_factory_base.cc',
+        'base/async_filter_factory_base.h',
         'base/buffers.cc',
         'base/buffers.h',
         'base/callback.cc',
         'base/callback.h',
         'base/clock.cc',
         'base/clock.h',
+        'base/composite_data_source_factory.cc',
+        'base/composite_data_source_factory.h',
         'base/composite_filter.cc',
         'base/composite_filter.h',
         'base/data_buffer.cc',
@@ -86,6 +90,8 @@
         'base/djb2.h',
         'base/filter_collection.cc',
         'base/filter_collection.h',
+        'base/filter_factories.cc',
+        'base/filter_factories.h',
         'base/filter_host.h',
         'base/filters.cc',
         'base/filters.h',
@@ -105,6 +111,7 @@
         'base/pipeline.h',
         'base/pipeline_impl.cc',
         'base/pipeline_impl.h',
+        'base/pipeline_status.h',
         'base/pts_heap.cc',
         'base/pts_heap.h',
         'base/seekable_buffer.cc',
@@ -146,6 +153,8 @@
         'filters/ffmpeg_video_decoder.h',
         'filters/file_data_source.cc',
         'filters/file_data_source.h',
+        'filters/file_data_source_factory.cc',
+        'filters/file_data_source_factory.h',
         'filters/null_audio_renderer.cc',
         'filters/null_audio_renderer.h',
         'filters/null_video_renderer.cc',
diff --git a/media/tools/player_wtl/movie.cc b/media/tools/player_wtl/movie.cc
index fe73490..371d178 100644
--- a/media/tools/player_wtl/movie.cc
+++ b/media/tools/player_wtl/movie.cc
@@ -1,4 +1,4 @@
-// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// 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.
 
@@ -14,7 +14,7 @@
 #include "media/filters/ffmpeg_audio_decoder.h"
 #include "media/filters/ffmpeg_demuxer.h"
 #include "media/filters/ffmpeg_video_decoder.h"
-#include "media/filters/file_data_source.h"
+#include "media/filters/file_data_source_factory.h"
 #include "media/filters/null_audio_renderer.h"
 #include "media/tools/player_wtl/wtl_renderer.h"
 
@@ -22,7 +22,7 @@ using media::AudioRendererImpl;
 using media::FFmpegAudioDecoder;
 using media::FFmpegDemuxer;
 using media::FFmpegVideoDecoder;
-using media::FileDataSource;
+using media::FileDataSourceFactory;
 using media::FilterCollection;
 using media::PipelineImpl;
 
@@ -65,7 +65,7 @@ bool Movie::Open(const wchar_t* url, WtlVideoRenderer* video_renderer) {
 
   // Create filter collection.
   scoped_ptr<FilterCollection> collection(new FilterCollection());
-  collection->AddDataSource(new FileDataSource());
+  collection->SetDataSourceFactory(new FileDataSourceFactory());
   collection->AddAudioDecoder(new FFmpegAudioDecoder(
       message_loop_factory_->GetMessageLoop("AudioDecoderThread")));
   collection->AddDemuxer(new FFmpegDemuxer(
diff --git a/media/tools/player_x11/player_x11.cc b/media/tools/player_x11/player_x11.cc
index 39504ff..68c56ae 100644
--- a/media/tools/player_x11/player_x11.cc
+++ b/media/tools/player_x11/player_x11.cc
@@ -23,7 +23,7 @@
 #include "media/filters/ffmpeg_audio_decoder.h"
 #include "media/filters/ffmpeg_demuxer.h"
 #include "media/filters/ffmpeg_video_decoder.h"
-#include "media/filters/file_data_source.h"
+#include "media/filters/file_data_source_factory.h"
 #include "media/filters/null_audio_renderer.h"
 #include "media/filters/omx_video_decoder.h"
 
@@ -103,7 +103,7 @@ bool InitPipeline(MessageLoop* message_loop,
   // Create our filter factories.
   scoped_ptr<media::FilterCollection> collection(
       new media::FilterCollection());
-  collection->AddDataSource(new media::FileDataSource());
+  collection->SetDataSourceFactory(new media::FileDataSourceFactory());
   collection->AddDemuxer(new media::FFmpegDemuxer(
       message_loop_factory->GetMessageLoop("DemuxThread")));
   collection->AddAudioDecoder(new media::FFmpegAudioDecoder(