Demuxer stream now uses a callback instead of Assignable template. Cleaning up Pipeine interfaces to conform with Chrome Base design.

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

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

6 files changed, 74 insertions, 57 deletions

diff --git a/media/base/filters.h b/media/base/filters.h
index 6290e28..e36d81a 100644
--- a/media/base/filters.h
+++ b/media/base/filters.h
@@ -29,6 +29,7 @@
 
 #include "base/logging.h"
 #include "base/ref_counted.h"
+#include "base/task.h"
 #include "base/time.h"
 #include "media/base/media_format.h"
 
@@ -156,8 +157,9 @@ class DemuxerStream : public base::RefCountedThreadSafe<DemuxerStream> {
   // Returns the MediaFormat for this filter.
   virtual const MediaFormat& media_format() = 0;
 
-  // Schedules a read and takes ownership of the given buffer.
-  virtual void Read(Assignable<Buffer>* buffer) = 0;
+  // Schedules a read.  When the |read_callback| is called, the downstream
+  // filter takes ownership of the buffer by AddRef()'ing the buffer.
+  virtual void Read(Callback1<Buffer*>::Type* read_callback) = 0;
 
   // Given a class that supports the |Interface| and a related static method
   // interface_id(), which returns a const char*, this method returns true if
diff --git a/media/base/mock_media_filters.h b/media/base/mock_media_filters.h
index 4695a65..1f7af93 100644
--- a/media/base/mock_media_filters.h
+++ b/media/base/mock_media_filters.h
@@ -214,7 +214,7 @@ class MockDemuxerStream : public DemuxerStream {
     return media_format_;
   }
 
-  virtual void Read(Assignable<Buffer>* buffer) {
+  virtual void Read(Callback1<Buffer*>::Type* read_callback) {
     NOTREACHED();  // TODO(ralphl): fix me!!
   }
 
diff --git a/media/filters/decoder_base.h b/media/filters/decoder_base.h
index adecbf2..76da44b 100644
--- a/media/filters/decoder_base.h
+++ b/media/filters/decoder_base.h
@@ -70,8 +70,7 @@ class DecoderBase : public Decoder {
     }
   }
 
-  // AssignableBuffer callback.
-  virtual void OnAssignment(Buffer* buffer) {
+  void OnReadComplete(Buffer* buffer) {
     AutoLock auto_lock(lock_);
     if (IsRunning()) {
       buffer->AddRef();
@@ -183,7 +182,7 @@ class DecoderBase : public Decoder {
         AutoUnlock unlock(lock_);
         while (read) {
           demuxer_stream_->
-              Read(new AssignableBuffer<DecoderBase, Buffer>(this));
+              Read(NewCallback(this, &DecoderBase::OnReadComplete));
           --read;
         }
       }
diff --git a/media/filters/ffmpeg_demuxer.cc b/media/filters/ffmpeg_demuxer.cc
index 0446325..906c067 100644
--- a/media/filters/ffmpeg_demuxer.cc
+++ b/media/filters/ffmpeg_demuxer.cc
@@ -88,8 +88,11 @@ FFmpegDemuxerStream::FFmpegDemuxerStream(FFmpegDemuxer* demuxer,
 }
 
 FFmpegDemuxerStream::~FFmpegDemuxerStream() {
-  // Since |input_queue_| and |output_queue_| use scoped_refptr everything
-  // should get released.
+  // Since |buffer_queue_| uses scoped_refptr everything will get released.
+  while (!read_queue_.empty()) {
+    delete read_queue_.front();
+    read_queue_.pop_front();
+  }
 }
 
 // static
@@ -108,7 +111,7 @@ void* FFmpegDemuxerStream::QueryInterface(const char* id) {
 
 bool FFmpegDemuxerStream::HasPendingReads() {
   AutoLock auto_lock(lock_);
-  return !output_queue_.empty();
+  return !read_queue_.empty();
 }
 
 void FFmpegDemuxerStream::EnqueuePacket(AVPacket* packet) {
@@ -118,7 +121,7 @@ void FFmpegDemuxerStream::EnqueuePacket(AVPacket* packet) {
   DCHECK(buffer);
   {
     AutoLock auto_lock(lock_);
-    input_queue_.push_back(buffer);
+    buffer_queue_.push_back(buffer);
   }
   FulfillPendingReads();
 }
@@ -127,11 +130,11 @@ const MediaFormat& FFmpegDemuxerStream::media_format() {
   return media_format_;
 }
 
-void FFmpegDemuxerStream::Read(Assignable<Buffer>* buffer) {
-  DCHECK(buffer);
+void FFmpegDemuxerStream::Read(Callback1<Buffer*>::Type* read_callback) {
+  DCHECK(read_callback);
   {
     AutoLock auto_lock(lock_);
-    output_queue_.push_back(scoped_refptr< Assignable<Buffer> >(buffer));
+    read_queue_.push_back(read_callback);
   }
   if (FulfillPendingReads()) {
     demuxer_->ScheduleDemux();
@@ -141,21 +144,20 @@ void FFmpegDemuxerStream::Read(Assignable<Buffer>* buffer) {
 bool FFmpegDemuxerStream::FulfillPendingReads() {
   bool pending_reads = false;
   while (true) {
-    scoped_refptr<Buffer> buffer_in;
-    scoped_refptr< Assignable<Buffer> > buffer_out;
+    scoped_refptr<Buffer> buffer;
+    scoped_ptr<Callback1<Buffer*>::Type> read_callback;
     {
       AutoLock auto_lock(lock_);
-      pending_reads = !output_queue_.empty();
-      if (input_queue_.empty() || output_queue_.empty()) {
+      pending_reads = !read_queue_.empty();
+      if (buffer_queue_.empty() || read_queue_.empty()) {
         break;
       }
-      buffer_in = input_queue_.front();
-      buffer_out = output_queue_.front();
-      input_queue_.pop_front();
-      output_queue_.pop_front();
+      buffer = buffer_queue_.front();
+      read_callback.reset(read_queue_.front());
+      buffer_queue_.pop_front();
+      read_queue_.pop_front();
     }
-    buffer_out->SetBuffer(buffer_in);
-    buffer_out->OnAssignment();
+    read_callback->Run(buffer);
   }
   return pending_reads;
 }
diff --git a/media/filters/ffmpeg_demuxer.h b/media/filters/ffmpeg_demuxer.h
index dea349d..bf03c2e 100644
--- a/media/filters/ffmpeg_demuxer.h
+++ b/media/filters/ffmpeg_demuxer.h
@@ -58,7 +58,7 @@ class FFmpegDemuxerStream : public DemuxerStream {
 
   // DemuxerStream implementation.
   virtual const MediaFormat& media_format();
-  virtual void Read(Assignable<Buffer>* buffer);
+  virtual void Read(Callback1<Buffer*>::Type* read_callback);
 
   AVStream* av_stream() {
     return av_stream_;
@@ -80,11 +80,11 @@ class FFmpegDemuxerStream : public DemuxerStream {
   base::TimeDelta duration_;
   Lock lock_;
 
-  typedef std::deque< scoped_refptr<Buffer> > InputQueue;
-  InputQueue input_queue_;
+  typedef std::deque< scoped_refptr<Buffer> > BufferQueue;
+  BufferQueue buffer_queue_;
 
-  typedef std::deque< scoped_refptr< Assignable<Buffer> > > OutputQueue;
-  OutputQueue output_queue_;
+  typedef std::deque<Callback1<Buffer*>::Type*> ReadQueue;
+  ReadQueue read_queue_;
 
   DISALLOW_COPY_AND_ASSIGN(FFmpegDemuxerStream);
 };
diff --git a/media/filters/ffmpeg_demuxer_unittest.cc b/media/filters/ffmpeg_demuxer_unittest.cc
index aeba6fe..21c0c45 100644
--- a/media/filters/ffmpeg_demuxer_unittest.cc
+++ b/media/filters/ffmpeg_demuxer_unittest.cc
@@ -112,29 +112,43 @@ void InitializeFFmpegMocks() {
   memset(&g_packet, 0, sizeof(g_packet));
 }
 
-// Simple implementation of Assignable<Buffer> that lets us poke at values.
-class TestBuffer : public Assignable<Buffer> {
+// Ref counted object so we can create callbacks to call DemuxerStream::Read().
+class TestReader : public base::RefCountedThreadSafe<TestReader> {
  public:
-  TestBuffer() : assigned_(false) {}
-  virtual ~TestBuffer() {}
+  TestReader() : called_(false), expecting_call_(false) {}
+  virtual ~TestReader() {}
+
+  void Reset() {
+    EXPECT_FALSE(expecting_call_);
+    expecting_call_ = false;
+    called_ = false;
+    buffer_ = NULL;
+  }
 
-  // Assignable<Buffer> implementation.
-  virtual void SetBuffer(Buffer* buffer) {
-    buffer_ = buffer;
+  void Read(DemuxerStream* stream) {
+    EXPECT_FALSE(expecting_call_);
+    called_ = false;
+    expecting_call_ = true;
+    stream->Read(NewCallback(this, &TestReader::ReadComplete));
   }
 
-  void OnAssignment() {
-    EXPECT_FALSE(assigned_);
-    assigned_ = true;
+  void ReadComplete(Buffer* buffer) {
+    EXPECT_FALSE(called_);
+    EXPECT_TRUE(expecting_call_);
+    expecting_call_ = false;
+    called_ = true;
+    buffer_ = buffer;
   }
 
   // Mock getters/setters.
   Buffer* buffer() { return buffer_; }
-  bool assigned() { return assigned_; }
+  bool called() { return called_; }
+  bool expecting_call() { return expecting_call_; }
 
  private:
   scoped_refptr<Buffer> buffer_;
-  bool assigned_;
+  bool called_;
+  bool expecting_call_;
 };
 
 }  // namespace
@@ -336,13 +350,13 @@ TEST(FFmpegDemuxerTest, Read) {
   g_packet.size = kDataSize;
 
   // Attempt a read from the audio stream and run the message loop until done.
-  scoped_refptr<TestBuffer> buffer(new TestBuffer());
-  audio_stream->Read(buffer);
+  scoped_refptr<TestReader> reader(new TestReader());
+  reader->Read(audio_stream);
   pipeline.RunAllTasks();
-  EXPECT_TRUE(buffer->assigned());
-  EXPECT_TRUE(buffer->buffer());
-  EXPECT_EQ(audio_data, buffer->buffer()->GetData());
-  EXPECT_EQ(kDataSize, buffer->buffer()->GetDataSize());
+  EXPECT_TRUE(reader->called());
+  EXPECT_TRUE(reader->buffer());
+  EXPECT_EQ(audio_data, reader->buffer()->GetData());
+  EXPECT_EQ(kDataSize, reader->buffer()->GetDataSize());
 
   // Prepare our test video packet.
   g_packet.stream_index = kVideo;
@@ -350,25 +364,25 @@ TEST(FFmpegDemuxerTest, Read) {
   g_packet.size = kDataSize;
 
   // Attempt a read from the video stream and run the message loop until done.
-  buffer = new TestBuffer();
-  video_stream->Read(buffer);
+  reader->Reset();
+  reader->Read(video_stream);
   pipeline.RunAllTasks();
-  EXPECT_TRUE(buffer->assigned());
-  EXPECT_TRUE(buffer->buffer());
-  EXPECT_EQ(video_data, buffer->buffer()->GetData());
-  EXPECT_EQ(kDataSize, buffer->buffer()->GetDataSize());
+  EXPECT_TRUE(reader->called());
+  EXPECT_TRUE(reader->buffer());
+  EXPECT_EQ(video_data, reader->buffer()->GetData());
+  EXPECT_EQ(kDataSize, reader->buffer()->GetDataSize());
 
   // Simulate end of stream.
   g_av_read_frame = AVERROR_IO;
 
   // Attempt a read from the audio stream and run the message loop until done.
-  buffer = new TestBuffer();
-  audio_stream->Read(buffer);
+  reader->Reset();
+  reader->Read(audio_stream);
   pipeline.RunAllTasks();
-  EXPECT_FALSE(buffer->assigned());
-  EXPECT_FALSE(buffer->buffer());
+  EXPECT_FALSE(reader->called());
+  EXPECT_FALSE(reader->buffer());
 
   // Manually release buffer, which should release any remaining AVPackets.
-  buffer = NULL;
+  reader = NULL;
   EXPECT_EQ(0, g_oustanding_packets);
 }