Implementation of Pipeline and FilterHost interfaces. This is a large change, but all of the objects are interrelated.

I am also checking in a basic unit test that creates pipeline,
and the data source hangs during initialization.  The test sleeps one second and then stops the pipeline.

Andrew has already done a first pass on this, and the code has come largely from our working experimental branch.
Review URL: http://codereview.chromium.org/18546

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

17 files changed, 989 insertions, 152 deletions

diff --git a/base/thread.cc b/base/thread.cc
index 387b1c5..f447fbb 100644
--- a/base/thread.cc
+++ b/base/thread.cc
@@ -1,4 +1,4 @@
-// Copyright (c) 2006-2008 The Chromium Authors. All rights reserved.
+// Copyright (c) 2006-2009 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.
 
@@ -29,7 +29,9 @@ struct Thread::StartupData {
   // Used to synchronize thread startup.
   WaitableEvent event;
 
-  StartupData(const Options& opt) : options(opt), event(false, false) {}
+  explicit StartupData(const Options& opt)
+      : options(opt),
+        event(false, false) {}
 };
 
 Thread::Thread(const char *name)
@@ -161,6 +163,7 @@ void Thread::ThreadMain() {
 
   // We can't receive messages anymore.
   message_loop_ = NULL;
+  thread_id_ = 0;
 }
 
 }  // namespace base
diff --git a/base/thread.h b/base/thread.h
index 13aa35b..5a51bec 100644
--- a/base/thread.h
+++ b/base/thread.h
@@ -1,4 +1,4 @@
-// Copyright (c) 2006-2008 The Chromium Authors. All rights reserved.
+// Copyright (c) 2006-2009 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.
 
