A minor change to the pattern for filter factories that makes the implementation more flexible, and prevents the use of templates by the client.

Now, the filters themselves simply expose a static FilterFactory() method that returns a new factory object.  The constructor and destructor of the class
should be private and the FilterFactory object shuold be declarded as a friend.  There is still a template factory, but the class uses this template class
directly (if they want to us it) to implement their factory.
Review URL: http://codereview.chromium.org/19435

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

4 files changed, 388 insertions, 308 deletions

diff --git a/media/base/factory.h b/media/base/factory.h
index 39ed351..41a73aa 100644
--- a/media/base/factory.h
+++ b/media/base/factory.h
@@ -43,12 +43,13 @@ class FilterFactory : public base::RefCountedThreadSafe<FilterFactory> {
  public:
   // Creates a filter implementing the specified interface.  Hides the casting
   // and FilterType constants from the callers and produces cleaner code:
-  //   AudioDecoder* filter = NULL;
-  //   bool success = Create<AudioDecoder>(media_format, &filter);
-  template <class T>
-  bool Create(const MediaFormat* media_format, T** filter_out) {
-    return Create(T::filter_type(), media_format,
-                  reinterpret_cast<MediaFilter**>(filter_out));
+  //   socped_refptr<MyAudioDecoder> d = Create<MyAudioDecoder>(media_format);
+  // If the factory does not support the specific filter type or does not
+  // support the |media_format| then NULL is returned.
+  template <class Filter>
+  Filter* Create(const MediaFormat* media_format) {
+    return reinterpret_cast<Filter*>(Create(Filter::filter_type(),
+                                            media_format));
   }
 
  protected:
@@ -56,51 +57,19 @@ class FilterFactory : public base::RefCountedThreadSafe<FilterFactory> {
   friend class FilterFactoryCollection;
 
   // Attempt to create a filter of the given type using the information stored
-  // in |media_format|.  If successful, the filter is assigned to |filter_out|
-  // and the method returns true.  If the filter cannot be created for any
-  // reason, |filter_out| is assigned NULL and false it returned.
+  // in |media_format|.  If successful, the filter is returned.  If the filter
+  // cannot be created for any reason, NULL is returned.
   //
-  // It is assumed that |filter_out| can be safely casted to the corresponding
-  // interface type (i.e., FILTER_AUDIO_DECODER -> AudioDecoder).
-  virtual bool Create(FilterType filter_type, const MediaFormat* media_format,
-                      MediaFilter** filter_out) = 0;
+  // It is assumed that the MediaFilter interface can be safely cast to the
+  // corresponding interface type (i.e., FILTER_AUDIO_DECODER -> AudioDecoder).
+  virtual MediaFilter* Create(FilterType filter_type,
+                              const MediaFormat* media_format) = 0;
 
   friend class base::RefCountedThreadSafe<FilterFactory>;
   virtual ~FilterFactory() {}
 };
 
 
-// Helper template class for implementing trivial filter factories.  If your
-// filter does not require any special handling during creation, you can create
-// a factory for it using this class.  It requires the following static method:
-//   bool Create(MediaFormat* media_format, YourFilterType** filter_out)
-//
-// You can create the filter factory like so:
-//   new TypeFilterFactory<YourFilterType>()
-template <class Filter>
-class TypeFilterFactory : public FilterFactory {
- public:
-  TypeFilterFactory() {}
-
- protected:
-  // Attempts to create a filter of the template type.  Assumes a static method
-  // Create is declared.
-  virtual bool Create(FilterType filter_type, const MediaFormat* media_format,
-                      MediaFilter** filter_out) {
-    Filter* filter;
-    if (Filter::filter_type() == filter_type &&
-        Filter::Create(media_format, &filter)) {
-      *filter_out = filter;
-      return true;
-    }
-    return false;
-  }
-
- private:
-  DISALLOW_COPY_AND_ASSIGN(TypeFilterFactory);
-};
-
-
 // Maintains a collection of FilterFactories.
 class FilterFactoryCollection : public FilterFactory {
  public:
@@ -113,16 +82,14 @@ class FilterFactoryCollection : public FilterFactory {
 
  protected:
   // Attempts to create a filter by walking down the list of filter factories.
-  bool Create(FilterType filter_type, const MediaFormat* media_format,
-              MediaFilter** filter_out) {
+  MediaFilter* Create(FilterType filter_type, const MediaFormat* media_format) {
+    MediaFilter* filter = NULL;
     for (FactoryVector::iterator factory = factories_.begin();
-         factory != factories_.end();
+         !filter && factory != factories_.end();
          ++factory) {
-      if ((*factory)->Create(filter_type, media_format, filter_out)) {
-        return true;
-      }
+      filter = (*factory)->Create(filter_type, media_format);
     }
-    return false;
+    return filter;
   }
 
  private:
@@ -132,6 +99,63 @@ class FilterFactoryCollection : public FilterFactory {
   DISALLOW_COPY_AND_ASSIGN(FilterFactoryCollection);
 };
 
+//-----------------------------------------------------------------------------
+
+// This template is used by classes to implement a type-safe filter factory.
+// If the derived class needs to examine the |media_format| passed to the
+// Create method then they should implement the static method
+// IsMediaFormatSupported.  Classes should implement their contructor as private
+// and make FilterFactoryImpl<MyClass> a friend class.
+template <class Filter>
+class FilterFactoryImpl0 : public FilterFactory {
+ public:
+  FilterFactoryImpl0() {}
+
+ protected:
+  virtual MediaFilter* Create(FilterType filter_type,
+                              const MediaFormat* media_format) {
+    Filter* filter = NULL;
+    if (Filter::filter_type() == filter_type &&
+        Filter::IsMediaFormatSupported(media_format)) {
+      filter = new Filter();
+    }
+    return filter;
+  }
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(FilterFactoryImpl0);
+};
+
+// This template can be used by classes that need to be constructed with a
+// parameter that needs to be used in the construction of the actual filter.
+// This would usually be a "parent" object which the instantiated filter needs
+// to communicate with.  The class's CreateFactory method would look like:
+//   static FilterFactory* CreateFactory(MyRequiredParentClass* parent) {
+//      return new FilterFactoryImpl1<MyClass>(parent);
+//   }
+// The class would be constructed with the same pointer passed to the
+// CreateFactory method.
+template <class Filter, class A>
+class FilterFactoryImpl1 : public FilterFactory {
+ public:
+  explicit FilterFactoryImpl1(A a) : a_(a) {}
+
+ protected:
+  virtual MediaFilter* Create(FilterType filter_type,
+                              const MediaFormat* media_format) {
+    Filter* filter = NULL;
+    if (Filter::filter_type() == filter_type &&
+        Filter::IsMediaFormatSupported(media_format)) {
+      filter = new Filter(a_);
+    }
+    return filter;
+  }
+
+ private:
+  A const a_;
+  DISALLOW_COPY_AND_ASSIGN(FilterFactoryImpl1);
+};
+
 }  // namespace media
 
 #endif  // MEDIA_BASE_FACTORY_H_
diff --git a/media/base/filters.h b/media/base/filters.h
index c1da89e..ea4a585 100644
--- a/media/base/filters.h
+++ b/media/base/filters.h
@@ -97,6 +97,12 @@ class DataSource : public MediaFilter {
     return FILTER_DATA_SOURCE;
   }
 
+  static bool IsMediaFormatSupported(const MediaFormat* media_format) {
+    std::string mime_type;
+    return (media_format->GetAsString(MediaFormat::kMimeType, &mime_type) &&
+            mime_type == mime_type::kURL);
+  }
+
   static const size_t kReadError = static_cast<size_t>(-1);
 
   // Initializes this filter, returns true if successful, false otherwise.
@@ -128,6 +134,12 @@ class Demuxer : public MediaFilter {
     return FILTER_DEMUXER;
   }
 
+  static bool IsMediaFormatSupported(const MediaFormat* media_format) {
+    std::string mime_type;
+    return (media_format->GetAsString(MediaFormat::kMimeType, &mime_type) &&
+            mime_type == mime_type::kApplicationOctetStream);
+  }
+
   // Initializes this filter, returns true if successful, false otherwise.
   virtual bool Initialize(DataSource* data_source) = 0;
 
@@ -148,7 +160,7 @@ class DemuxerStream {
   virtual void Read(Assignable<Buffer>* buffer) = 0;
 
  protected:
-  virtual ~DemuxerStream() = 0;
+  virtual ~DemuxerStream() {}
 };
 
 
diff --git a/media/base/pipeline_impl.cc b/media/base/pipeline_impl.cc
index e49e719..4f11d62 100644
--- a/media/base/pipeline_impl.cc
+++ b/media/base/pipeline_impl.cc
@@ -65,23 +65,23 @@ PipelineError PipelineImpl::GetError() const {
 bool PipelineImpl::Start(FilterFactory* factory,
                          const std::string& url,
                          Callback1<bool>::Type* init_complete_callback) {
-  DCHECK(!pipeline_thread_);

-  DCHECK(factory);

-  DCHECK(!initialized_);

-  if (!pipeline_thread_ && factory) {

-    pipeline_thread_ = new PipelineThread(this);

-    if (pipeline_thread_) {

-      // TODO(ralphl): Does the callback get copied by these fancy templates?

-      // if so, then do I want to always delete it here???

-      if (pipeline_thread_->Start(factory, url, init_complete_callback)) {

-        return true;

-      }

-      pipeline_thread_ = NULL;  // Releases reference to destroy thread

-    }

-  }

-  delete init_complete_callback;

-  return false;

-}

+  DCHECK(!pipeline_thread_);
+  DCHECK(factory);
+  DCHECK(!initialized_);
+  if (!pipeline_thread_ && factory) {
+    pipeline_thread_ = new PipelineThread(this);
+    if (pipeline_thread_) {
+      // TODO(ralphl): Does the callback get copied by these fancy templates?
+      // if so, then do I want to always delete it here???
+      if (pipeline_thread_->Start(factory, url, init_complete_callback)) {
+        return true;
+      }
+      pipeline_thread_ = NULL;  // Releases reference to destroy thread
+    }
+  }
+  delete init_complete_callback;
+  return false;
+}
 
 // Stop the PipelineThread and return to a state identical to that of a newly
 // created PipelineImpl object.
@@ -141,37 +141,37 @@ void PipelineImpl::SetVideoSize(size_t width, size_t height) {
 
 //-----------------------------------------------------------------------------
 
-PipelineThread::PipelineThread(PipelineImpl* pipeline)

-    : pipeline_(pipeline),

-      thread_("PipelineThread"),

-      time_update_callback_scheduled_(false),

-      host_initializing_(NULL) {

-}

-

-PipelineThread::~PipelineThread() {

-  Stop();

-}

-

-// This method is called on the client's thread.  It starts the pipeline's

-// dedicated thread and posts a task to call the StartTask method on that

-// thread.

-bool PipelineThread::Start(FilterFactory* filter_factory,

-                           const std::string& url,

-                           Callback1<bool>::Type* init_complete_callback) {

-  if (thread_.Start()) {

-    filter_factory->AddRef();

-    PostTask(NewRunnableMethod(this,

-                               &PipelineThread::StartTask,

-                               filter_factory,

-                               url,

-                               // TODO(ralphl): what happens to this callback?

-                               // is it copied by NewRunnableTask?  Just pointer

-                               // or is the callback itself copied?

-                               init_complete_callback));

-    return true;

-  }

-  return false;

-}

+PipelineThread::PipelineThread(PipelineImpl* pipeline)
+    : pipeline_(pipeline),
+      thread_("PipelineThread"),
+      time_update_callback_scheduled_(false),
+      host_initializing_(NULL) {
+}
+
+PipelineThread::~PipelineThread() {
+  Stop();
+}
+
+// This method is called on the client's thread.  It starts the pipeline's
+// dedicated thread and posts a task to call the StartTask method on that
+// thread.
+bool PipelineThread::Start(FilterFactory* filter_factory,
+                           const std::string& url,
+                           Callback1<bool>::Type* init_complete_callback) {
+  if (thread_.Start()) {
+    filter_factory->AddRef();
+    PostTask(NewRunnableMethod(this,
+                               &PipelineThread::StartTask,
+                               filter_factory,
+                               url,
+                               // TODO(ralphl): what happens to this callback?
+                               // is it copied by NewRunnableTask?  Just pointer
+                               // or is the callback itself copied?
+                               init_complete_callback));
+    return true;
+  }
+  return false;
+}
 
 // Called on the client's thread.  If the thread has been started, then posts
 // a task to call the StopTask method, then waits until the thread has stopped.
@@ -179,44 +179,44 @@ bool PipelineThread::Start(FilterFactory* filter_factory,
 // method.  This method waits for that Lock to be released so that we know
 // that the thread is not executing a nested message loop.  This way we know
 // that that Thread::Stop call will quit the appropriate message loop.
-void PipelineThread::Stop() {

-  if (thread_.IsRunning()) {

-    PostTask(NewRunnableMethod(this, &PipelineThread::StopTask));

-    AutoLock lock_crit(initialization_lock_);

-    thread_.Stop();

-  }

-  DCHECK(filter_hosts_.empty());

-}

-

-// Called on client's thread.

-void PipelineThread::SetPlaybackRate(float rate) {

-  PostTask(NewRunnableMethod(this, &PipelineThread::SetPlaybackRateTask, rate));

-}

-

-// Called on client's thread.

-void PipelineThread::Seek(base::TimeDelta time) {

-  PostTask(NewRunnableMethod(this, &PipelineThread::SeekTask, time));

-}

-
-// Called on client's thread.

+void PipelineThread::Stop() {
+  if (thread_.IsRunning()) {
+    PostTask(NewRunnableMethod(this, &PipelineThread::StopTask));
+    AutoLock lock_crit(initialization_lock_);
+    thread_.Stop();
+  }
+  DCHECK(filter_hosts_.empty());
+}
+
+// Called on client's thread.
+void PipelineThread::SetPlaybackRate(float rate) {
+  PostTask(NewRunnableMethod(this, &PipelineThread::SetPlaybackRateTask, rate));
+}
+
+// Called on client's thread.
+void PipelineThread::Seek(base::TimeDelta time) {
+  PostTask(NewRunnableMethod(this, &PipelineThread::SeekTask, time));
+}
+
+// Called on client's thread.
 void PipelineThread::SetVolume(float volume) {
   PostTask(NewRunnableMethod(this, &PipelineThread::SetVolumeTask, volume));
 }
 
-// May be called on any thread, and therefore we always assume the worst

-// possible race condition.  This could, for example, be called from a filter's

-// thread just as the pipeline thread is exiting the call to the filter's

-// Initialize() method.  Therefore, we make NO assumptions, and post work

-// in every case, even the trivial one of a thread calling this method from

-// within it's Initialize method.  This means that we will always run a nested

-// message loop, and the InitializationCompleteTask will Quit that loop

-// immediately in the trivial case.

+// May be called on any thread, and therefore we always assume the worst
+// possible race condition.  This could, for example, be called from a filter's
+// thread just as the pipeline thread is exiting the call to the filter's
+// Initialize() method.  Therefore, we make NO assumptions, and post work
+// in every case, even the trivial one of a thread calling this method from
+// within it's Initialize method.  This means that we will always run a nested
+// message loop, and the InitializationCompleteTask will Quit that loop
+// immediately in the trivial case.
 void PipelineThread::InitializationComplete(FilterHostImpl* host) {
-  DCHECK(host == host_initializing_);

-  PostTask(NewRunnableMethod(this,

-                             &PipelineThread::InitializationCompleteTask,

-                             host));

-}

+  DCHECK(host == host_initializing_);
+  PostTask(NewRunnableMethod(this,
+                             &PipelineThread::InitializationCompleteTask,
+                             host));
+}
 
 // Called from any thread.  Updates the pipeline time and schedules a task to
 // call back to filters that have registered a callback for time updates.
@@ -291,41 +291,41 @@ void PipelineThread::StartTask(FilterFactory* filter_factory,
   if (success && HasVideo()) {
     success = CreateDecoder<VideoDecoder>(filter_factory);
     if (success) {
-      success = CreateAndConnect<VideoRenderer, VideoDecoder>(filter_factory);

-    }

-  }

-  if (success) {

-    pipeline_->initialized_ = true;

-  } else if (PIPELINE_OK == pipeline_->error_) {

-    Error(PIPELINE_ERROR_INITIALIZATION_FAILED);

-  }

-

-  // No matter what, we're done with the filter factory, and

-  // client callback so get rid of them.

-  filter_factory->Release();

-  if (init_complete_callback) {

-    init_complete_callback->Run(success);

-    delete init_complete_callback;

-  }

-}

+      success = CreateAndConnect<VideoRenderer, VideoDecoder>(filter_factory);
+    }
+  }
+  if (success) {
+    pipeline_->initialized_ = true;
+  } else if (PIPELINE_OK == pipeline_->error_) {
+    Error(PIPELINE_ERROR_INITIALIZATION_FAILED);
+  }
+
+  // No matter what, we're done with the filter factory, and
+  // client callback so get rid of them.
+  filter_factory->Release();
+  if (init_complete_callback) {
+    init_complete_callback->Run(success);
+    delete init_complete_callback;
+  }
+}
 
 // This method is called as a result of the client calling Pipeline::Stop() or
 // as the result of an error condition.  If there is no error, then set the
 // pipeline's error_ member to PIPELINE_STOPPING.  We stop the filters in the
 // reverse order.
 void PipelineThread::StopTask() {
-  if (PIPELINE_OK == pipeline_->error_) {

-    pipeline_->error_ = PIPELINE_STOPPING;

-  }

+  if (PIPELINE_OK == pipeline_->error_) {
+    pipeline_->error_ = PIPELINE_STOPPING;
+  }
   FilterHostVector::reverse_iterator riter = filter_hosts_.rbegin();
   while (riter != filter_hosts_.rend()) {
     (*riter)->Stop();
-    ++riter;

-  }

-  if (host_initializing_) {

-    host_initializing_ = NULL;

-    message_loop()->Quit();

-  }

+    ++riter;
+  }
+  if (host_initializing_) {
+    host_initializing_ = NULL;
+    message_loop()->Quit();
+  }
 }
 
 // Task runs as a result of a filter calling InitializationComplete.  If for
@@ -345,155 +345,150 @@ void PipelineThread::SetPlaybackRateTask(float rate) {
   pipeline_->playback_rate_ = rate;
   FilterHostVector::iterator iter = filter_hosts_.begin();
   while (iter != filter_hosts_.end()) {
-    (*iter)->media_filter()->SetPlaybackRate(rate);

-    ++iter;

-  }

+    (*iter)->media_filter()->SetPlaybackRate(rate);
+    ++iter;
+  }
 }
 
 void PipelineThread::SeekTask(base::TimeDelta time) {
   FilterHostVector::iterator iter = filter_hosts_.begin();
   while (iter != filter_hosts_.end()) {
-    (*iter)->media_filter()->Seek(time);

-    ++iter;

-  }

+    (*iter)->media_filter()->Seek(time);
+    ++iter;
+  }
 }
 
 void PipelineThread::SetVolumeTask(float volume) {
   pipeline_->volume_ = volume;
-  AudioRenderer* audio_renderer = GetFilter<AudioRenderer>();

-  if (audio_renderer) {

-    audio_renderer->SetVolume(volume);

-  }

-}

