3 files changed, 146 insertions, 13 deletions

diff --git a/media/filters/decoder_base.h b/media/filters/decoder_base.h
index 5729531..26b1d18 100644
--- a/media/filters/decoder_base.h
+++ b/media/filters/decoder_base.h
@@ -208,13 +208,13 @@ class DecoderBase : public Decoder {
 
     // Enqueue the callback and attempt to fulfill it immediately.
     read_queue_.push_back(read_callback);
-    FulfillPendingRead();
+    if (FulfillPendingRead())
+      return;
 
-    // Issue reads as necessary.
-    while (pending_reads_ < read_queue_.size()) {
-      demuxer_stream_->Read(NewCallback(this, &DecoderBase::OnReadComplete));
-      ++pending_reads_;
-    }
+    // Since we can't fulfill a read request now then submit a read
+    // request to the demuxer stream.
+    demuxer_stream_->Read(NewCallback(this, &DecoderBase::OnReadComplete));
+    ++pending_reads_;
   }
 
   void ReadCompleteTask(scoped_refptr<Buffer> buffer) {
@@ -239,17 +239,16 @@ class DecoderBase : public Decoder {
   void OnDecodeComplete() {
     // Attempt to fulfill a pending read callback and schedule additional reads
     // if necessary.
-    FulfillPendingRead();
+    bool fulfilled = FulfillPendingRead();
 
     // Issue reads as necessary.
     //
     // Note that it's possible for us to decode but not produce a frame, in
     // which case |pending_reads_| will remain less than |read_queue_| so we
     // need to schedule an additional read.
-    // TODO(hclam): Enable this line again to make sure we don't break the
-    // flow control. (BUG=32947)
-    // DCHECK_LE(pending_reads_, read_queue_.size());
-    while (pending_reads_ < read_queue_.size()) {
+    DCHECK_LE(pending_reads_, read_queue_.size());
+    if (!fulfilled) {
+      DCHECK_LT(pending_reads_, read_queue_.size());
       demuxer_stream_->Read(NewCallback(this, &DecoderBase::OnReadComplete));
       ++pending_reads_;
     }
@@ -257,10 +256,12 @@ class DecoderBase : public Decoder {
 
   // Attempts to fulfill a single pending read by dequeuing a buffer and read
   // callback pair and executing the callback.
-  void FulfillPendingRead() {
+  //
+  // Return true if one read request is fulfilled.
+  bool FulfillPendingRead() {
     DCHECK_EQ(MessageLoop::current(), this->message_loop());
     if (read_queue_.empty() || result_queue_.empty()) {
-      return;
+      return false;
     }
 
     // Dequeue a frame and read callback pair.
@@ -271,6 +272,7 @@ class DecoderBase : public Decoder {
 
     // Execute the callback!
     read_callback->Run(output);
+    return true;
   }
 
   // Tracks the number of asynchronous reads issued to |demuxer_stream_|.
diff --git a/media/filters/decoder_base_unittest.cc b/media/filters/decoder_base_unittest.cc
new file mode 100644
index 0000000..26fed48
--- /dev/null
+++ b/media/filters/decoder_base_unittest.cc
@@ -0,0 +1,130 @@
+// Copyright (c) 2010 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 <vector>
+
+#include "media/base/mock_filters.h"
+#include "media/filters/decoder_base.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+using ::testing::NotNull;
+
+namespace media {
+
+class MockOutput : public StreamSample {
+  MOCK_CONST_METHOD0(IsEndOfStream, bool());
+};
+
+class MockDecoder : public MediaFilter {};
+
+class MockBuffer : public Buffer {
+ public:
+  MOCK_CONST_METHOD0(GetData, const uint8*());
+  MOCK_CONST_METHOD0(GetDataSize, size_t());
+};
+
+ACTION(Initialize) {
+  AutoTaskRunner done_runner(arg2);
+  *arg1 = true;
+}
+
+ACTION_P(SaveDecodeRequest, list) {
+  scoped_refptr<Buffer> buffer = arg0;
+  list->push_back(arg1);
+}
+
+ACTION(CompleteDemuxRequest) {
+  arg0->Run(new MockBuffer());
+  delete arg0;
+}
+
+class MockDecoderImpl : public DecoderBase<MockDecoder, MockOutput> {
+ public:
+  MockDecoderImpl() {
+    media_format_.SetAsString(MediaFormat::kMimeType, "mock");
+  }
+
+  // DecoderBase Implementations.
+  MOCK_METHOD3(DoInitialize,
+               void(DemuxerStream* demuxer_stream, bool* success,
+                    Task* done_cb));
+  MOCK_METHOD0(DoStop, void());
+  MOCK_METHOD2(DoSeek, void(base::TimeDelta time, Task* done_cb));
+  MOCK_METHOD2(DoDecode, void(Buffer* input, Task* done_cb));
+
+ private:
+  FRIEND_TEST(DecoderBaseTest, FlowControl);
+
+  DISALLOW_COPY_AND_ASSIGN(MockDecoderImpl);
+};
+
+class MockReadCallback {
+ public:
+  MockReadCallback() {}
+  MOCK_METHOD1(ReadCallback, void(MockOutput* output));
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(MockReadCallback);
+};
+
+// Test the flow control of decoder base by the following sequnce of actions:
+// - Read() -> DecoderStream
+//   \ Read() -> DemuxerStream
+// - Read() -> DecoderBase
+//   \ Read() -> DemixerStream
+// - ReadCallback() -> DecoderBase
+//   \ DoDecode() -> Decoder
+// - ReadCallback() -> DecoderBase
+//   \ DoDecode() -> Decoder
+// - DecodeCallback() -> DecoderBase
+//   \ ReadCallback() -> client
+// - DecodeCallback() -> DecoderBase
+//   \ ReadCallback() -> client
+TEST(DecoderBaseTest, FlowControl) {
+  MessageLoop message_loop;
+  scoped_refptr<MockDecoderImpl> decoder = new MockDecoderImpl();
+  scoped_refptr<MockDemuxerStream> demuxer_stream = new MockDemuxerStream();
+  MockFilterCallback callback;
+  MockReadCallback read_callback;
+  decoder->set_message_loop(&message_loop);
+
+  // Initailize.
+  EXPECT_CALL(callback, OnFilterCallback());
+  EXPECT_CALL(callback, OnCallbackDestroyed());
+  EXPECT_CALL(*decoder, DoInitialize(NotNull(), NotNull(), NotNull()))
+      .WillOnce(Initialize());
+  decoder->Initialize(demuxer_stream.get(), callback.NewCallback());
+  message_loop.RunAllPending();
+
+  // Read.
+  std::vector<Task*> decode_requests;
+  EXPECT_CALL(*demuxer_stream, Read(NotNull()))
+      .Times(2)
+      .WillRepeatedly(CompleteDemuxRequest());
+  EXPECT_CALL(*decoder, DoDecode(NotNull(), NotNull()))
+      .Times(2)
+      .WillRepeatedly(SaveDecodeRequest(&decode_requests));
+  decoder->Read(NewCallback(&read_callback,
+                            &MockReadCallback::ReadCallback));
+  decoder->Read(NewCallback(&read_callback,
+                            &MockReadCallback::ReadCallback));
+  message_loop.RunAllPending();
+
+  // Fulfill the decode request.
+  EXPECT_CALL(read_callback, ReadCallback(NotNull())).Times(2);
+  for (size_t i = 0; i < decode_requests.size(); ++i) {
+    decoder->EnqueueResult(new MockOutput());
+    AutoTaskRunner done_cb(decode_requests[i]);
+  }
+  decode_requests.clear();
+  message_loop.RunAllPending();
+
+  // Stop.
+  EXPECT_CALL(*decoder, DoStop());
+  decoder->Stop();
+  message_loop.RunAllPending();
+}
+
+}  // namespace media
diff --git a/media/media.gyp b/media/media.gyp
index e8b8e07..9a2e264 100644
--- a/media/media.gyp
+++ b/media/media.gyp
@@ -190,6 +190,7 @@
         'filters/audio_renderer_algorithm_ola_unittest.cc',
         'filters/audio_renderer_base_unittest.cc',
         'filters/bitstream_converter_unittest.cc',
+        'filters/decoder_base_unittest.cc',
         'filters/ffmpeg_demuxer_unittest.cc',
         'filters/ffmpeg_glue_unittest.cc',
         'filters/ffmpeg_video_decode_engine_unittest.cc',