@@ -22,7 +22,7 @@ class Thread : PlatformThread::Delegate {
   struct Options {
     // Specifies the type of message loop that will be allocated on the thread.
     MessageLoop::Type message_loop_type;
-    
+
     // Specifies the maximum stack size that the thread is allowed to use.
     // This does not necessarily correspond to the thread's initial stack size.
     // A value of 0 indicates that the default maximum should be used.
@@ -104,6 +104,10 @@ class Thread : PlatformThread::Delegate {
   // The thread ID.
   PlatformThreadId thread_id() const { return thread_id_; }
 
+  // Returns true if the thread has been started, and not yet stopped.
+  // When a thread is running, the thread_id_ is non-zero.
+  bool IsRunning() const { return thread_id_ != 0; }
+
  protected:
   // Called just prior to starting the message loop
   virtual void Init() {}
diff --git a/base/thread_unittest.cc b/base/thread_unittest.cc
index f741951..f72c9c4 100644
--- a/base/thread_unittest.cc
+++ b/base/thread_unittest.cc
@@ -57,16 +57,22 @@ TEST_F(ThreadTest, Restart) {
   Thread a("Restart");
   a.Stop();
   EXPECT_FALSE(a.message_loop());
+  EXPECT_FALSE(a.IsRunning());
   EXPECT_TRUE(a.Start());
   EXPECT_TRUE(a.message_loop());
+  EXPECT_TRUE(a.IsRunning());
   a.Stop();
   EXPECT_FALSE(a.message_loop());
+  EXPECT_FALSE(a.IsRunning());
   EXPECT_TRUE(a.Start());
   EXPECT_TRUE(a.message_loop());
+  EXPECT_TRUE(a.IsRunning());
   a.Stop();
   EXPECT_FALSE(a.message_loop());
+  EXPECT_FALSE(a.IsRunning());
   a.Stop();
   EXPECT_FALSE(a.message_loop());
+  EXPECT_FALSE(a.IsRunning());
 }
 
 TEST_F(ThreadTest, StartWithOptions_StackSize) {
@@ -77,6 +83,7 @@ TEST_F(ThreadTest, StartWithOptions_StackSize) {
   options.stack_size = 12*1024;
   EXPECT_TRUE(a.StartWithOptions(options));
   EXPECT_TRUE(a.message_loop());
+  EXPECT_TRUE(a.IsRunning());
 
   bool was_invoked = false;
   a.message_loop()->PostTask(FROM_HERE, new ToggleValue(&was_invoked));
@@ -110,10 +117,12 @@ TEST_F(ThreadTest, StopSoon) {
   Thread a("StopSoon");
   EXPECT_TRUE(a.Start());
   EXPECT_TRUE(a.message_loop());
+  EXPECT_TRUE(a.IsRunning());
   a.StopSoon();
   a.StopSoon();
   a.Stop();
   EXPECT_FALSE(a.message_loop());
+  EXPECT_FALSE(a.IsRunning());
 }
 
 TEST_F(ThreadTest, ThreadName) {
diff --git a/media/base/factory.h b/media/base/factory.h
index 0c8dadd..39ed351 100644
--- a/media/base/factory.h
+++ b/media/base/factory.h
@@ -47,7 +47,7 @@ class FilterFactory : public base::RefCountedThreadSafe<FilterFactory> {
   //   bool success = Create<AudioDecoder>(media_format, &filter);
   template <class T>
   bool Create(const MediaFormat* media_format, T** filter_out) {
-    return Create(T::kFilterType, media_format,
+    return Create(T::filter_type(), media_format,
                   reinterpret_cast<MediaFilter**>(filter_out));
   }
 
@@ -77,7 +77,7 @@ class FilterFactory : public base::RefCountedThreadSafe<FilterFactory> {
 //
 // You can create the filter factory like so:
 //   new TypeFilterFactory<YourFilterType>()
-template <class T>
+template <class Filter>
 class TypeFilterFactory : public FilterFactory {
  public:
   TypeFilterFactory() {}
@@ -87,9 +87,10 @@ class TypeFilterFactory : public FilterFactory {
   // Create is declared.
   virtual bool Create(FilterType filter_type, const MediaFormat* media_format,
                       MediaFilter** filter_out) {
-    T* typed_out;
-    if (T::kFilterType == filter_type && T::Create(media_format, &typed_out)) {
-      *filter_out = typed_out;
+    Filter* filter;
+    if (Filter::filter_type() == filter_type &&
+        Filter::Create(media_format, &filter)) {
+      *filter_out = filter;
       return true;
     }
     return false;
diff --git a/media/base/filter_host.h b/media/base/filter_host.h
index a3e2185..437b3f8 100644
--- a/media/base/filter_host.h
+++ b/media/base/filter_host.h
@@ -43,8 +43,8 @@ class FilterHost {
   // may call this method passing NULL for the callback argument.
   //
   // Callback arguments:
-  //   int64    the new pipeline time, in microseconds
-  virtual void SetTimeUpdateCallback(Callback1<int64>::Type* callback) = 0;
+  //   base::TimeDelta - the new pipeline time, in microseconds.
+  virtual void SetTimeUpdateCallback(Callback1<base::TimeDelta>::Type* cb) = 0;
 
   // Filters must call this method to indicate that their initialization is
   // complete.  They may call this from within their Initialize() method or may
@@ -59,15 +59,15 @@ class FilterHost {
 
   // Sets the current time.  Any filters that have registered a callback through
   // the SetTimeUpdateCallback method will be notified of the change.
-  virtual void SetTime(int64 time) = 0;
+  virtual void SetTime(base::TimeDelta time) = 0;
 
   // Get the duration of the media in microseconds.  If the duration has not
   // been determined yet, then returns 0.
-  virtual void SetDuration(int64 duration) = 0;
+  virtual void SetDuration(base::TimeDelta duration) = 0;
 
   // Set the approximate amount of playable data buffered so far in micro-
   // seconds.
-  virtual void SetBufferedTime(int64 buffered_time) = 0;
+  virtual void SetBufferedTime(base::TimeDelta buffered_time) = 0;
 
   // Set the total size of the media file.
   virtual void SetTotalBytes(int64 total_bytes) = 0;
@@ -80,7 +80,7 @@ class FilterHost {
   virtual void SetVideoSize(size_t width, size_t height) = 0;
 
  protected:
-  virtual ~FilterHost() = 0;
+  virtual ~FilterHost() {}
 };
 
 }  // namespace media
diff --git a/media/base/filter_host_impl.cc b/media/base/filter_host_impl.cc
index 2b2d071..5e7522e 100644
--- a/media/base/filter_host_impl.cc
+++ b/media/base/filter_host_impl.cc
@@ -2,70 +2,70 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-#include "base/task.h"
 #include "media/base/filter_host_impl.h"
 
 namespace media {
 
-FilterHostImpl::FilterHostImpl() {
-  // TODO(ralphl): implement FilterHostImpl constructor.
-  NOTIMPLEMENTED();
+void FilterHostImpl::SetTimeUpdateCallback(
+      Callback1<base::TimeDelta>::Type* callback) {
+  time_update_callback_.reset(callback);
 }
 
-const PipelineStatus* FilterHostImpl::GetPipelineStatus() const {
-  // TODO(ralphl): implement GetPipelineStatus.
-  NOTIMPLEMENTED();
-  return NULL;
-}
-
-void FilterHostImpl::SetTimeUpdateCallback(Callback1<int64>::Type* callback) {
-  // TODO(ralphl): implement SetTimeUpdateCallback.
-  NOTIMPLEMENTED();
+void FilterHostImpl::RunTimeUpdateCallback(base::TimeDelta time) {
+  if (time_update_callback_.get())
+    time_update_callback_->Run(time);
 }
 
 void FilterHostImpl::InitializationComplete() {
-  // TODO(ralphl): implement InitializationComplete.
-  NOTIMPLEMENTED();
+  pipeline_thread_->InitializationComplete(this);
 }
-

+
 void FilterHostImpl::PostTask(Task* task) {
-  // TODO(ralphl): implement PostTask.
-  NOTIMPLEMENTED();
+  DCHECK(!stopped_);
+  if (stopped_) {
+    delete task;
+  } else {
+    pipeline_thread_->PostTask(task);
+  }
 }
 
 void FilterHostImpl::Error(PipelineError error) {
-  // TODO(ralphl): implement Error.
-  NOTIMPLEMENTED();
+  pipeline_thread_->Error(error);
 }
 
-void FilterHostImpl::SetTime(int64 time) {
-  // TODO(ralphl): implement SetTime.
-  NOTIMPLEMENTED();
+void FilterHostImpl::SetTime(base::TimeDelta time) {
+  pipeline_thread_->SetTime(time);
 }
 
-void FilterHostImpl::SetDuration(int64 duration) {
-  // TODO(ralphl): implement SetDuration.
-  NOTIMPLEMENTED();
+void FilterHostImpl::SetDuration(base::TimeDelta duration) {
+  pipeline()->duration_ = duration;
 }
 
-void FilterHostImpl::SetBufferedTime(int64 buffered_time) {
-  // TODO(ralphl): implement SetBufferedTime.
-  NOTIMPLEMENTED();
+void FilterHostImpl::SetBufferedTime(base::TimeDelta buffered_time) {
+  pipeline()->buffered_time_ = buffered_time;
 }
 
 void FilterHostImpl::SetTotalBytes(int64 total_bytes) {
-  // TODO(ralphl): implement.
-  NOTIMPLEMENTED();
+  pipeline()->total_bytes_ = total_bytes;
 }
 
 void FilterHostImpl::SetBufferedBytes(int64 buffered_bytes) {
-  // TODO(ralphl): implement.
-  NOTIMPLEMENTED();
+  pipeline()->buffered_bytes_ = buffered_bytes;
+}
+
+void FilterHostImpl::SetVideoSize(size_t width, size_t height) {
+  pipeline()->SetVideoSize(width, height);
+}
+
+const PipelineStatus* FilterHostImpl::GetPipelineStatus() const {
+  return pipeline();
 }
 
-void SetVideoSize(size_t width, size_t height) {
-  // TODO(ralphl): implement.
-  NOTIMPLEMENTED();
+void FilterHostImpl::Stop() {
+  if (!stopped_) {
+    filter_->Stop();
+    stopped_ = true;
+  }
 }
 
 }  // namespace media
diff --git a/media/base/filter_host_impl.h b/media/base/filter_host_impl.h
index 1112289..bdbb5e1 100644
--- a/media/base/filter_host_impl.h
+++ b/media/base/filter_host_impl.h
@@ -9,30 +9,81 @@
 
 #include "base/task.h"
 #include "media/base/filter_host.h"
+#include "media/base/pipeline_impl.h"
 
 namespace media {
 
 class FilterHostImpl : public FilterHost {
  public:
-  FilterHostImpl();
-
   // FilterHost interface.
   virtual const PipelineStatus* GetPipelineStatus() const;
-  virtual void SetTimeUpdateCallback(Callback1<int64>::Type* callback);
-  virtual void InitializationComplete();

+  virtual void SetTimeUpdateCallback(Callback1<base::TimeDelta>::Type* cb);
+  virtual void InitializationComplete();
   virtual void PostTask(Task* task);
   virtual void Error(PipelineError error);
-  virtual void SetTime(int64 time);
-  virtual void SetDuration(int64 duration);
-  virtual void SetBufferedTime(int64 buffered_time);
+  virtual void SetTime(base::TimeDelta time);
+  virtual void SetDuration(base::TimeDelta duration);
+  virtual void SetBufferedTime(base::TimeDelta buffered_time);
   virtual void SetTotalBytes(int64 total_bytes);
   virtual void SetBufferedBytes(int64 buffered_bytes);
   virtual void SetVideoSize(size_t width, size_t height);
 
- protected:
-  virtual ~FilterHostImpl() {}
+  // These methods are public, but are intended for use by the
+  // PipelineThread class only.
+
+  // Creates a FilterHostImpl object and populates the filter_type_ member
+  // by calling the Filter class's static filter_type() method.  This ensures
+  // that the GetFilter method can safely cast the filter interface from the
+  // MediaFilter base class interface to the specific Filter interface.
+  template <class Filter>
+  FilterHostImpl(PipelineThread* pipeline_thread, Filter* filter)
+      : pipeline_thread_(pipeline_thread),
+        filter_type_(Filter::filter_type()),
+        filter_(filter),
+        stopped_(false) {
+  }
+  ~FilterHostImpl() {}
+
+  // If this FilterHost contains a filter of the specifed Filter class, then
+  // this method returns a pointer to the interface, otherwise it returns NULL.
+  template <class Filter>
+  Filter* GetFilter() const {
+    Filter* filter = NULL;
+    if (Filter::filter_type() == filter_type_) {
+      filter = reinterpret_cast<Filter*>(media_filter());
+    }
+    return filter;
+  }
+
+  // Call the filter if it has registered a time update callback if the filter
+  // has registered one though the FilterHost::SetTimeUpdateCallback method.
+  void RunTimeUpdateCallback(base::TimeDelta time);
+
+  // Stops the filter.
+  void Stop();
+
+  // Used by the PipelineThread to call Seek and SetRate methods on filters.
+  MediaFilter* media_filter() const { return filter_; }
 
  private:
+  PipelineImpl* pipeline() const { return pipeline_thread_->pipeline(); }
+
+  // PipelineThread that owns this FilterHostImpl.
+  PipelineThread* const pipeline_thread_;
+
+  // The FilterType of the filter this host contains.
+  FilterType const filter_type_;
+
+  // A pointer to the filter's MediaFilter base interface.
+  scoped_refptr<MediaFilter> filter_;
+
+  // An optional callback that will be called when the time is updated.
+  scoped_ptr<Callback1<base::TimeDelta>::Type> time_update_callback_;
+
+  // Used to avoid calling Filter's Stop method multiplie times.  It is also
+  // used to prevent a filter that has been stopped from calling PostTask.
+  bool stopped_;
+
   DISALLOW_COPY_AND_ASSIGN(FilterHostImpl);
 };
 
diff --git a/media/base/filters.h b/media/base/filters.h
index e447f5c..c1da89e 100644
--- a/media/base/filters.h
+++ b/media/base/filters.h
@@ -29,6 +29,8 @@
 
 #include "base/logging.h"
 #include "base/ref_counted.h"
+#include "base/time.h"
+#include "media/base/media_format.h"
 
 namespace media {
 
@@ -37,7 +39,6 @@ class Buffer;
 class Decoder;
 class DemuxerStream;
 class FilterHost;
-class MediaFormat;
 class VideoFrame;
 class WritableBuffer;
 
@@ -78,7 +79,7 @@ class MediaFilter : public base::RefCountedThreadSafe<MediaFilter> {
 
   // The pipeline is being seeked to the specified time.  Filters may implement
   // this method if they need to respond to this call.
-  virtual void Seek(int64 time) {}
+  virtual void Seek(base::TimeDelta time) {}
 
  protected:
   FilterHost* host_;
@@ -92,11 +93,14 @@ class MediaFilter : public base::RefCountedThreadSafe<MediaFilter> {
 
 class DataSource : public MediaFilter {
  public:
-  static const FilterType kFilterType = FILTER_DATA_SOURCE;
+  static const FilterType filter_type() {
+    return FILTER_DATA_SOURCE;
+  }
+
   static const size_t kReadError = static_cast<size_t>(-1);
 
   // Initializes this filter, returns true if successful, false otherwise.
-  virtual bool Initialize(const std::string& uri) = 0;
+  virtual bool Initialize(const std::string& url) = 0;
 
   // Returns the MediaFormat for this filter.
   virtual const MediaFormat* GetMediaFormat() = 0;
@@ -120,7 +124,9 @@ class DataSource : public MediaFilter {
 
 class Demuxer : public MediaFilter {
  public:
-  static const FilterType kFilterType = FILTER_DEMUXER;
+  static const FilterType filter_type() {
+    return FILTER_DEMUXER;
+  }
 
   // Initializes this filter, returns true if successful, false otherwise.
   virtual bool Initialize(DataSource* data_source) = 0;
@@ -148,7 +154,13 @@ class DemuxerStream {
 
 class VideoDecoder : public MediaFilter {
  public:
-  static const FilterType kFilterType = FILTER_VIDEO_DECODER;
+  static const FilterType filter_type() {
+    return FILTER_VIDEO_DECODER;
+  }
+
+  static const char* major_mime_type() {
+    return mime_type::kMajorTypeVideo;
+  }
 
   // Initializes this filter, returns true if successful, false otherwise.
   virtual bool Initialize(DemuxerStream* demuxer_stream) = 0;
@@ -163,7 +175,13 @@ class VideoDecoder : public MediaFilter {
 
 class AudioDecoder : public MediaFilter {
  public:
-  static const FilterType kFilterType = FILTER_AUDIO_DECODER;
+  static const FilterType filter_type() {
+    return FILTER_AUDIO_DECODER;
+  }
+
+  static const char* major_mime_type() {
+    return mime_type::kMajorTypeAudio;
+  }
 
   // Initializes this filter, returns true if successful, false otherwise.
   virtual bool Initialize(DemuxerStream* demuxer_stream) = 0;
@@ -178,7 +196,13 @@ class AudioDecoder : public MediaFilter {
 
 class VideoRenderer : public MediaFilter {
  public:
-  static const FilterType kFilterType = FILTER_VIDEO_RENDERER;
+  static const FilterType filter_type() {
+    return FILTER_VIDEO_RENDERER;
+  }
+
+  static const char* major_mime_type() {
+    return mime_type::kMajorTypeVideo;
+  }
 
   // Initializes this filter, returns true if successful, false otherwise.
   virtual bool Initialize(VideoDecoder* decoder) = 0;
@@ -187,7 +211,13 @@ class VideoRenderer : public MediaFilter {
 
 class AudioRenderer : public MediaFilter {
  public:
-  static const FilterType kFilterType = FILTER_AUDIO_RENDERER;
+  static const FilterType filter_type() {
+    return FILTER_AUDIO_RENDERER;
+  }
+
+  static const char* major_mime_type() {
+    return mime_type::kMajorTypeAudio;
+  }
 
   // Initializes this filter, returns true if successful, false otherwise.
   virtual bool Initialize(AudioDecoder* decoder) = 0;
diff --git a/media/base/media_format.cc b/media/base/media_format.cc
index 304956f..543ada8 100644
--- a/media/base/media_format.cc
+++ b/media/base/media_format.cc
@@ -8,10 +8,10 @@ namespace media {
 
 namespace mime_type {
 
-// Represents a URI, typically used to create a DataSourceInterface.
+// Represents a URL, typically used to create a DataSourceInterface.
 // Expected keys:
-//   kUri             String          The URI
-const char kURI[]                           = "text/x-uri";
+//   kURL             String          The URL
+const char kURL[]                           = "text/x-url";
 
 // Represents a generic byte stream, typically from a DataSourceInterface.
 // Expected keys:
@@ -44,11 +44,15 @@ const char kUncompressedAudio[]             = "audio/x-uncompressed";
 //   kHeight          Integer         Display height of the surface
 const char kUncompressedVideo[]             = "video/x-uncompressed";
 
+// Major types of media types begin with the prefix "audio/" or "video/".
+const char kMajorTypeVideo[]                = "video/";
+const char kMajorTypeAudio[]                = "audio/";
+
 }  // namespace mime_type
 
 // Common keys.
 const char MediaFormat::kMimeType[]         = "MimeType";
-const char MediaFormat::kURI[]              = "Uri";
+const char MediaFormat::kURL[]              = "URL";
 const char MediaFormat::kSurfaceFormat[]    = "SurfaceFormat";
 const char MediaFormat::kSampleRate[]       = "SampleRate";
 const char MediaFormat::kSampleBits[]       = "SampleBits";
diff --git a/media/base/media_format.h b/media/base/media_format.h
index 6df9f17..298d68c 100644
--- a/media/base/media_format.h
+++ b/media/base/media_format.h
@@ -5,19 +5,24 @@
 #ifndef MEDIA_BASE_MEDIA_FORMAT_H_
 #define MEDIA_BASE_MEDIA_FORMAT_H_
 
+#include <map>
+#include <string>
+
 #include "base/values.h"
 
 namespace media {
 
 // Common MIME types.
 namespace mime_type {
-extern const char kURI[];
+extern const char kURL[];
 extern const char kApplicationOctetStream[];
 extern const char kMPEGAudio[];
 extern const char kAACAudio[];
 extern const char kH264AnnexB[];
 extern const char kUncompressedAudio[];
 extern const char kUncompressedVideo[];
+extern const char kMajorTypeAudio[];
+extern const char kMajorTypeVideo[];
 }  // namespace mime_type
 
 // MediaFormat is used to describe the output of a MediaFilterInterface to
@@ -40,7 +45,7 @@ class MediaFormat {
  public:
   // Common keys.
   static const char kMimeType[];
-  static const char kURI[];
+  static const char kURL[];
   static const char kSurfaceFormat[];
   static const char kSampleRate[];
   static const char kSampleBits[];
diff --git a/media/base/pipeline.h b/media/base/pipeline.h
index 99bdf49..50e1a06 100644
--- a/media/base/pipeline.h
+++ b/media/base/pipeline.h
@@ -12,6 +12,7 @@
 #include <string>
 
 #include "base/task.h"
+#include "base/time.h"
 #include "media/base/factory.h"
 
 namespace media {
@@ -40,11 +41,11 @@ class PipelineStatus {
 
   // Get the duration of the media in microseconds.  If the duration has not
   // been determined yet, then returns 0.
-  virtual int64 GetDuration() const = 0;
+  virtual base::TimeDelta GetDuration() const = 0;
 
   // Get the approximate amount of playable data buffered so far in micro-
   // seconds.
-  virtual int64 GetBufferedTime() const = 0;
+  virtual base::TimeDelta GetBufferedTime() const = 0;
 
   // Get the total size of the media file.  If the size has not yet been
   // determined or can not be determined, this value is 0.
@@ -71,14 +72,14 @@ class PipelineStatus {
 
   // Gets the current pipeline time in microseconds.  For a pipeline "time"
   // progresses from 0 to the end of the media.
-  virtual int64 GetTime() const = 0;
+  virtual base::TimeDelta GetTime() const = 0;
 
   // Gets the current error status for the pipeline.  If the pipeline is
   // operating correctly, this will return OK.
   virtual PipelineError GetError() const = 0;
 
  protected:
-  virtual ~PipelineStatus() = 0;
+  virtual ~PipelineStatus() {}
 };
 
 
@@ -116,29 +117,32 @@ class Pipeline : public PipelineStatus {
   // torn down and reset to an uninitialized state.  After calling Stop, it
   // is acceptable to call Start again since Stop leaves the pipeline
   // in a state identical to a newly created pipeline.
+  // Calling this method is not strictly required because the pipeline
+  // destructor will stop it pipeline if it has not been stopped already.
   virtual void Stop() = 0;
 
-  // Attempt to adjust the playback rate.  Returns true if successful,
-  // false otherwise.  Setting a playback rate of 0.0f pauses all rendering
-  // of the media.  A rate of 1.0f indicates a normal playback rate.  Values
-  // for the playback rate must be greater than or equal to 0.0f.
+  // Attempt to adjust the playback rate. Setting a playback rate of 0.0f pauses
+  // all rendering of the media.  A rate of 1.0f indicates a normal playback
+  // rate.  Values for the playback rate must be greater than or equal to 0.0f.
   // TODO(ralphl) What about maximum rate?  Does HTML5 specify a max?
   //
   // This method must be called only after initialization has completed.
-  virtual bool SetPlaybackRate(float playback_rate) = 0;
+  virtual void SetPlaybackRate(float playback_rate) = 0;
 
-  // Attempt to seek to the position in microseconds.  Returns true if
-  // successful, false otherwise.  Playback is paused after the seek completes.
+  // Attempt to seek to the position in microseconds.
   //
   // This method must be called only after initialization has completed.
-  virtual bool Seek(int64 time) = 0;
+  virtual void Seek(base::TimeDelta time) = 0;
 
   // Attempt to set the volume of the audio renderer.  Valid values for volume
   // range from 0.0f (muted) to 1.0f (full volume).  This value affects all
   // channels proportionately for multi-channel audio streams.
   //
   // This method must be called only after initialization has completed.
-  virtual bool SetVolume(float volume) = 0;
+  virtual void SetVolume(float volume) = 0;
+
+ protected:
+  virtual ~Pipeline() {}
 };
 
 }  // namespace media
diff --git a/media/base/pipeline_impl.cc b/media/base/pipeline_impl.cc
index 50e8776..e49e719 100644
--- a/media/base/pipeline_impl.cc
+++ b/media/base/pipeline_impl.cc
@@ -2,110 +2,513 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include "base/compiler_specific.h"
+#include "media/base/filter_host_impl.h"
+#include "media/base/media_format.h"
 #include "media/base/pipeline_impl.h"
 
 namespace media {
 
 PipelineImpl::PipelineImpl() {
-  // TODO(ralphl): implement PipelineImpl constructor.
-  NOTIMPLEMENTED();
+  ResetState();
 }
 
 PipelineImpl::~PipelineImpl() {
-  // TODO(ralphl): implement PipelineImpl destructor.
-  NOTIMPLEMENTED();
+  Stop();
 }
 
 bool PipelineImpl::IsInitialized() const {
-  // TODO(ralphl): implement IsInitialized.
-  NOTIMPLEMENTED();
-  return false;
+  return initialized_;
 }
 
-int64 PipelineImpl::GetDuration() const {
-  // TODO(ralphl): implement GetDuration.
-  NOTIMPLEMENTED();
-  return 0;
+base::TimeDelta PipelineImpl::GetDuration() const {
+  return duration_;
 }
 
-int64 PipelineImpl::GetBufferedTime() const {
-  // TODO(ralphl): implement GetBufferedTime.
-  NOTIMPLEMENTED();
-  return 0;
+base::TimeDelta PipelineImpl::GetBufferedTime() const {
+  return buffered_time_;
 }
 
 int64 PipelineImpl::GetTotalBytes() const {
-  // TODO(ralphl): implement GetTotalBytes.
-  NOTIMPLEMENTED();
-  return 0;
+  return total_bytes_;
 }
 
 int64 PipelineImpl::GetBufferedBytes() const {
-  // TODO(ralphl): implement GetBufferedBytes.
-  NOTIMPLEMENTED();
-  return 0;
+  return buffered_bytes_;
 }
 
 void PipelineImpl::GetVideoSize(size_t* width_out, size_t* height_out) const {
-  // TODO(ralphl): implement GetVideoSize.
-  NOTIMPLEMENTED();
-  width_out = 0;
-  height_out = 0;
+  DCHECK(width_out);
+  DCHECK(height_out);
+  AutoLock auto_lock(const_cast<Lock&>(video_size_access_lock_));
+  *width_out = video_width_;
+  *height_out = video_height_;
 }
 
 float PipelineImpl::GetVolume() const {
-  // TODO(ralphl): implement GetVolume.
-  NOTIMPLEMENTED();
-  return 0;
+  return volume_;
 }
 
 float PipelineImpl::GetPlaybackRate() const {
-  // TODO(ralphl): implement GetPlaybackRate.
-  NOTIMPLEMENTED();
-  return 0;
+  return playback_rate_;
 }
 
-int64 PipelineImpl::GetTime() const {
-  // TODO(ralphl): implement GetTime.
-  NOTIMPLEMENTED();
-  return 0;
+base::TimeDelta PipelineImpl::GetTime() const {
+  return time_;
 }
 
 PipelineError PipelineImpl::GetError() const {
-  // TODO(ralphl): implement GetError.
-  NOTIMPLEMENTED();
-  return PIPELINE_ERROR_INITIALIZATION_FAILED;
+  return error_;
 }
 
-bool PipelineImpl::Start(FilterFactory* filter_factory,
-                         const std::string& uri,
+// Creates the PipelineThread and calls it's start method.
+bool PipelineImpl::Start(FilterFactory* factory,
+                         const std::string& url,
                          Callback1<bool>::Type* init_complete_callback) {
-  // TODO(ralphl): implement Start.
-  NOTIMPLEMENTED();
-  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.
 void PipelineImpl::Stop() {
-  // TODO(ralphl): implement Stop.
-  NOTIMPLEMENTED();
+  if (pipeline_thread_) {
+    pipeline_thread_->Stop();
+  }
+  ResetState();
 }
 
-bool PipelineImpl::SetPlaybackRate(float rate) {
-  // TODO(ralphl): implement SetPlaybackRate.
-  NOTIMPLEMENTED();
-  return false;
+
+
+void PipelineImpl::SetPlaybackRate(float rate) {
+  if (OkToCallThread() && rate >= 0.0f) {
+    pipeline_thread_->SetPlaybackRate(rate);
+  } else {
+    NOTREACHED();
+  }
 }
 
-bool PipelineImpl::Seek(int64 time) {
-  // TODO(ralphl): implement Seek.
-  NOTIMPLEMENTED();
-  return false;
+void PipelineImpl::Seek(base::TimeDelta time) {
+  if (OkToCallThread()) {
+    pipeline_thread_->Seek(time);
+  } else {
+    NOTREACHED();
+  }
+}
+
+void PipelineImpl::SetVolume(float volume) {
+  if (OkToCallThread() && volume >= 0.0f && volume <= 1.0f) {
+    pipeline_thread_->SetVolume(volume);
+  } else {
+    NOTREACHED();
+  }
+}
+
+void PipelineImpl::ResetState() {
+  pipeline_thread_  = NULL;
+  initialized_      = false;
+  duration_         = base::TimeDelta();
+  buffered_time_    = base::TimeDelta();
+  buffered_bytes_   = 0;
+  total_bytes_      = 0;
+  video_width_      = 0;
+  video_height_     = 0;
+  volume_           = 0.0f;
+  playback_rate_    = 0.0f;
+  time_             = base::TimeDelta();
+  error_            = PIPELINE_OK;
+}
+
+void PipelineImpl::SetVideoSize(size_t width, size_t height) {
+  AutoLock auto_lock(video_size_access_lock_);
+  width = width;
+  height = 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;

+}

+
+// 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.
+// There is a critical section that wraps the entire duration of the StartTask
+// 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::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.

+void PipelineThread::InitializationComplete(FilterHostImpl* 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.
+void PipelineThread::SetTime(base::TimeDelta time) {
+  pipeline()->time_ = time;
+  if (!time_update_callback_scheduled_) {
+    time_update_callback_scheduled_ = true;
+    PostTask(NewRunnableMethod(this, &PipelineThread::SetTimeTask));
+  }
+}
+
+// Called from any thread.  Sets the pipeline error_ member and schedules a
+// task to stop all the filters in the pipeline.  Note that the thread will
+// continue to run until the client calls Pipeline::Stop, but nothing will
+// be processed since filters will not be able to post tasks.
+void PipelineThread::Error(PipelineError error) {
+  DCHECK(PIPELINE_OK != error);
+  if (PIPELINE_OK == pipeline()->error_) {
+    pipeline()->error_ = error;
+    PostTask(NewRunnableMethod(this, &PipelineThread::StopTask));
+  }
+}
+
+// Called from any thread.  Used by FilterHostImpl::PostTask method and used
+// internally.
+void PipelineThread::PostTask(Task* task) {
+  message_loop()->PostTask(FROM_HERE, task);
+}
+
+
+// Main initialization method called on the pipeline thread.  This code attempts
+// to use the specified filter factory to build a pipeline.  It starts by
+// creating a DataSource, connects it to a Demuxer, and then connects the
+// Demuxer's audio stream to an AudioDecoder which is then connected to an
+// AudioRenderer.  If the media has video, then it connects a VideoDecoder to
+// the Demuxer's video stream, and then connects the VideoDecoder to a
+// VideoRenderer.  When all required filters have been created and have called
+// their FilterHost's InitializationComplete method, the pipeline's
+// initialized_ member is set to true, and, if the client provided an
+// init_complete_callback, it is called with "true".
+// If initializatoin fails, the client's callback will still be called, but
+// the bool parameter passed to it will be false.
+//
+// Note that at each step in this process, the initialization of any filter
+// may require running the pipeline thread's message loop recursively.  This is
+// handled by the CreateFilter method.
+void PipelineThread::StartTask(FilterFactory* filter_factory,
+                               const std::string& url,
+                               Callback1<bool>::Type* init_complete_callback) {
+  bool success = true;
+
+  // During the entire StartTask we hold the initialization_lock_ so that
+  // if the client calls the Pipeline::Stop method while we are running a
+  // nested message loop, we can correctly unwind out of it before calling
+  // the Thread::Stop method.
+  AutoLock auto_lock(initialization_lock_);
+
+  // Add ourselves as a destruction observer of the thread's message loop so
+  // we can delete filters at an appropriate time (when all tasks have been
+  // processed and the thread is about to be destroyed).
+  message_loop()->AddDestructionObserver(this);
+  success = CreateDataSource(filter_factory, url);
+  if (success) {
+    success = CreateAndConnect<Demuxer, DataSource>(filter_factory);
+  }
+  if (success) {
+    success = CreateDecoder<AudioDecoder>(filter_factory);
+  }
+  if (success) {
+    success = CreateAndConnect<AudioRenderer, AudioDecoder>(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;

+  }

+}

+
+// 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;

+  }

+  FilterHostVector::reverse_iterator riter = filter_hosts_.rbegin();
+  while (riter != filter_hosts_.rend()) {
+    (*riter)->Stop();
+    ++riter;

+  }

+  if (host_initializing_) {

+    host_initializing_ = NULL;

+    message_loop()->Quit();

+  }

+}
+
+// Task runs as a result of a filter calling InitializationComplete.  If for
+// some reason StopTask has been executed prior to this, the host_initializing_
+// member will be NULL, and the message loop will have been quit already, so
+// we don't want to do it again.
+void PipelineThread::InitializationCompleteTask(FilterHostImpl* host) {
+  if (host == host_initializing_) {
+    host_initializing_ = NULL;
+    message_loop()->Quit();
+  } else {
+    DCHECK(!host_initializing_);
+  }
+}
+
+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;

+  }

+}
+
+void PipelineThread::SeekTask(base::TimeDelta time) {
+  FilterHostVector::iterator iter = filter_hosts_.begin();
+  while (iter != filter_hosts_.end()) {
+    (*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;

+  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;

+  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;

 }
 
-bool PipelineImpl::SetVolume(float volume) {
-  // TODO(ralphl): implement SetVolume.
-  NOTIMPLEMENTED();
+// 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::IsMajorMimeType(const MediaFormat* media_format,
+                                     const std::string& major_mime_type) const {
+  std::string value;
+  if (media_format->GetAsString(MediaFormat::kMimeType, &value)) {
+    return (0 == value.compare(0, major_mime_type.length(), major_mime_type));
+  }
   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();

+  }

+}

 
 }  // namespace media
diff --git a/media/base/pipeline_impl.h b/media/base/pipeline_impl.h
index 7c920e9..dc46c02 100644
--- a/media/base/pipeline_impl.h
+++ b/media/base/pipeline_impl.h
@@ -8,11 +8,24 @@
 #define MEDIA_BASE_PIPELINE_IMPL_H_
 
 #include <string>
+#include <vector>
 
+#include "base/message_loop.h"
+#include "base/ref_counted.h"
+#include "base/thread.h"
+#include "base/time.h"
 #include "media/base/pipeline.h"
 
 namespace media {
 
+class FilterHostImpl;
+class PipelineThread;
+
+// Class which implements the Media::Pipeline contract.  The majority of the
+// actual code for this object lives in the PipelineThread class, which is
+// responsible for actually building and running the pipeline.  This object
+// is basically a simple container for state information, and is responsible
+// for creating and communicating with the PipelineThread object.
 class PipelineImpl : public Pipeline {
  public:
   PipelineImpl();
@@ -20,29 +33,259 @@ class PipelineImpl : public Pipeline {
 
   // Implementation of PipelineStatus methods.
   virtual bool IsInitialized() const;
-  virtual int64 GetDuration() const;
-  virtual int64 GetBufferedTime() const;
+  virtual base::TimeDelta GetDuration() const;
+  virtual base::TimeDelta GetBufferedTime() const;
   virtual int64 GetTotalBytes() const;
   virtual int64 GetBufferedBytes()const;
   virtual void GetVideoSize(size_t* width_out, size_t* height_out) const;
   virtual float GetVolume() const;
   virtual float GetPlaybackRate() const;
-  virtual int64 GetTime() const;
+  virtual base::TimeDelta GetTime() const;
   virtual PipelineError GetError() const;
 
   // Impementation of Pipeline methods.
   virtual bool Start(FilterFactory* filter_factory,
-                     const std::string& uri,
+                     const std::string& url,
                      Callback1<bool>::Type* init_complete_callback);
   virtual void Stop();
-  virtual bool SetPlaybackRate(float rate);
-  virtual bool Seek(int64 time);
-  virtual bool SetVolume(float volume);
+  virtual void SetPlaybackRate(float rate);
+  virtual void Seek(base::TimeDelta time);
+  virtual void SetVolume(float volume);
 
  private:
+  friend class FilterHostImpl;
+  friend class PipelineThread;
+
+  // Reset the state of the pipline object to the initial state.  This method
+  // is used by the constructor, and the Stop method.
+  void ResetState();
+
+  // Used internally to make sure that the thread is in a state that is
+  // acceptable to post a task to.  It must exist, be initialized, and there
+  // must not be an error.
+  bool OkToCallThread() const {
+    return (pipeline_thread_ && initialized_ && PIPELINE_OK == error_);
+  }
+
+  // Called directly by the FilterHostImpl object to set the video size.
+  void SetVideoSize(size_t width, size_t height);
+
+  // Holds a ref counted reference to the PipelineThread object associated
+  // with this pipeline.  Prior to the call to the Start method, this member
+  // will be NULL, since no thread is running.
+  scoped_refptr<PipelineThread> pipeline_thread_;
+
+  // After calling Start, if all of the required filters are created and
+  // initialized, this member will be set to true by the pipeline thread.
+  bool initialized_;
+
+  // Duration of the media in microseconds.  Set by a FilterHostImpl object on
+  // behalf of a filter.
+  base::TimeDelta duration_;
+
+  // Amount of available buffered data in microseconds.  Set by a
+  // FilterHostImpl object on behalf of a filter.
+  base::TimeDelta buffered_time_;
+
+  // Amount of available buffered data.  Set by a FilterHostImpl object
+  // on behalf of a filter.
+  int64 buffered_bytes_;
+
+  // Total size of the media.  Set by a FilterHostImpl object on behalf
+  // of a filter.
+  int64 total_bytes_;
+
+  // Video width and height.  Set by a FilterHostImpl object on behalf
+  // of a filter.  The video_size_access_lock_ is used to make sure access
+  // to the pair of width and height are modified or read in thread safe way.
+  Lock video_size_access_lock_;
+  size_t video_width_;
+  size_t video_height_;
+
+  // Current volume level (from 0.0f to 1.0f).  The volume reflects the last
+  // value the audio filter was called with SetVolume, so there will be a short
+  // period of time between the client calling SetVolume on the pipeline and
+  // this value being updated.  Set by the PipelineThread just prior to calling
+  // the audio renderer.
+  float volume_;
+
+  // Current playback rate (>= 0.0f).  This member reflects the last value
+  // that the filters in the pipeline were called with, so there will be a short
+  // period of time between the client calling SetPlaybackRate and this value
+  // being updated.  Set by the PipelineThread just prior to calling filters.
+  float playback_rate_;
+
+  // Current playback time.  Set by a FilterHostImpl object on behalf of the
+  // audio renderer filter.
+  base::TimeDelta time_;
+
+  // Status of the pipeline.  Initialized to PIPELINE_OK which indicates that
+  // the pipeline is operating correctly. Any other value indicates that the
+  // pipeline is stopped or is stopping.  Clients can call the Stop method to
+  // reset the pipeline state, and restore this to PIPELINE_OK.
+  PipelineError error_;
+
   DISALLOW_COPY_AND_ASSIGN(PipelineImpl);
 };
 
+
+// The PipelineThread contains most of the logic involved with running the
+// media pipeline.  Filters are created and called on a dedicated thread owned
+// by this object.
+class PipelineThread : public base::RefCountedThreadSafe<PipelineThread>,
+                       public MessageLoop::DestructionObserver {
+ public:
+  // Methods called by PipelineImpl object on the client's thread.  These
+  // methods post a task to call a corresponding xxxTask() method on the
+  // pipeline thread.  For example, Seek posts a task to call SeekTask.
+  explicit PipelineThread(PipelineImpl* pipeline);
+
+  bool Start(FilterFactory* filter_factory,
+             const std::string& url_media_source,
+             Callback1<bool>::Type* init_complete_callback);
+  void Stop();
+  void SetPlaybackRate(float rate);
+  void Seek(base::TimeDelta time);
+  void SetVolume(float volume);
+
+  // Methods called by a FilterHostImpl object.  These methods may be called
+  // on any thread, either the pipeline's thread or any other.
+
+  // When a filter calls it's FilterHost, the filter host calls back to the
+  // pipeline thread.  If the pipeline thread is running a nested message loop
+  // then it will be exited.
+  void InitializationComplete(FilterHostImpl* host);
+
+  // Sets the pipeline time and schedules a task to call back to any filters
+  // that have registered a time update callback.
+  void SetTime(base::TimeDelta time);
+
+  // Called by a FilterHostImpl on behalf of a filter calling FilerHost::Error.
+  // If the pipeline is running a nested message loop, it will be exited.
+  void Error(PipelineError error);
+
+  // Called by a FilterHostImpl on behalf of a filter that calls the
+  // FilterHost::PostTask method.
+  void PostTask(Task* task);

+

+  // Simple accessor used by the FilterHostImpl class to get access to the

+  // pipeline object.

+  PipelineImpl* pipeline() const { return pipeline_; }
+
+ private:
+  // Implementation of MessageLoop::DestructionObserver.  StartTask registers
+  // this class as a destruction observer on the thread's message loop.
+  // It is used to destroy the list of FilterHosts
+  // (and thus destroy the associated filters) when all tasks have been
+  // processed and the message loop has been quit.
+  virtual void WillDestroyCurrentMessageLoop();
+
+  friend class base::RefCountedThreadSafe<PipelineThread>;
+  virtual ~PipelineThread();
+
+  MessageLoop* message_loop() const { return thread_.message_loop(); }

+

+  // The following "task" methods correspond to the public methods, but these

+  // methods are run as the result of posting a task to the PipelineThread's

+  // message loop.  For example, the Start method posts a task to call the

+  // StartTask message on the pipeline thread.

+  void StartTask(FilterFactory* filter_factory,
+                 const std::string& url,
+                 Callback1<bool>::Type* init_complete_callback);
+  void StopTask();
+  void SetPlaybackRateTask(float rate);
+  void SeekTask(base::TimeDelta time);
+  void SetVolumeTask(float volume);
+  void SetTimeTask();
+  void InitializationCompleteTask(FilterHostImpl* FilterHost);
+
+  // Internal methods used in the implementation of the pipeline thread.  All
+  // of these methods are only called on the pipeline thread.
+
+  // Calls the Stop method on every filter in the pipeline
+  void StopFilters();
+
+  // Examines the demuxer filter output streams.  If one contains video then
+  // returns true.
+  bool HasVideo() const;
+
+  // The following template funcions make use of the fact that media filter
+  // derived interfaces are self-describing in the sense that they all contain
+  // the static method filter_type() which returns a FilterType enum that
+  // uniquely identifies the filer's interface.  In addition, filters that are
+  // specific to audio or video also support a static method major_mime_type()
+  // which returns a string of "audio/" or "video/".
+
+  // Uses the FilterFactory to create a new filter of the NewFilter class, and
+  // initiaializes it using the Source object.  The source may be another filter
+  // or it could be a string in the case of a DataSource.
+  //
+  // The CreateFilter method actually does much more than simply creating the
+  // filter.  It creates the FilterHostImpl object, creates the filter using
+  // the filter factory, calls the MediaFilter::SetHost method on the filter,
+  // and then calls the filter's type-specific Initialize(source) method to
+  // initialize the filter.  It then runs the thread's message loop and waits
+  // until one of the following occurs:
+  //  1. The filter calls FilterHost::InitializationComplete()
+  //  2. A filter calls FilterHost::Error()
+  //  3. The client calls Pipeline::Stop()
+  template <class NewFilter, class Source>
+  bool CreateFilter(FilterFactory* filter_factory,
+                    Source source,
+                    const MediaFormat* source_media_format);
+
+  // Creates a DataSource (the first filter in a pipeline), and initializes it
+  // with the specified URL.
+  bool CreateDataSource(FilterFactory* filter_factory,
+                        const std::string& url);
+
+  // Examines the list of existing filters to find a Source, then creates a
+  // NewFilter, and initializes it with the Source filter.
+  template <class NewFilter, class Source>
+  bool CreateAndConnect(FilterFactory* filter_factory);
+
+  // Creates and initiaializes a decoder.
+  template <class Decoder>
+  bool CreateDecoder(FilterFactory* filter_factory);
+
+  // Examine the list of existing filters to find one that supports the
+  // specified Filter interface. If one exists, the specified Filter interface
+  // is returned otherwise the method retuns NULL.
+  template <class Filter>
+  Filter* GetFilter() const;
+
+  // Simple function that returns true if the specified MediaFormat object
+  // has a mime type that matches the major_mime_type.  Examples of major mime
+  // types are "audio/" and "video/"
+  bool IsMajorMimeType(const MediaFormat* media_format,
+                       const std::string& major_mime_type) const;
+
+  // Pointer to the pipeline that owns this PipelineThread.
+  PipelineImpl* const pipeline_;
+
+  // The actual thread.
+  base::Thread thread_;
+
+  // Used to avoid scheduling multiple time update tasks.  If this member is
+  // true then a task that will call the SetTimeTask() method is in the message
+  // loop's queue.
+  bool time_update_callback_scheduled_;
+
+  // During initialization of a filter, this member points to the FilterHostImpl
+  // that is being initialized.
+  FilterHostImpl* host_initializing_;
+
+  // This lock is held through the entire StartTask method to prevent the
+  // Stop method from quitting the nested message loop of the StartTask method.
+  Lock initialization_lock_;
+
+  // Vector of FilterHostImpl objects that contian the filters for the pipeline.
+  typedef std::vector<FilterHostImpl*> FilterHostVector;
+  FilterHostVector filter_hosts_;
+
+  DISALLOW_COPY_AND_ASSIGN(PipelineThread);
+};
+
 }  // namespace media
 
 #endif  // MEDIA_BASE_PIPELINE_IMPL_H_
diff --git a/media/base/pipeline_impl_unittest.cc b/media/base/pipeline_impl_unittest.cc
new file mode 100644
index 0000000..43a113c
--- /dev/null
+++ b/media/base/pipeline_impl_unittest.cc
@@ -0,0 +1,60 @@
+// Copyright (c) 2006-2009 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 <string>
+
+#include "base/platform_thread.h"
+#include "media/base/pipeline_impl.h"
+#include "media/base/media_format.h"
+#include "media/base/filters.h"
+#include "media/base/factory.h"
+#include "media/base/filter_host.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+using media::FilterFactory;
+using media::FilterFactoryCollection;
+using media::FilterHost;
+using media::MediaFormat;
+using media::PipelineImpl;
+using media::TypeFilterFactory;
+
+class TestDataSource : public media::DataSource {
+ public:
+  static bool Create(const MediaFormat* media_format,
+                     TestDataSource** filter_out) {
+    *filter_out = new TestDataSource();
+    return true;
+  }
+  virtual void Stop() {}
+  // This filter will hang in initialization because it never calls
+  // FilterHost::InitializationComplete
+  virtual bool Initialize(const std::string& uri) {
+    return true;
+  }
+  virtual const MediaFormat* GetMediaFormat() {
+    return NULL;  // TODO(ralphl):  Return octot thingie...
+  }
+  virtual size_t Read(char* data, size_t size) {
+    return 0;
+  }
+  virtual bool GetPosition(int64* position_out) {
+    return 0;
+  }
+  virtual bool SetPosition(int64 position) {
+    return true;
+  }
+  virtual bool GetSize(int64* size_out) {
+    return 0;
+  }
+};
+
+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();
+}
diff --git a/media/build/media_unittests.vcproj b/media/build/media_unittests.vcproj
index 575e7d8..333c899 100644
--- a/media/build/media_unittests.vcproj
+++ b/media/build/media_unittests.vcproj
@@ -159,6 +159,10 @@
 					RelativePath="..\base\data_buffer_unittest.cc"
 					>
 				</File>
+				<File
+					RelativePath="..\base\pipeline_impl_unittest.cc"
+					>
+				</File>
 			</Filter>
 			<Filter
 				Name="audio"
diff --git a/media/media.xcodeproj/project.pbxproj b/media/media.xcodeproj/project.pbxproj
index 4f8b384..488391e 100644
--- a/media/media.xcodeproj/project.pbxproj
+++ b/media/media.xcodeproj/project.pbxproj
@@ -51,6 +51,8 @@
 		DC5E24F60EFC4CF7006AD1A7 /* data_buffer_unittest.cc in Sources */ = {isa = PBXBuildFile; fileRef = DC5E24E30EFC4CC8006AD1A7 /* data_buffer_unittest.cc */; };
 		DC5E24F70EFC4CF7006AD1A7 /* run_all_unittests.cc in Sources */ = {isa = PBXBuildFile; fileRef = DC5E24E80EFC4CC8006AD1A7 /* run_all_unittests.cc */; };
 		DC66E6E30F13FE6A00E13554 /* filter_host_impl.cc in Sources */ = {isa = PBXBuildFile; fileRef = DC96BFAF0F13F19300668B64 /* filter_host_impl.cc */; };
+		DCB6EF0B0F2FDBE90084FC00 /* pipeline_impl.cc in Sources */ = {isa = PBXBuildFile; fileRef = DCB6EF090F2FDBE90084FC00 /* pipeline_impl.cc */; };
+		DCE176C60F2FDDF600C3FF5C /* pipeline_impl_unittest.cc in Sources */ = {isa = PBXBuildFile; fileRef = DCE176C50F2FDDF600C3FF5C /* pipeline_impl_unittest.cc */; };
 /* End PBXBuildFile section */
 
 /* Begin PBXContainerItemProxy section */
@@ -235,6 +237,11 @@
 		DC5E24E80EFC4CC8006AD1A7 /* run_all_unittests.cc */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = run_all_unittests.cc; sourceTree = "<group>"; };
 		DC96BFAF0F13F19300668B64 /* filter_host_impl.cc */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = filter_host_impl.cc; sourceTree = "<group>"; };
 		DC96BFB00F13F19300668B64 /* filter_host_impl.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; path = filter_host_impl.h; sourceTree = "<group>"; };
+		DCB6EF070F2FDBE90084FC00 /* factory.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; path = factory.h; sourceTree = "<group>"; };
+		DCB6EF080F2FDBE90084FC00 /* pipeline.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; path = pipeline.h; sourceTree = "<group>"; };
+		DCB6EF090F2FDBE90084FC00 /* pipeline_impl.cc */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = pipeline_impl.cc; sourceTree = "<group>"; };
+		DCB6EF0A0F2FDBE90084FC00 /* pipeline_impl.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; path = pipeline_impl.h; sourceTree = "<group>"; };
+		DCE176C50F2FDDF600C3FF5C /* pipeline_impl_unittest.cc */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = pipeline_impl_unittest.cc; sourceTree = "<group>"; };
 		E4AFA6230E523E2900201347 /* media_unittests */ = {isa = PBXFileReference; explicitFileType = "compiled.mach-o.executable"; includeInIndex = 0; path = media_unittests; sourceTree = BUILT_PRODUCTS_DIR; };
 		E4AFA62E0E5240A300201347 /* gtest.xcodeproj */ = {isa = PBXFileReference; lastKnownFileType = "wrapper.pb-project"; name = gtest.xcodeproj; path = testing/gtest.xcodeproj; sourceTree = "<group>"; };
 		E4AFA6450E5241D300201347 /* base.xcodeproj */ = {isa = PBXFileReference; lastKnownFileType = "wrapper.pb-project"; name = base.xcodeproj; path = base/base.xcodeproj; sourceTree = "<group>"; };
@@ -381,10 +388,15 @@
 				DC5E24E10EFC4CC8006AD1A7 /* data_buffer.cc */,
 				DC5E24E20EFC4CC8006AD1A7 /* data_buffer.h */,
 				DC5E24E30EFC4CC8006AD1A7 /* data_buffer_unittest.cc */,
+				DCB6EF070F2FDBE90084FC00 /* factory.h */,
 				DC5E24E40EFC4CC8006AD1A7 /* filter_host.h */,
 				DC96BFAF0F13F19300668B64 /* filter_host_impl.cc */,
 				DC96BFB00F13F19300668B64 /* filter_host_impl.h */,
 				DC5E24E50EFC4CC8006AD1A7 /* filters.h */,
+				DCB6EF080F2FDBE90084FC00 /* pipeline.h */,
+				DCB6EF090F2FDBE90084FC00 /* pipeline_impl.cc */,
+				DCB6EF0A0F2FDBE90084FC00 /* pipeline_impl.h */,
+				DCE176C50F2FDDF600C3FF5C /* pipeline_impl_unittest.cc */,
 				DC5E24E60EFC4CC8006AD1A7 /* media_format.cc */,
 				DC5E24E70EFC4CC8006AD1A7 /* media_format.h */,
 				DC5E24E80EFC4CC8006AD1A7 /* run_all_unittests.cc */,
@@ -611,6 +623,7 @@
 				DC5E24F40EFC4CE5006AD1A7 /* data_buffer.cc in Sources */,
 				DC66E6E30F13FE6A00E13554 /* filter_host_impl.cc in Sources */,
 				DC5E24F50EFC4CE5006AD1A7 /* media_format.cc in Sources */,
+				DCB6EF0B0F2FDBE90084FC00 /* pipeline_impl.cc in Sources */,
 			);
 			runOnlyForDeploymentPostprocessing = 0;
 		};
@@ -620,6 +633,7 @@
 			files = (
 				DC5E24F60EFC4CF7006AD1A7 /* data_buffer_unittest.cc in Sources */,
 				DC5E24F70EFC4CF7006AD1A7 /* run_all_unittests.cc in Sources */,
+				DCE176C60F2FDDF600C3FF5C /* pipeline_impl_unittest.cc in Sources */,
 			);
 			runOnlyForDeploymentPostprocessing = 0;
 		};
diff --git a/media/media_unittests.scons b/media/media_unittests.scons
index 984be8f..8930e95 100644
--- a/media/media_unittests.scons
+++ b/media/media_unittests.scons
@@ -48,6 +48,7 @@ if env.Bit('windows'):
 
 input_files = [
     'base/data_buffer_unittest.cc',
+    'base/pipeline_impl_unittest.cc',
     'base/run_all_unittests.cc',
 ]
 
@@ -61,6 +62,7 @@ input_files = ChromeFileList([
   MSVSFilter('tests', [
     MSVSFilter('base', [
       'base/data_buffer_unittest.cc',
+      'base/pipeline_impl_unittest.cc',
     ]),
     MSVSFilter('audio', [
       'audio/win/audio_output_win_unittest.cc',