-
-void PipelineThread::SetTimeTask() {

-  time_update_callback_scheduled_ = false;

+  AudioRenderer* audio_renderer = GetFilter<AudioRenderer>();
+  if (audio_renderer) {
+    audio_renderer->SetVolume(volume);
+  }
+}
+
+void PipelineThread::SetTimeTask() {
+  time_update_callback_scheduled_ = false;
   FilterHostVector::iterator iter = filter_hosts_.begin();
   while (iter != filter_hosts_.end()) {
-    (*iter)->RunTimeUpdateCallback(pipeline_->time_);

-    ++iter;

-  }

-}

-

-template <class Filter>

-Filter* PipelineThread::GetFilter() const {

-  Filter* filter = NULL;

+    (*iter)->RunTimeUpdateCallback(pipeline_->time_);
+    ++iter;
+  }
+}
+
+template <class Filter>
+Filter* PipelineThread::GetFilter() const {
+  Filter* filter = NULL;
   FilterHostVector::const_iterator iter = filter_hosts_.begin();
   while (iter != filter_hosts_.end() && NULL == filter) {
     filter = (*iter)->GetFilter<Filter>();
     ++iter;
   }
   return filter;
-}

-

-template <class NewFilter, class Source>

-bool PipelineThread::CreateFilter(FilterFactory* filter_factory,

-                                  Source source,

-                                  const MediaFormat* source_media_format) {

-  NewFilter* new_filter;

-  bool success;

-  success = filter_factory->Create(source_media_format, &new_filter);

-  if (success) {

-    DCHECK(!host_initializing_);

-    host_initializing_ = new FilterHostImpl(this, new_filter);

-    if (!host_initializing_) {

-      success = false;

-      new_filter->AddRef();

-      new_filter->Release();

-    }

-  }

-  if (success) {

-    filter_hosts_.push_back(host_initializing_);

-    new_filter->SetFilterHost(host_initializing_);

-

-    // The filter must return true from initialize and there must still not

-    // be an error or it's not successful.

-    success = (new_filter->Initialize(source) &&

-               PIPELINE_OK == pipeline_->error_);

-  }

-  if (success) {

-    // Now we run the thread's message loop recursively.  We want all

-    // pending tasks to be processed, so we set nestable tasks to be allowed

-    // and then run the loop.  The only way we exit the loop is as the result

-    // of a call to FilterHost::InitializationComplete, FilterHost::Error, or

-    // Pipeline::Stop.  In each of these cases, the corresponding task method

-    // sets host_initializing_ to NULL to signal that the message loop's Quit

-    // method has already been called, and then calls message_loop()->Quit().

-    // The setting of |host_initializing_| to NULL in the task prevents a

-    // subsequent task from accidentally quitting the wrong (non-nested) loop.

-    message_loop()->SetNestableTasksAllowed(true);

-    message_loop()->Run();

-    message_loop()->SetNestableTasksAllowed(false);

-    DCHECK(!host_initializing_);

-

-    // If an error occurred while we were in the nested Run state, then

-    // not successful.  When stopping, the |error_| member is set to a value of

-    // PIPELINE_STOPPING so we will exit in that case also with false.

-    success = (PIPELINE_OK == pipeline_->error_);

-  }

-

-  // This could still be set if we never ran the message loop (for example,

-  // if the fiter returned false from it's Initialize method), so make sure

-  // to reset it.

-  host_initializing_ = NULL;

-

-  // If this method fails, but no error set, then indicate a general

-  // initialization failure.

-  if (PIPELINE_OK == pipeline_->error_ && (!success)) {

-    Error(PIPELINE_ERROR_INITIALIZATION_FAILED);

-  }

-  return success;

-}

