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

41 files changed, 1397 insertions, 373 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(
diff --git a/webkit/glue/media/buffered_data_source.cc b/webkit/glue/media/buffered_data_source.cc
index 9cb0284..135c49e 100644
--- a/webkit/glue/media/buffered_data_source.cc
+++ b/webkit/glue/media/buffered_data_source.cc
@@ -6,6 +6,7 @@
 
 #include "media/base/filter_host.h"
 #include "net/base/net_errors.h"
+#include "webkit/glue/media/web_data_source_factory.h"
 #include "webkit/glue/webkit_glue.h"
 
 using WebKit::WebFrame;
@@ -27,10 +28,25 @@ static const int kReadTrials = 3;
 // of FFmpeg.
 static const int kInitialReadBufferSize = 32768;
 
+static WebDataSource* NewBufferedDataSource(MessageLoop* render_loop,
+                                            WebKit::WebFrame* frame) {
+  return new BufferedDataSource(render_loop, frame);
+}
+
+// static
+media::DataSourceFactory* BufferedDataSource::CreateFactory(
+    MessageLoop* render_loop,
+    WebKit::WebFrame* frame,
+    WebDataSourceBuildObserverHack* build_observer) {
+  return new WebDataSourceFactory(render_loop, frame, &NewBufferedDataSource,
+                                  build_observer);
+}
+
 BufferedDataSource::BufferedDataSource(
     MessageLoop* render_loop,
     WebFrame* frame)
     : total_bytes_(kPositionNotSpecified),
+      buffered_bytes_(0),
       loaded_(false),
       streaming_(false),
       frame_(frame),
@@ -73,18 +89,25 @@ base::TimeDelta BufferedDataSource::GetTimeoutMilliseconds() {
   return base::TimeDelta::FromMilliseconds(kTimeoutMilliseconds);
 }
 
-/////////////////////////////////////////////////////////////////////////////
-// media::Filter implementation.
+void BufferedDataSource::set_host(media::FilterHost* host) {
+  DataSource::set_host(host);
+
+  if (loader_.get())
+    UpdateHostState();
+}
+
 void BufferedDataSource::Initialize(const std::string& url,
-                                    media::FilterCallback* callback) {
+                                    media::PipelineStatusCallback* callback) {
   // Saves the url.
   url_ = GURL(url);
 
-  if (!IsProtocolSupportedForMedia(url_)) {
-    // This method is called on the thread where host() lives so it is safe
-    // to make this call.
-    host()->SetError(media::PIPELINE_ERROR_NETWORK);
-    callback->Run();
+  // This data source doesn't support data:// protocol so reject it.
+  if (url_.SchemeIs(kDataScheme)) {
+    callback->Run(media::DATASOURCE_ERROR_URL_NOT_SUPPORTED);
+    delete callback;
+    return;
+  } else if (!IsProtocolSupportedForMedia(url_)) {
+    callback->Run(media::PIPELINE_ERROR_NETWORK);
     delete callback;
     return;
   }
@@ -99,13 +122,18 @@ void BufferedDataSource::Initialize(const std::string& url,
       NewRunnableMethod(this, &BufferedDataSource::InitializeTask));
 }
 
-bool BufferedDataSource::IsUrlSupported(const std::string& url) {
-  GURL gurl(url);
+void BufferedDataSource::CancelInitialize() {
+  base::AutoLock auto_lock(lock_);
+  DCHECK(initialize_callback_.get());
 
-  // This data source doesn't support data:// protocol so reject it.
-  return IsProtocolSupportedForMedia(gurl) && !gurl.SchemeIs(kDataScheme);
+  initialize_callback_.reset();
+
+  render_loop_->PostTask(
+      FROM_HERE, NewRunnableMethod(this, &BufferedDataSource::CleanupTask));
 }
 
+/////////////////////////////////////////////////////////////////////////////
+// media::Filter implementation.
 void BufferedDataSource::Stop(media::FilterCallback* callback) {
   {
     base::AutoLock auto_lock(lock_);
@@ -190,7 +218,7 @@ void BufferedDataSource::Abort() {
 void BufferedDataSource::InitializeTask() {
   DCHECK(MessageLoop::current() == render_loop_);
   DCHECK(!loader_.get());
-  if (stopped_on_render_loop_)
+  if (stopped_on_render_loop_ || !initialize_callback_.get())
     return;
 
   // Kick starts the watch dog task that will handle connection timeout.
@@ -384,13 +412,14 @@ void BufferedDataSource::DoneRead_Locked(int error) {
   read_buffer_ = 0;
 }
 
-void BufferedDataSource::DoneInitialization_Locked() {
+void BufferedDataSource::DoneInitialization_Locked(media::PipelineError error) {
   DCHECK(MessageLoop::current() == render_loop_);
   DCHECK(initialize_callback_.get());
   lock_.AssertAcquired();
 
-  initialize_callback_->Run();
-  initialize_callback_.reset();
+  scoped_ptr<media::PipelineStatusCallback> initialize_callback(
+      initialize_callback_.release());
+  initialize_callback->Run(error);
 }
 
 /////////////////////////////////////////////////////////////////////////////
@@ -403,6 +432,11 @@ void BufferedDataSource::HttpInitialStartCallback(int error) {
   bool partial_response = loader_->partial_response();
   bool success = error == net::OK;
 
+  if (!initialize_callback_.get()) {
+    loader_->Stop();
+    return;
+  }
+
   if (success) {
     // TODO(hclam): Needs more thinking about supporting servers without range
     // request or their partial response is not complete.
@@ -427,74 +461,82 @@ void BufferedDataSource::HttpInitialStartCallback(int error) {
     return;
   }
 
-  // We need to prevent calling to filter host and running the callback if
-  // we have received the stop signal. We need to lock down the whole callback
-  // method to prevent bad things from happening. The reason behind this is
-  // that we cannot guarantee tasks on render thread have completely stopped
-  // when we receive the Stop() method call. The only way to solve this is to
-  // let tasks on render thread to run but make sure they don't call outside
-  // this object when Stop() method is ever called. Locking this method is safe
-  // because |lock_| is only acquired in tasks on render thread.
-  base::AutoLock auto_lock(lock_);
-  if (stop_signal_received_)
-    return;
+  // Reference to prevent destruction while inside the |initialize_callback_|
+  // call. This is a temporary fix to prevent crashes caused by holding the
+  // lock and running the destructor.
+  // TODO: Review locking in this class and figure out a way to run the callback
+  //       w/o the lock.
+  scoped_refptr<BufferedDataSource> destruction_guard(this);
+  {
+    // We need to prevent calling to filter host and running the callback if
+    // we have received the stop signal. We need to lock down the whole callback
+    // method to prevent bad things from happening. The reason behind this is
+    // that we cannot guarantee tasks on render thread have completely stopped
+    // when we receive the Stop() method call. The only way to solve this is to
+    // let tasks on render thread to run but make sure they don't call outside
+    // this object when Stop() method is ever called. Locking this method is
+    // safe because |lock_| is only acquired in tasks on render thread.
+    base::AutoLock auto_lock(lock_);
+    if (stop_signal_received_)
+      return;
 
-  if (!success) {
-    host()->SetError(media::PIPELINE_ERROR_NETWORK);
-    DoneInitialization_Locked();
-    return;
-  }
+    if (!success) {
+      DoneInitialization_Locked(media::PIPELINE_ERROR_NETWORK);
+      return;
+    }
 
-  if (streaming_) {
-    // If the server didn't reply with an instance size, it is likely this
-    // is a streaming response.
-    host()->SetStreaming(true);
-  } else {
-    // This value governs the range that we can seek to.
-    // TODO(hclam): Report the correct value of buffered bytes.
-    host()->SetTotalBytes(total_bytes_);
-    host()->SetBufferedBytes(0);
+    UpdateHostState();
+    DoneInitialization_Locked(media::PIPELINE_OK);
   }
-
-  // Currently, only files can be used reliably w/o a network.
-  host()->SetLoaded(false);
-  DoneInitialization_Locked();
 }
 
 void BufferedDataSource::NonHttpInitialStartCallback(int error) {
   DCHECK(MessageLoop::current() == render_loop_);
   DCHECK(loader_.get());
 
+  if (!initialize_callback_.get()) {
+    loader_->Stop();
+    return;
+  }
+
   int64 instance_size = loader_->instance_size();
   bool success = error == net::OK && instance_size != kPositionNotSpecified;
 
   if (success) {
     total_bytes_ = instance_size;
+    buffered_bytes_ = total_bytes_;
     loaded_ = true;
   } else {
     loader_->Stop();
   }
 
-  // We need to prevent calling to filter host and running the callback if
-  // we have received the stop signal. We need to lock down the whole callback
-  // method to prevent bad things from happening. The reason behind this is
-  // that we cannot guarantee tasks on render thread have completely stopped
-  // when we receive the Stop() method call. The only way to solve this is to
-  // let tasks on render thread to run but make sure they don't call outside
-  // this object when Stop() method is ever called. Locking this method is safe
-  // because |lock_| is only acquired in tasks on render thread.
-  base::AutoLock auto_lock(lock_);
-  if (stop_signal_received_)
-    return;
+  // Reference to prevent destruction while inside the |initialize_callback_|
+  // call. This is a temporary fix to prevent crashes caused by holding the
+  // lock and running the destructor.
+  // TODO: Review locking in this class and figure out a way to run the callback
+  //       w/o the lock.
+  scoped_refptr<BufferedDataSource> destruction_guard(this);
+  {
+    // We need to prevent calling to filter host and running the callback if
+    // we have received the stop signal. We need to lock down the whole callback
+    // method to prevent bad things from happening. The reason behind this is
+    // that we cannot guarantee tasks on render thread have completely stopped
+    // when we receive the Stop() method call. The only way to solve this is to
+    // let tasks on render thread to run but make sure they don't call outside
+    // this object when Stop() method is ever called. Locking this method is
+    // safe because |lock_| is only acquired in tasks on render thread.
+    base::AutoLock auto_lock(lock_);
+    if (stop_signal_received_ || !initialize_callback_.get())
+      return;
 
-  if (success) {
-    host()->SetTotalBytes(total_bytes_);
-    host()->SetBufferedBytes(total_bytes_);
-    host()->SetLoaded(loaded_);
-  } else {
-    host()->SetError(media::PIPELINE_ERROR_NETWORK);
+    if (!success) {
+      DoneInitialization_Locked(media::PIPELINE_ERROR_NETWORK);
+      return;
+    }
+
+    UpdateHostState();
+    DoneInitialization_Locked(media::PIPELINE_OK);
   }
-  DoneInitialization_Locked();
 }
 
 void BufferedDataSource::PartialReadStartCallback(int error) {
@@ -594,9 +636,28 @@ void BufferedDataSource::NetworkEventCallback() {
 
   if (network_activity != network_activity_) {
     network_activity_ = network_activity;
-    host()->SetNetworkActivity(network_activity);
+    if (host())
+      host()->SetNetworkActivity(network_activity);
+  }
+
+  buffered_bytes_ = buffered_position + 1;
+  if (host())
+    host()->SetBufferedBytes(buffered_bytes_);
+}
+
+void BufferedDataSource::UpdateHostState() {
+  media::FilterHost* filter_host = host();
+  if (!filter_host)
+    return;
+
+  filter_host->SetLoaded(loaded_);
+
+  if (streaming_) {
+    filter_host->SetStreaming(true);
+  } else {
+    filter_host->SetTotalBytes(total_bytes_);
+    filter_host->SetBufferedBytes(buffered_bytes_);
   }
-  host()->SetBufferedBytes(buffered_position + 1);
 }
 
 }  // namespace webkit_glue
diff --git a/webkit/glue/media/buffered_data_source.h b/webkit/glue/media/buffered_data_source.h
index 895b259..9e043b0 100644
--- a/webkit/glue/media/buffered_data_source.h
+++ b/webkit/glue/media/buffered_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.
 
@@ -10,21 +10,26 @@
 #include "base/callback.h"
 #include "base/scoped_ptr.h"
 #include "base/synchronization/lock.h"
+#include "media/base/filter_factories.h"
 #include "webkit/glue/media/buffered_resource_loader.h"
 
 namespace webkit_glue {
 
 class BufferedDataSource : public WebDataSource {
  public:
+  // Creates a DataSourceFactory for building BufferedDataSource objects.
+  static media::DataSourceFactory* CreateFactory(
+      MessageLoop* render_loop,
+      WebKit::WebFrame* frame,
+      WebDataSourceBuildObserverHack* build_observer);
+
   BufferedDataSource(MessageLoop* render_loop,
                      WebKit::WebFrame* frame);
 
   virtual ~BufferedDataSource();
 
   // media::Filter implementation.
-  virtual void Initialize(const std::string& url,
-                          media::FilterCallback* callback);
-  virtual bool IsUrlSupported(const std::string& url);
+  virtual void set_host(media::FilterHost* host);
   virtual void Stop(media::FilterCallback* callback);
   virtual void SetPlaybackRate(float playback_rate);
 
@@ -41,6 +46,9 @@ class BufferedDataSource : public WebDataSource {
   }
 
   // webkit_glue::WebDataSource implementation.
+  virtual void Initialize(const std::string& url,
+                          media::PipelineStatusCallback* callback);
+  virtual void CancelInitialize();
   virtual bool HasSingleOrigin();
   virtual void Abort();
 
@@ -90,7 +98,7 @@ class BufferedDataSource : public WebDataSource {
   void DoneRead_Locked(int error);
 
   // Calls |initialize_callback_| and reset it.
-  void DoneInitialization_Locked();
+  void DoneInitialization_Locked(media::PipelineError error);
 
   // Callback method for |loader_| if URL for the resource requested is using
   // HTTP protocol. This method is called when response for initial request is
@@ -116,6 +124,8 @@ class BufferedDataSource : public WebDataSource {
   // Callback method when a network event is received.
   void NetworkEventCallback();
 
+  void UpdateHostState();
+
   media::MediaFormat media_format_;
 
   // URL of the resource requested.
@@ -126,6 +136,7 @@ class BufferedDataSource : public WebDataSource {
   // member is guaranteed to happen after it is first written, so we don't
   // need to protect it.
   int64 total_bytes_;
+  int64 buffered_bytes_;
 
   // True if this data source is considered loaded.
   bool loaded_;
@@ -144,7 +155,7 @@ class BufferedDataSource : public WebDataSource {
   bool network_activity_;
 
   // Callback method from the pipeline for initialization.
-  scoped_ptr<media::FilterCallback> initialize_callback_;
+  scoped_ptr<media::PipelineStatusCallback> initialize_callback_;
 
   // Read parameters received from the Read() method call.
   scoped_ptr<media::DataSource::ReadCallback> read_callback_;
diff --git a/webkit/glue/media/buffered_data_source_unittest.cc b/webkit/glue/media/buffered_data_source_unittest.cc
index 41f37d6..9763678 100644
--- a/webkit/glue/media/buffered_data_source_unittest.cc
+++ b/webkit/glue/media/buffered_data_source_unittest.cc
@@ -168,6 +168,7 @@ class BufferedDataSourceTest : public testing::Test {
         .WillByDefault(Return(partial_response));
     ON_CALL(*loader_, url())
         .WillByDefault(ReturnRef(gurl_));
+    media::PipelineError expected_init_error = media::PIPELINE_OK;
     if (initialized_ok) {
       // Expected loaded or not.
       EXPECT_CALL(host_, SetLoaded(loaded));
@@ -183,12 +184,13 @@ class BufferedDataSourceTest : public testing::Test {
         EXPECT_CALL(host_, SetStreaming(true));
       }
     } else {
-      EXPECT_CALL(host_, SetError(media::PIPELINE_ERROR_NETWORK));
+      expected_init_error = media::PIPELINE_ERROR_NETWORK;
       EXPECT_CALL(*loader_, Stop());
     }
 
     // Actual initialization of the data source.
-    data_source_->Initialize(url, media::NewExpectedCallback());
+    data_source_->Initialize(url,
+        media::NewExpectedStatusCallback(expected_init_error));
     message_loop_->RunAllPending();
 
     if (initialized_ok) {
diff --git a/webkit/glue/media/buffered_resource_loader.cc b/webkit/glue/media/buffered_resource_loader.cc
index 4a3ae8b..c6295bf 100644
--- a/webkit/glue/media/buffered_resource_loader.cc
+++ b/webkit/glue/media/buffered_resource_loader.cc
@@ -244,7 +244,7 @@ void BufferedResourceLoader::willSendRequest(
     return;
   }
 
-    // Only allow |single_origin_| if we haven't seen a different origin yet.
+  // Only allow |single_origin_| if we haven't seen a different origin yet.
   if (single_origin_)
     single_origin_ = url_.GetOrigin() == GURL(newRequest.url()).GetOrigin();
 
diff --git a/webkit/glue/media/simple_data_source.cc b/webkit/glue/media/simple_data_source.cc
index f1564d9..ca258d4 100644
--- a/webkit/glue/media/simple_data_source.cc
+++ b/webkit/glue/media/simple_data_source.cc
@@ -12,12 +12,27 @@
 #include "net/url_request/url_request_status.h"
 #include "third_party/WebKit/Source/WebKit/chromium/public/WebKit.h"
 #include "third_party/WebKit/Source/WebKit/chromium/public/WebKitClient.h"
+#include "webkit/glue/media/web_data_source_factory.h"
 #include "webkit/glue/webkit_glue.h"
 
 namespace webkit_glue {
 
 static const char kDataScheme[] = "data";
 
+static WebDataSource* NewSimpleDataSource(MessageLoop* render_loop,
+                                          WebKit::WebFrame* frame) {
+  return new SimpleDataSource(render_loop, frame);
+}
+
+// static
+media::DataSourceFactory* SimpleDataSource::CreateFactory(
+    MessageLoop* render_loop,
+    WebKit::WebFrame* frame,
+    WebDataSourceBuildObserverHack* build_observer) {
+  return new WebDataSourceFactory(render_loop, frame, &NewSimpleDataSource,
+                                  build_observer);
+}
+
 SimpleDataSource::SimpleDataSource(
     MessageLoop* render_loop,
     WebKit::WebFrame* frame)
@@ -35,6 +50,15 @@ SimpleDataSource::~SimpleDataSource() {
   DCHECK(state_ == UNINITIALIZED || state_ == STOPPED);
 }
 
+void SimpleDataSource::set_host(media::FilterHost* host) {
+  DataSource::set_host(host);
+
+  base::AutoLock auto_lock(lock_);
+  if (state_ == INITIALIZED) {
+    UpdateHostState();
+  }
+}
+
 void SimpleDataSource::Stop(media::FilterCallback* callback) {
   base::AutoLock auto_lock(lock_);
   state_ = STOPPED;
@@ -48,26 +72,42 @@ void SimpleDataSource::Stop(media::FilterCallback* callback) {
       NewRunnableMethod(this, &SimpleDataSource::CancelTask));
 }
 
-void SimpleDataSource::Initialize(const std::string& url,
-                                  media::FilterCallback* callback) {
-  base::AutoLock auto_lock(lock_);
-  DCHECK_EQ(state_, UNINITIALIZED);
-  DCHECK(callback);
-  state_ = INITIALIZING;
-  initialize_callback_.reset(callback);
-
-  // Validate the URL.
-  SetURL(GURL(url));
-  if (!url_.is_valid() || !IsProtocolSupportedForMedia(url_)) {
-    host()->SetError(media::PIPELINE_ERROR_NETWORK);
-    initialize_callback_->Run();
-    initialize_callback_.reset();
-    return;
+void SimpleDataSource::Initialize(
+    const std::string& url,
+    media::PipelineStatusCallback* callback) {
+  // Reference to prevent destruction while inside the |initialize_callback_|
+  // call. This is a temporary fix to prevent crashes caused by holding the
+  // lock and running the destructor.
+  scoped_refptr<SimpleDataSource> destruction_guard(this);
+  {
+    base::AutoLock auto_lock(lock_);
+    DCHECK_EQ(state_, UNINITIALIZED);
+    DCHECK(callback);
+    state_ = INITIALIZING;
+    initialize_callback_.reset(callback);
+
+    // Validate the URL.
+    SetURL(GURL(url));
+    if (!url_.is_valid() || !IsProtocolSupportedForMedia(url_)) {
+      DoneInitialization_Locked(false);
+      return;
+    }
+
+    // Post a task to the render thread to start loading the resource.
+    render_loop_->PostTask(FROM_HERE,
+        NewRunnableMethod(this, &SimpleDataSource::StartTask));
   }
+}
 
-  // Post a task to the render thread to start loading the resource.
+void SimpleDataSource::CancelInitialize() {
+  base::AutoLock auto_lock(lock_);
+  DCHECK(initialize_callback_.get());
+  state_ = STOPPED;
+  initialize_callback_.reset();
+
+  // Post a task to the render thread to cancel loading the resource.
   render_loop_->PostTask(FROM_HERE,
-      NewRunnableMethod(this, &SimpleDataSource::StartTask));
+      NewRunnableMethod(this, &SimpleDataSource::CancelTask));
 }
 
 const media::MediaFormat& SimpleDataSource::media_format() {
@@ -157,44 +197,56 @@ void SimpleDataSource::didFinishLoading(
     WebKit::WebURLLoader* loader,
     double finishTime) {
   DCHECK(MessageLoop::current() == render_loop_);
-  base::AutoLock auto_lock(lock_);
-  // It's possible this gets called after Stop(), in which case |host_| is no
-  // longer valid.
-  if (state_ == STOPPED)
-    return;
-
-  // Otherwise we should be initializing and have created a WebURLLoader.
-  DCHECK_EQ(state_, INITIALIZING);
-
-  // If we don't get a content length or the request has failed, report it
-  // as a network error.
-  if (size_ == -1)
-    size_ = data_.length();
-  DCHECK(static_cast<size_t>(size_) == data_.length());
-
-  DoneInitialization_Locked(true);
+  // Reference to prevent destruction while inside the |initialize_callback_|
+  // call. This is a temporary fix to prevent crashes caused by holding the
+  // lock and running the destructor.
+  scoped_refptr<SimpleDataSource> destruction_guard(this);
+  {
+    base::AutoLock auto_lock(lock_);
+    // It's possible this gets called after Stop(), in which case |host_| is no
+    // longer valid.
+    if (state_ == STOPPED)
+      return;
+
+    // Otherwise we should be initializing and have created a WebURLLoader.
+    DCHECK_EQ(state_, INITIALIZING);
+
+    // If we don't get a content length or the request has failed, report it
+    // as a network error.
+    if (size_ == -1)
+      size_ = data_.length();
+    DCHECK(static_cast<size_t>(size_) == data_.length());
+
+    DoneInitialization_Locked(true);
+  }
 }
 
 void SimpleDataSource::didFail(
     WebKit::WebURLLoader* loader,
     const WebKit::WebURLError& error) {
   DCHECK(MessageLoop::current() == render_loop_);
-  base::AutoLock auto_lock(lock_);
-  // It's possible this gets called after Stop(), in which case |host_| is no
-  // longer valid.
-  if (state_ == STOPPED)
-    return;
-
-  // Otherwise we should be initializing and have created a WebURLLoader.
-  DCHECK_EQ(state_, INITIALIZING);
-
-  // If we don't get a content length or the request has failed, report it
-  // as a network error.
-  if (size_ == -1)
-    size_ = data_.length();
-  DCHECK(static_cast<size_t>(size_) == data_.length());
-
-  DoneInitialization_Locked(false);
+  // Reference to prevent destruction while inside the |initialize_callback_|
+  // call. This is a temporary fix to prevent crashes caused by holding the
+  // lock and running the destructor.
+  scoped_refptr<SimpleDataSource> destruction_guard(this);
+  {
+    base::AutoLock auto_lock(lock_);
+    // It's possible this gets called after Stop(), in which case |host_| is no
+    // longer valid.
+    if (state_ == STOPPED)
+      return;
+
+    // Otherwise we should be initializing and have created a WebURLLoader.
+    DCHECK_EQ(state_, INITIALIZING);
+
+    // If we don't get a content length or the request has failed, report it
+    // as a network error.
+    if (size_ == -1)
+      size_ = data_.length();
+    DCHECK(static_cast<size_t>(size_) == data_.length());
+
+    DoneInitialization_Locked(false);
+  }
 }
 
 bool SimpleDataSource::HasSingleOrigin() {
@@ -215,37 +267,43 @@ void SimpleDataSource::SetURL(const GURL& url) {
 
 void SimpleDataSource::StartTask() {
   DCHECK(MessageLoop::current() == render_loop_);
-  base::AutoLock auto_lock(lock_);
-
-  // We may have stopped.
-  if (state_ == STOPPED)
-    return;
-
-  CHECK(frame_);
-
-  DCHECK_EQ(state_, INITIALIZING);
-
-  if (url_.SchemeIs(kDataScheme)) {
-    // If this using data protocol, we just need to decode it.
-    std::string mime_type, charset;
-    bool success = net::DataURL::Parse(url_, &mime_type, &charset, &data_);
-
-    // Don't care about the mime-type just proceed if decoding was successful.
-    size_ = data_.length();
-    DoneInitialization_Locked(success);
-  } else {
-    // Prepare the request.
-    WebKit::WebURLRequest request(url_);
-    request.setTargetType(WebKit::WebURLRequest::TargetIsMedia);
-
-    frame_->setReferrerForRequest(request, WebKit::WebURL());
-
-    // This flag is for unittests as we don't want to reset |url_loader|
-    if (!keep_test_loader_)
-      url_loader_.reset(frame_->createAssociatedURLLoader());
-
-    // Start the resource loading.
-    url_loader_->loadAsynchronously(request, this);
+  // Reference to prevent destruction while inside the |initialize_callback_|
+  // call. This is a temporary fix to prevent crashes caused by holding the
+  // lock and running the destructor.
+  scoped_refptr<SimpleDataSource> destruction_guard(this);
+  {
+    base::AutoLock auto_lock(lock_);
+
+    // We may have stopped.
+    if (state_ == STOPPED)
+      return;
+
+    CHECK(frame_);
+
+    DCHECK_EQ(state_, INITIALIZING);
+
+    if (url_.SchemeIs(kDataScheme)) {
+      // If this using data protocol, we just need to decode it.
+      std::string mime_type, charset;
+      bool success = net::DataURL::Parse(url_, &mime_type, &charset, &data_);
+
+      // Don't care about the mime-type just proceed if decoding was successful.
+      size_ = data_.length();
+      DoneInitialization_Locked(success);
+    } else {
+      // Prepare the request.
+      WebKit::WebURLRequest request(url_);
+      request.setTargetType(WebKit::WebURLRequest::TargetIsMedia);
+
+      frame_->setReferrerForRequest(request, WebKit::WebURL());
+
+      // This flag is for unittests as we don't want to reset |url_loader|
+      if (!keep_test_loader_)
+        url_loader_.reset(frame_->createAssociatedURLLoader());
+
+      // Start the resource loading.
+      url_loader_->loadAsynchronously(request, this);
+    }
   }
 }
 
@@ -263,17 +321,29 @@ void SimpleDataSource::CancelTask() {
 
 void SimpleDataSource::DoneInitialization_Locked(bool success) {
   lock_.AssertAcquired();
+  media::PipelineError error = media::PIPELINE_ERROR_NETWORK;
   if (success) {
     state_ = INITIALIZED;
+
+    UpdateHostState();
+    error = media::PIPELINE_OK;
+  } else {
+    state_ = UNINITIALIZED;
+    url_loader_.reset();
+  }
+
+  scoped_ptr<media::PipelineStatusCallback> initialize_callback(
+      initialize_callback_.release());
+  initialize_callback->Run(error);
+}
+
+void SimpleDataSource::UpdateHostState() {
+  if (host()) {
     host()->SetTotalBytes(size_);
     host()->SetBufferedBytes(size_);
     // If scheme is file or data, say we are loaded.
     host()->SetLoaded(url_.SchemeIsFile() || url_.SchemeIs(kDataScheme));
-  } else {
-    host()->SetError(media::PIPELINE_ERROR_NETWORK);
   }
-  initialize_callback_->Run();
-  initialize_callback_.reset();
 }
 
 }  // namespace webkit_glue
diff --git a/webkit/glue/media/simple_data_source.h b/webkit/glue/media/simple_data_source.h
index 10d93ff..5e59ead 100644
--- a/webkit/glue/media/simple_data_source.h
+++ b/webkit/glue/media/simple_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.
 
@@ -15,6 +15,7 @@
 
 #include "base/message_loop.h"
 #include "base/scoped_ptr.h"
+#include "media/base/filter_factories.h"
 #include "media/base/filters.h"
 #include "third_party/WebKit/Source/WebKit/chromium/public/WebFrame.h"
 #include "third_party/WebKit/Source/WebKit/chromium/public/WebURLLoader.h"
@@ -31,15 +32,20 @@ namespace webkit_glue {
 class SimpleDataSource : public WebDataSource,
                          public WebKit::WebURLLoaderClient {
  public:
+  // Creates a DataSourceFactory for building SimpleDataSource objects.
+  static media::DataSourceFactory* CreateFactory(
+      MessageLoop* render_loop,
+      WebKit::WebFrame* frame,
+      WebDataSourceBuildObserverHack* build_observer);
+
   SimpleDataSource(MessageLoop* render_loop, WebKit::WebFrame* frame);
   virtual ~SimpleDataSource();
 
   // media::Filter implementation.
+  virtual void set_host(media::FilterHost* host);
   virtual void Stop(media::FilterCallback* callback);
 
   // media::DataSource implementation.
-  virtual void Initialize(const std::string& url,
-                          media::FilterCallback* callback);
   virtual const media::MediaFormat& media_format();
   virtual void Read(int64 position, size_t size,
                     uint8* data, ReadCallback* read_callback);
@@ -79,6 +85,9 @@ class SimpleDataSource : public WebDataSource,
       const WebKit::WebURLError&);
 
   // webkit_glue::WebDataSource implementation.
+  virtual void Initialize(const std::string& url,
+                          media::PipelineStatusCallback* callback);
+  virtual void CancelInitialize();
   virtual bool HasSingleOrigin();
   virtual void Abort();
 
@@ -95,6 +104,9 @@ class SimpleDataSource : public WebDataSource,
   // Perform initialization completion tasks under a lock.
   void DoneInitialization_Locked(bool success);
 
+  // Update host() stats like total bytes & buffered bytes.
+  void UpdateHostState();
+
   // Primarily used for asserting the bridge is loading on the render thread.
   MessageLoop* render_loop_;
 
@@ -123,7 +135,7 @@ class SimpleDataSource : public WebDataSource,
   base::Lock lock_;
 
   // Filter callbacks.
-  scoped_ptr<media::FilterCallback> initialize_callback_;
+  scoped_ptr<media::PipelineStatusCallback> initialize_callback_;
 
   // Used to ensure mocks for unittests are used instead of reset in Start().
   bool keep_test_loader_;
diff --git a/webkit/glue/media/simple_data_source_unittest.cc b/webkit/glue/media/simple_data_source_unittest.cc
index 977807b..d3177ae 100644
--- a/webkit/glue/media/simple_data_source_unittest.cc
+++ b/webkit/glue/media/simple_data_source_unittest.cc
@@ -59,7 +59,7 @@ class SimpleDataSourceTest : public testing::Test {
   }
 
   void InitializeDataSource(const char* url,
-                            media::MockCallback* callback) {
+                            media::MockStatusCallback* callback) {
     gurl_ = GURL(url);
 
     frame_.reset(new NiceMock<MockWebFrame>());
@@ -103,7 +103,6 @@ class SimpleDataSourceTest : public testing::Test {
 
   void RequestFailed() {
     InSequence s;
-    EXPECT_CALL(host_, SetError(media::PIPELINE_ERROR_NETWORK));
 
     WebURLError error;
     error.reason = net::ERR_FAILED;
@@ -158,19 +157,22 @@ class SimpleDataSourceTest : public testing::Test {
 };
 
 TEST_F(SimpleDataSourceTest, InitializeHTTP) {
-  InitializeDataSource(kHttpUrl, media::NewExpectedCallback());
+  InitializeDataSource(kHttpUrl,
+                       media::NewExpectedStatusCallback(media::PIPELINE_OK));
   RequestSucceeded(false);
   DestroyDataSource();
 }
 
 TEST_F(SimpleDataSourceTest, InitializeHTTPS) {
-  InitializeDataSource(kHttpsUrl, media::NewExpectedCallback());
+  InitializeDataSource(kHttpsUrl,
+                       media::NewExpectedStatusCallback(media::PIPELINE_OK));
   RequestSucceeded(false);
   DestroyDataSource();
 }
 
 TEST_F(SimpleDataSourceTest, InitializeFile) {
-  InitializeDataSource(kFileUrl, media::NewExpectedCallback());
+  InitializeDataSource(kFileUrl,
+                       media::NewExpectedStatusCallback(media::PIPELINE_OK));
   RequestSucceeded(true);
   DestroyDataSource();
 }
@@ -181,8 +183,6 @@ TEST_F(SimpleDataSourceTest, InitializeData) {
 
   data_source_ = new SimpleDataSource(MessageLoop::current(),
                                       frame_.get());
-  EXPECT_TRUE(data_source_->IsUrlSupported(kDataUrl));
-
   // There is no need to provide a message loop to data source.
   data_source_->set_host(&host_);
   data_source_->SetURLLoaderForTest(url_loader_);
@@ -191,14 +191,16 @@ TEST_F(SimpleDataSourceTest, InitializeData) {
   EXPECT_CALL(host_, SetTotalBytes(sizeof(kDataUrlDecoded)));
   EXPECT_CALL(host_, SetBufferedBytes(sizeof(kDataUrlDecoded)));
 
-  data_source_->Initialize(kDataUrl, media::NewExpectedCallback());
+  data_source_->Initialize(kDataUrl,
+      media::NewExpectedStatusCallback(media::PIPELINE_OK));
   MessageLoop::current()->RunAllPending();
 
   DestroyDataSource();
 }
 
 TEST_F(SimpleDataSourceTest, RequestFailed) {
-  InitializeDataSource(kHttpUrl, media::NewExpectedCallback());
+  InitializeDataSource(kHttpUrl,
+      media::NewExpectedStatusCallback(media::PIPELINE_ERROR_NETWORK));
   RequestFailed();
   DestroyDataSource();
 }
@@ -206,8 +208,8 @@ TEST_F(SimpleDataSourceTest, RequestFailed) {
 TEST_F(SimpleDataSourceTest, StopWhenDownloading) {
   // The callback should be deleted, but not executed.
   // TODO(scherkus): should this really be the behaviour?  Seems strange...
-  StrictMock<media::MockCallback>* callback =
-      new StrictMock<media::MockCallback>();
+  StrictMock<media::MockStatusCallback>* callback =
+      new StrictMock<media::MockStatusCallback>();
   EXPECT_CALL(*callback, Destructor());
 
   InitializeDataSource(kHttpUrl, callback);
@@ -217,7 +219,8 @@ TEST_F(SimpleDataSourceTest, StopWhenDownloading) {
 }
 
 TEST_F(SimpleDataSourceTest, AsyncRead) {
-  InitializeDataSource(kFileUrl, media::NewExpectedCallback());
+  InitializeDataSource(kFileUrl,
+                       media::NewExpectedStatusCallback(media::PIPELINE_OK));
   RequestSucceeded(true);
   AsyncRead();
   DestroyDataSource();
@@ -228,20 +231,23 @@ TEST_F(SimpleDataSourceTest, AsyncRead) {
 // is fixed.
 TEST_F(SimpleDataSourceTest, HasSingleOrigin) {
   // Make sure no redirect case works as expected.
-  InitializeDataSource(kHttpUrl, media::NewExpectedCallback());
+  InitializeDataSource(kHttpUrl,
+                       media::NewExpectedStatusCallback(media::PIPELINE_OK));
   RequestSucceeded(false);
   EXPECT_TRUE(data_source_->HasSingleOrigin());
   DestroyDataSource();
 
   // Test redirect to the same domain.
-  InitializeDataSource(kHttpUrl, media::NewExpectedCallback());
+  InitializeDataSource(kHttpUrl,
+                       media::NewExpectedStatusCallback(media::PIPELINE_OK));
   Redirect(kHttpRedirectToSameDomainUrl1);
   RequestSucceeded(false);
   EXPECT_TRUE(data_source_->HasSingleOrigin());
   DestroyDataSource();
 
   // Test redirect twice to the same domain.
-  InitializeDataSource(kHttpUrl, media::NewExpectedCallback());
+  InitializeDataSource(kHttpUrl,
+                       media::NewExpectedStatusCallback(media::PIPELINE_OK));
   Redirect(kHttpRedirectToSameDomainUrl1);
   Redirect(kHttpRedirectToSameDomainUrl2);
   RequestSucceeded(false);
@@ -249,14 +255,16 @@ TEST_F(SimpleDataSourceTest, HasSingleOrigin) {
   DestroyDataSource();
 
   // Test redirect to a different domain.
-  InitializeDataSource(kHttpUrl, media::NewExpectedCallback());
+  InitializeDataSource(kHttpUrl,
+                       media::NewExpectedStatusCallback(media::PIPELINE_OK));
   Redirect(kHttpRedirectToDifferentDomainUrl1);
   RequestSucceeded(false);
   EXPECT_FALSE(data_source_->HasSingleOrigin());
   DestroyDataSource();
 
   // Test redirect to the same domain and then to a different domain.
-  InitializeDataSource(kHttpUrl, media::NewExpectedCallback());
+  InitializeDataSource(kHttpUrl,
+                       media::NewExpectedStatusCallback(media::PIPELINE_OK));
   Redirect(kHttpRedirectToSameDomainUrl1);
   Redirect(kHttpRedirectToDifferentDomainUrl1);
   RequestSucceeded(false);
diff --git a/webkit/glue/media/web_data_source.h b/webkit/glue/media/web_data_source.h
index 956ac9e..25f1258 100644
--- a/webkit/glue/media/web_data_source.h
+++ b/webkit/glue/media/web_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.
 
@@ -6,6 +6,7 @@
 #define WEBKIT_GLUE_MEDIA_WEB_DATA_SOURCE_H_
 
 #include "media/base/filters.h"
+#include "media/base/pipeline_status.h"
 
 namespace webkit_glue {
 
@@ -16,6 +17,18 @@ class WebDataSource : public media::DataSource {
   WebDataSource();
   virtual ~WebDataSource();
 
+  // Initialize this object using |url|. This object calls |callback| when
+  // initialization has completed.
+  virtual void Initialize(const std::string& url,
+                          media::PipelineStatusCallback* callback) = 0;
+
+  // Called to cancel initialization. The callback passed in Initialize() will
+  // be destroyed and will not be called after this method returns. Once this
+  // method returns, the object will be in an uninitialized state and
+  // Initialize() cannot be called again. The caller is expected to release
+  // its handle to this object and never call it again.
+  virtual void CancelInitialize() = 0;
+
   // Returns true if the media resource has a single origin, false otherwise.
   //
   // Method called on the render thread.
@@ -31,6 +44,15 @@ class WebDataSource : public media::DataSource {
   DISALLOW_COPY_AND_ASSIGN(WebDataSource);
 };
 
+// Temporary hack to allow WebMediaPlayerImpl::Proxy::AddDataSource() to
+// be called when WebDataSource objects are created. This can be removed
+// once WebMediaPlayerImpl::Proxy is fixed so it doesn't have to track
+// WebDataSources. Proxy only has to track WebDataSources so it can call Abort()
+// on them at shutdown. Once cancellation is added to DataSource and pause
+// support in Demuxers cancel pending reads, Proxy shouldn't have to keep
+// a WebDataSource list or call Abort().
+typedef Callback1<WebDataSource*>::Type WebDataSourceBuildObserverHack;
+
 }  // namespace webkit_glue
 
 #endif  // WEBKIT_GLUE_MEDIA_WEB_DATA_SOURCE_H_
diff --git a/webkit/glue/media/web_data_source_factory.cc b/webkit/glue/media/web_data_source_factory.cc
new file mode 100644
index 0000000..f9036c7
--- /dev/null
+++ b/webkit/glue/media/web_data_source_factory.cc
@@ -0,0 +1,102 @@
+// 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 "webkit/glue/media/web_data_source_factory.h"
+
+#include "base/logging.h"
+
+namespace webkit_glue {
+
+class WebDataSourceFactory::BuildRequest
+    : public media::AsyncDataSourceFactoryBase::BuildRequest {
+ public:
+  BuildRequest(const std::string& url, BuildCallback* callback,
+               WebDataSource* data_source,
+               WebDataSourceBuildObserverHack* build_observer);
+  virtual ~BuildRequest();
+
+ protected:
+  // AsyncDataSourceFactoryBase::BuildRequest method.
+  virtual void DoStart();
+
+ private:
+  void InitDone(media::PipelineError error);
+
+  scoped_refptr<WebDataSource> data_source_;
+  WebDataSourceBuildObserverHack* build_observer_;
+
+  DISALLOW_COPY_AND_ASSIGN(BuildRequest);
+};
+
+WebDataSourceFactory::WebDataSourceFactory(
+    MessageLoop* render_loop,
+    WebKit::WebFrame* frame,
+    FactoryFunction factory_function,
+    WebDataSourceBuildObserverHack* build_observer)
+    : render_loop_(render_loop),
+      frame_(frame),
+      factory_function_(factory_function),
+      build_observer_(build_observer) {
+  DCHECK(render_loop_);
+  DCHECK(frame_);
+  DCHECK(factory_function_);
+}
+
+WebDataSourceFactory::~WebDataSourceFactory() {}
+
+media::DataSourceFactory* WebDataSourceFactory::Clone() const {
+  return new WebDataSourceFactory(render_loop_, frame_, factory_function_,
+                                  build_observer_);
+}
+
+bool WebDataSourceFactory::AllowRequests() const {
+  return true;
+}
+
+media::AsyncDataSourceFactoryBase::BuildRequest*
+WebDataSourceFactory::CreateRequest(const std::string& url,
+                                    BuildCallback* callback) {
+  WebDataSource* data_source = factory_function_(render_loop_, frame_);
+
+  return new WebDataSourceFactory::BuildRequest(url, callback, data_source,
+                                                build_observer_);
+}
+
+WebDataSourceFactory::BuildRequest::BuildRequest(
+    const std::string& url,
+    BuildCallback* callback,
+    WebDataSource* data_source,
+    WebDataSourceBuildObserverHack* build_observer)
+    : AsyncDataSourceFactoryBase::BuildRequest(url, callback),
+      data_source_(data_source),
+      build_observer_(build_observer) {
+}
+
+WebDataSourceFactory::BuildRequest::~BuildRequest() {
+  if (data_source_.get()) {
+    data_source_->CancelInitialize();
+    data_source_ = NULL;
+  }
+}
+
+void WebDataSourceFactory::BuildRequest::DoStart() {
+  data_source_->Initialize(url(), NewCallback(this, &BuildRequest::InitDone));
+}
+
+void WebDataSourceFactory::BuildRequest::InitDone(media::PipelineError error) {
+  scoped_refptr<WebDataSource> data_source;
+
+  data_source = (error == media::PIPELINE_OK) ? data_source_ : NULL;
+  data_source_ = NULL;
+
+  if (build_observer_ && data_source.get()) {
+    build_observer_->Run(data_source.get());
+  }
+
+  RequestComplete(error, data_source);
+  // Don't do anything after this line. This object is deleted by
+  // RequestComplete().
+}
+
+}  // namespace webkit_glue
diff --git a/webkit/glue/media/web_data_source_factory.h b/webkit/glue/media/web_data_source_factory.h
new file mode 100644
index 0000000..7163ef4
--- /dev/null
+++ b/webkit/glue/media/web_data_source_factory.h
@@ -0,0 +1,51 @@
+// 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 WEBKIT_GLUE_MEDIA_BUFFERED_DATA_SOURCE_FACTORY_H_
+#define WEBKIT_GLUE_MEDIA_BUFFERED_DATA_SOURCE_FACTORY_H_
+
+#include "media/base/async_filter_factory_base.h"
+#include "webkit/glue/media/web_data_source.h"
+
+class MessageLoop;
+
+namespace WebKit {
+class WebFrame;
+}
+
+namespace webkit_glue {
+
+class WebDataSourceFactory : public media::AsyncDataSourceFactoryBase {
+ public:
+  typedef WebDataSource* (*FactoryFunction)(MessageLoop* render_loop,
+                                            WebKit::WebFrame* frame);
+
+  WebDataSourceFactory(MessageLoop* render_loop, WebKit::WebFrame* frame,
+                       FactoryFunction factory_function,
+                       WebDataSourceBuildObserverHack* build_observer);
+  virtual ~WebDataSourceFactory();
+
+  // DataSourceFactory method.
+  virtual media::DataSourceFactory* Clone() const;
+
+ protected:
+  // AsyncDataSourceFactoryBase methods.
+  virtual bool AllowRequests() const;
+  virtual AsyncDataSourceFactoryBase::BuildRequest* CreateRequest(
+      const std::string& url, BuildCallback* callback);
+
+ private:
+  class BuildRequest;
+
+  MessageLoop* render_loop_;
+  WebKit::WebFrame* frame_;
+  FactoryFunction factory_function_;
+  WebDataSourceBuildObserverHack* build_observer_;
+
+  DISALLOW_COPY_AND_ASSIGN(WebDataSourceFactory);
+};
+
+}  // namespace webkit_glue
+
+#endif  // WEBKIT_GLUE_MEDIA_BUFFERED_DATA_SOURCE_FACTORY_H_
diff --git a/webkit/glue/webkit_glue.gypi b/webkit/glue/webkit_glue.gypi
index 74cb61c..8c87826 100644
--- a/webkit/glue/webkit_glue.gypi
+++ b/webkit/glue/webkit_glue.gypi
@@ -368,6 +368,8 @@
         'media/video_renderer_impl.h',
         'media/web_data_source.cc',
         'media/web_data_source.h',
+        'media/web_data_source_factory.cc',
+        'media/web_data_source_factory.h',
         'media/web_video_renderer.h',
         'alt_error_page_resource_fetcher.cc',
         'alt_error_page_resource_fetcher.h',
diff --git a/webkit/glue/webmediaplayer_impl.cc b/webkit/glue/webmediaplayer_impl.cc
index 3801406..2d5e6a6 100644
--- a/webkit/glue/webmediaplayer_impl.cc
+++ b/webkit/glue/webmediaplayer_impl.cc
@@ -8,6 +8,7 @@
 
 #include "base/callback.h"
 #include "base/command_line.h"
+#include "media/base/composite_data_source_factory.h"
 #include "media/base/filter_collection.h"
 #include "media/base/limits.h"
 #include "media/base/media_format.h"
@@ -113,10 +114,10 @@ void WebMediaPlayerImpl::Proxy::SetVideoRenderer(
   video_renderer_ = video_renderer;
 }
 
-void WebMediaPlayerImpl::Proxy::AddDataSource(
-    scoped_refptr<WebDataSource> data_source) {
-  base::AutoLock auto_lock(data_sources_lock_);
-  data_sources_.push_back(data_source);
+WebDataSourceBuildObserverHack* WebMediaPlayerImpl::Proxy::GetBuildObserver() {
+  if (!build_observer_.get())
+    build_observer_.reset(NewCallback(this, &Proxy::AddDataSource));
+  return build_observer_.get();
 }
 
 void WebMediaPlayerImpl::Proxy::Paint(skia::PlatformCanvas* canvas,
@@ -195,6 +196,11 @@ void WebMediaPlayerImpl::Proxy::NetworkEventCallback() {
       &WebMediaPlayerImpl::Proxy::NetworkEventTask));
 }
 
+void WebMediaPlayerImpl::Proxy::AddDataSource(WebDataSource* data_source) {
+  base::AutoLock auto_lock(data_sources_lock_);
+  data_sources_.push_back(make_scoped_refptr(data_source));
+}
+
 void WebMediaPlayerImpl::Proxy::RepaintTask() {
   DCHECK(MessageLoop::current() == render_loop_);
   {
@@ -309,22 +315,28 @@ bool WebMediaPlayerImpl::Initialize(
                   &WebMediaPlayerImpl::Proxy::NetworkEventCallback));
 
   // A simple data source that keeps all data in memory.
-  scoped_refptr<SimpleDataSource> simple_data_source(
-    new SimpleDataSource(MessageLoop::current(), frame));
+  scoped_ptr<media::DataSourceFactory> simple_data_source_factory(
+      SimpleDataSource::CreateFactory(MessageLoop::current(), frame,
+                                      proxy_->GetBuildObserver()));
 
   // A sophisticated data source that does memory caching.
-  scoped_refptr<BufferedDataSource> buffered_data_source(
-      new BufferedDataSource(MessageLoop::current(), frame));
-  proxy_->AddDataSource(buffered_data_source);
+  scoped_ptr<media::DataSourceFactory> buffered_data_source_factory(
+      BufferedDataSource::CreateFactory(MessageLoop::current(), frame,
+                                        proxy_->GetBuildObserver()));
+
+  scoped_ptr<media::CompositeDataSourceFactory> data_source_factory(
+      new media::CompositeDataSourceFactory());
 
   if (use_simple_data_source) {
-    filter_collection_->AddDataSource(simple_data_source);
-    filter_collection_->AddDataSource(buffered_data_source);
+    data_source_factory->AddFactory(simple_data_source_factory.release());
+    data_source_factory->AddFactory(buffered_data_source_factory.release());
   } else {
-    filter_collection_->AddDataSource(buffered_data_source);
-    filter_collection_->AddDataSource(simple_data_source);
+    data_source_factory->AddFactory(buffered_data_source_factory.release());
+    data_source_factory->AddFactory(simple_data_source_factory.release());
   }
 
+  filter_collection_->SetDataSourceFactory(data_source_factory.release());
+
   // Add in the default filter factories.
   filter_collection_->AddDemuxer(new media::FFmpegDemuxer(
       message_loop_factory_->GetMessageLoop("DemuxThread")));
@@ -797,6 +809,7 @@ void WebMediaPlayerImpl::OnPipelineError() {
     case media::DEMUXER_ERROR_COULD_NOT_PARSE:
     case media::DEMUXER_ERROR_NO_SUPPORTED_STREAMS:
     case media::DEMUXER_ERROR_COULD_NOT_CREATE_THREAD:
+    case media::DATASOURCE_ERROR_URL_NOT_SUPPORTED:
       // Format error.
       SetNetworkState(WebMediaPlayer::FormatError);
       break;
diff --git a/webkit/glue/webmediaplayer_impl.h b/webkit/glue/webmediaplayer_impl.h
index cadf4d7..e7acf27 100644
--- a/webkit/glue/webmediaplayer_impl.h
+++ b/webkit/glue/webmediaplayer_impl.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.
 
@@ -67,6 +67,7 @@
 #include "third_party/WebKit/Source/WebKit/chromium/public/WebMediaPlayerClient.h"
 #include "ui/gfx/rect.h"
 #include "ui/gfx/size.h"
+#include "webkit/glue/media/web_data_source.h"
 
 class GURL;
 
@@ -77,7 +78,6 @@ class WebFrame;
 namespace webkit_glue {
 
 class MediaResourceLoaderBridgeFactory;
-class WebDataSource;
 class WebVideoRenderer;
 
 class WebMediaPlayerImpl : public WebKit::WebMediaPlayer,
@@ -99,7 +99,7 @@ class WebMediaPlayerImpl : public WebKit::WebMediaPlayer,
     // Methods for Filter -> WebMediaPlayerImpl communication.
     void Repaint();
     void SetVideoRenderer(scoped_refptr<WebVideoRenderer> video_renderer);
-    void AddDataSource(scoped_refptr<WebDataSource> data_source);
+    WebDataSourceBuildObserverHack* GetBuildObserver();
 
     // Methods for WebMediaPlayerImpl -> Filter communication.
     void Paint(skia::PlatformCanvas* canvas, const gfx::Rect& dest_rect);
@@ -125,6 +125,9 @@ class WebMediaPlayerImpl : public WebKit::WebMediaPlayer,
 
     virtual ~Proxy();
 
+    // Adds a data source to data_sources_.
+    void AddDataSource(WebDataSource* data_source);
+
     // Invoke |webmediaplayer_| to perform a repaint.
     void RepaintTask();
 
@@ -150,6 +153,7 @@ class WebMediaPlayerImpl : public WebKit::WebMediaPlayer,
     base::Lock data_sources_lock_;
     typedef std::list<scoped_refptr<WebDataSource> > DataSourceList;
     DataSourceList data_sources_;
+    scoped_ptr<WebDataSourceBuildObserverHack> build_observer_;
 
     scoped_refptr<WebVideoRenderer> video_renderer_;