-
-bool PipelineThread::CreateDataSource(FilterFactory* filter_factory,

-                                      const std::string& url) {

-  MediaFormat url_format;

-  url_format.SetAsString(MediaFormat::kMimeType, mime_type::kURL);

-  url_format.SetAsString(MediaFormat::kURL, url);

-  return CreateFilter<DataSource>(filter_factory, url, &url_format);

-}

-
-template <class Decoder>

-bool PipelineThread::CreateDecoder(FilterFactory* filter_factory) {

-  Demuxer* demuxer = GetFilter<Demuxer>();

-  if (demuxer) {

-    int num_outputs = demuxer->GetNumberOfStreams();

-    for (int i = 0; i < num_outputs; ++i) {

-      DemuxerStream* stream = demuxer->GetStream(i);

-      const MediaFormat* stream_format = stream->GetMediaFormat();

-      if (IsMajorMimeType(stream_format, Decoder::major_mime_type())) {

-        return CreateFilter<Decoder>(filter_factory, stream, stream_format);

-      }

-    }

-  }

-  return false;

-}

-

-template <class NewFilter, class SourceFilter>

-bool PipelineThread::CreateAndConnect(FilterFactory* filter_factory) {

-  SourceFilter* source_filter = GetFilter<SourceFilter>();

-  bool success = (source_filter &&

-                  CreateFilter<NewFilter>(filter_factory,

-                                          source_filter,

-                                          source_filter->GetMediaFormat()));

-  return success;

+}
+
+template <class Filter, class Source>
+bool PipelineThread::CreateFilter(FilterFactory* filter_factory,
+                                  Source source,
+                                  const MediaFormat* media_format) {
+  scoped_refptr<Filter> filter = filter_factory->Create<Filter>(media_format);
+  bool success = (NULL != filter);
+  if (success) {
+    DCHECK(!host_initializing_);
+    host_initializing_ = new FilterHostImpl(this, filter.get());
+    success = (NULL != host_initializing_);
+  }
+  if (success) {
+    filter_hosts_.push_back(host_initializing_);
+    filter->SetFilterHost(host_initializing_);
+
+    // The filter must return true from initialize and there must still not
+    // be an error or it's not successful.
+    success = (filter->Initialize(source) &&
+               PIPELINE_OK == pipeline_->error_);
+  }
+  if (success) {
+    // Now we run the thread's message loop recursively.  We want all
+    // pending tasks to be processed, so we set nestable tasks to be allowed
+    // and then run the loop.  The only way we exit the loop is as the result
+    // of a call to FilterHost::InitializationComplete, FilterHost::Error, or
+    // Pipeline::Stop.  In each of these cases, the corresponding task method
+    // sets host_initializing_ to NULL to signal that the message loop's Quit
+    // method has already been called, and then calls message_loop()->Quit().
+    // The setting of |host_initializing_| to NULL in the task prevents a
+    // subsequent task from accidentally quitting the wrong (non-nested) loop.
+    message_loop()->SetNestableTasksAllowed(true);
+    message_loop()->Run();
+    message_loop()->SetNestableTasksAllowed(false);
+    DCHECK(!host_initializing_);
+
+    // If an error occurred while we were in the nested Run state, then
+    // not successful.  When stopping, the |error_| member is set to a value of
+    // PIPELINE_STOPPING so we will exit in that case also with false.
+    success = (PIPELINE_OK == pipeline_->error_);
+  }
+
+  // This could still be set if we never ran the message loop (for example,
+  // if the fiter returned false from it's Initialize method), so make sure
+  // to reset it.
+  host_initializing_ = NULL;
+
+  // If this method fails, but no error set, then indicate a general
+  // initialization failure.
+  if (PIPELINE_OK == pipeline_->error_ && (!success)) {
+    Error(PIPELINE_ERROR_INITIALIZATION_FAILED);
+  }
+  return success;
+}
+
+bool PipelineThread::CreateDataSource(FilterFactory* filter_factory,
+                                      const std::string& url) {
+  MediaFormat url_format;
+  url_format.SetAsString(MediaFormat::kMimeType, mime_type::kURL);
+  url_format.SetAsString(MediaFormat::kURL, url);
+  return CreateFilter<DataSource>(filter_factory, url, &url_format);
+}
+
+template <class Decoder>
+bool PipelineThread::CreateDecoder(FilterFactory* filter_factory) {
+  Demuxer* demuxer = GetFilter<Demuxer>();
+  if (demuxer) {
+    int num_outputs = demuxer->GetNumberOfStreams();
+    for (int i = 0; i < num_outputs; ++i) {
+      DemuxerStream* stream = demuxer->GetStream(i);
+      const MediaFormat* stream_format = stream->GetMediaFormat();
+      if (IsMajorMimeType(stream_format, Decoder::major_mime_type())) {
+        return CreateFilter<Decoder>(filter_factory, stream, stream_format);
+      }
+    }
+  }
+  return false;
+}
+
+template <class NewFilter, class SourceFilter>
+bool PipelineThread::CreateAndConnect(FilterFactory* filter_factory) {
+  SourceFilter* source_filter = GetFilter<SourceFilter>();
+  bool success = (source_filter &&
+                  CreateFilter<NewFilter>(filter_factory,
+                                          source_filter,
+                                          source_filter->GetMediaFormat()));
+  return success;
 }
 
 // TODO(ralphl): Consider making this part of the demuxer interface.
-bool PipelineThread::HasVideo() const {

-  Demuxer* demuxer = GetFilter<Demuxer>();

-  if (demuxer) {

-    int num_outputs = demuxer->GetNumberOfStreams();

-    for (int i = 0; i < num_outputs; ++i) {

-      if (IsMajorMimeType(demuxer->GetStream(i)->GetMediaFormat(),

-                          mime_type::kMajorTypeVideo)) {

-        return true;

-      }

-    }

-  }

-  return false;

-}

-

+bool PipelineThread::HasVideo() const {
+  Demuxer* demuxer = GetFilter<Demuxer>();
+  if (demuxer) {
+    int num_outputs = demuxer->GetNumberOfStreams();
+    for (int i = 0; i < num_outputs; ++i) {
+      if (IsMajorMimeType(demuxer->GetStream(i)->GetMediaFormat(),
+                          mime_type::kMajorTypeVideo)) {
+        return true;
+      }
+    }
+  }
+  return false;
+}
+
 bool PipelineThread::IsMajorMimeType(const MediaFormat* media_format,
                                      const std::string& major_mime_type) const {
   std::string value;
@@ -502,13 +497,13 @@ bool PipelineThread::IsMajorMimeType(const MediaFormat* media_format,
   }
   return false;
 }
-

-// Called as a result of destruction of the thread.

-void PipelineThread::WillDestroyCurrentMessageLoop() {

-  while (!filter_hosts_.empty()) {

-    delete filter_hosts_.back();

-    filter_hosts_.pop_back();

-  }

-}

+
+// Called as a result of destruction of the thread.
+void PipelineThread::WillDestroyCurrentMessageLoop() {
+  while (!filter_hosts_.empty()) {
+    delete filter_hosts_.back();
+    filter_hosts_.pop_back();
+  }
+}
 
 }  // namespace media
diff --git a/media/base/pipeline_impl_unittest.cc b/media/base/pipeline_impl_unittest.cc
index 43a113c..082f926 100644
--- a/media/base/pipeline_impl_unittest.cc
+++ b/media/base/pipeline_impl_unittest.cc
@@ -5,6 +5,8 @@
 #include <string>
 
 #include "base/platform_thread.h"
+#include "base/time.h"
+#include "base/waitable_event.h"
 #include "media/base/pipeline_impl.h"
 #include "media/base/media_format.h"
 #include "media/base/filters.h"
@@ -17,19 +19,20 @@ using media::FilterFactoryCollection;
 using media::FilterHost;
 using media::MediaFormat;
 using media::PipelineImpl;
-using media::TypeFilterFactory;
+using media::FilterFactoryImpl1;
 
 class TestDataSource : public media::DataSource {
  public:
-  static bool Create(const MediaFormat* media_format,
-                     TestDataSource** filter_out) {
-    *filter_out = new TestDataSource();
-    return true;
+  static FilterFactory* CreateFactory(bool hang_in_init) {
+     return new FilterFactoryImpl1<TestDataSource, bool>(hang_in_init);
   }
   virtual void Stop() {}
   // This filter will hang in initialization because it never calls
   // FilterHost::InitializationComplete
   virtual bool Initialize(const std::string& uri) {
+    if (!hang_in_init_) {
+      host_->InitializationComplete();
+    }
     return true;
   }
   virtual const MediaFormat* GetMediaFormat() {
@@ -47,14 +50,60 @@ class TestDataSource : public media::DataSource {
   virtual bool GetSize(int64* size_out) {
     return 0;
   }
+
+ protected:
+  bool hang_in_init_;
+  friend class media::FilterFactoryImpl1<TestDataSource, bool>;
+  explicit TestDataSource(bool hang_in_init) : hang_in_init_(hang_in_init) {}
+  ~TestDataSource() {}
+
+  DISALLOW_COPY_AND_ASSIGN(TestDataSource);
+};
+
+
+class InitWaiter : public base::WaitableEvent {
+ public:
+  InitWaiter()
+    : WaitableEvent(true, false),
+      callback_success_status_(false) {}
+  void InitializationComplete(bool success) {
+    callback_success_status_ = success;
+    Signal();
+  }
+  bool HasBeenCalled() { return IsSignaled(); }
+  bool CallbackSuccessStatus() { return callback_success_status_; }
+  Callback1<bool>::Type* NewInitCallback() {
+    return NewCallback(this, &InitWaiter::InitializationComplete);
+  }
+  void Reset() {
+    base::WaitableEvent::Reset();
+    callback_success_status_ = false;
+  }
+
+ private:
+  bool callback_success_status_;
+
+  DISALLOW_COPY_AND_ASSIGN(InitWaiter);
 };
 
+
 TEST(PipelineImplTest, Basic) {
-  std::string uri("test.mov");
-  PipelineImpl pipeline;
-  scoped_refptr<FilterFactoryCollection> f = new FilterFactoryCollection();
-  f->AddFactory(new TypeFilterFactory<TestDataSource>);
-  pipeline.Start(f, uri, NULL);
-  PlatformThread::Sleep(10);
-  pipeline.Stop();
+  std::string url("test.mov");
+  PipelineImpl p;
+  InitWaiter w;
+  p.Start(TestDataSource::CreateFactory(true), url, w.NewInitCallback());
+  w.TimedWait(base::TimeDelta::FromSeconds(1));
+  EXPECT_FALSE(w.HasBeenCalled());
+  EXPECT_FALSE(p.IsInitialized());
+  EXPECT_TRUE(media::PIPELINE_OK == p.GetError());
+  p.Stop();
+
+  w.Reset();
+  p.Start(TestDataSource::CreateFactory(false), url, w.NewInitCallback());
+  w.TimedWait(base::TimeDelta::FromSeconds(1));
+  EXPECT_TRUE(w.HasBeenCalled());
+  EXPECT_FALSE(w.CallbackSuccessStatus());
+  EXPECT_FALSE(p.IsInitialized());
+  EXPECT_FALSE(media::PIPELINE_OK == p.GetError());
+  p.Stop();
 }