Buffered data source that does range request to provide data to media pipeline

The implementation of buffered data source that uses buffered resource loader
to download a media object and does buffering. This implementation knows
when to do defer loading and when to restarts the loading. The implementation
uses one buffered resource loading for one data source and does no connection
recovery.

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

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

5 files changed, 874 insertions, 1 deletions

diff --git a/chrome/chrome.gyp b/chrome/chrome.gyp
index 88ba594..aed7068 100644
--- a/chrome/chrome.gyp
+++ b/chrome/chrome.gyp
@@ -1556,6 +1556,8 @@
         'renderer/loadtimes_extension_bindings.cc',
         'renderer/media/audio_renderer_impl.cc',
         'renderer/media/audio_renderer_impl.h',
+        'renderer/media/buffered_data_source.cc',
+        'renderer/media/buffered_data_source.h',
         'renderer/media/data_source_impl.cc',
         'renderer/media/data_source_impl.h',
         'renderer/media/video_renderer_impl.cc',
diff --git a/chrome/renderer/media/buffered_data_source.cc b/chrome/renderer/media/buffered_data_source.cc
new file mode 100644
index 0000000..939c76d
--- /dev/null
+++ b/chrome/renderer/media/buffered_data_source.cc
@@ -0,0 +1,635 @@
+// Copyright (c) 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 "base/compiler_specific.h"
+#include "base/message_loop.h"
+#include "base/process_util.h"
+#include "base/string_util.h"
+#include "chrome/common/extensions/url_pattern.h"
+#include "chrome/renderer/media/buffered_data_source.h"
+#include "chrome/renderer/render_view.h"
+#include "chrome/renderer/webmediaplayer_delegate_impl.h"
+#include "chrome/renderer/render_thread.h"
+#include "media/base/filter_host.h"
+#include "net/base/load_flags.h"
+#include "net/base/net_errors.h"
+#include "net/http/http_response_headers.h"
+#include "webkit/glue/webappcachecontext.h"
+
+namespace {
+
+const char kHttpScheme[] = "http";
+const char kHttpsScheme[] = "https";
+const int64 kPositionNotSpecified = -1;
+const int kHttpOK = 200;
+const int kHttpPartialContent = 206;
+
+// A helper method that accepts only HTTP, HTTPS and FILE protocol.
+bool IsSchemeSupported(const GURL& url) {
+  return url.SchemeIs(kHttpScheme) ||
+         url.SchemeIs(kHttpsScheme) ||
+         url.SchemeIsFile();
+}
+
+}  // namespace
+
+/////////////////////////////////////////////////////////////////////////////
+// BufferedResourceLoader
+BufferedResourceLoader::BufferedResourceLoader(int route_id,
+                                               const GURL& url,
+                                               int64 first_byte_position,
+                                               int64 last_byte_position)
+    : start_callback_(NULL),
+      bridge_(NULL),
+      offset_(0),
+      content_length_(kPositionNotSpecified),
+      buffered_bytes_(0),
+      buffer_limit_(10240000),  // By default 10MB.
+      buffer_event_(false, false),
+      deferred_(false),
+      stopped_(false),
+      completed_(false),
+      range_requested_(false),
+      async_start_(false),
+      route_id_(route_id),
+      url_(url),
+      first_byte_position_(first_byte_position),
+      last_byte_position_(last_byte_position),
+      render_loop_(RenderThread::current()->message_loop()) {
+}
+
+BufferedResourceLoader::~BufferedResourceLoader() {
+  for (size_t i = 0; i < buffers_.size(); ++i)
+    delete buffers_[i];
+  buffers_.clear();
+}
+
+int BufferedResourceLoader::Start(net::CompletionCallback* start_callback) {
+  // Make sure we only start no more than once.
+  DCHECK(!bridge_.get());
+  DCHECK(!start_callback_.get());
+  start_callback_.reset(start_callback);
+
+  // Save the information that we are doing an asynchronous start since
+  // start_callback_ may get reset, we can't rely on it.
+  if (start_callback_.get())
+    async_start_ = true;
+
+  std::string header;
+  if (first_byte_position_ != kPositionNotSpecified &&
+      last_byte_position_ != kPositionNotSpecified) {
+    header = StringPrintf("Range: bytes=%lld-%lld",
+                          first_byte_position_,
+                          last_byte_position_);
+    range_requested_ = true;
+    offset_ = first_byte_position_;
+  } else if (first_byte_position_ != kPositionNotSpecified) {
+    header = StringPrintf("Range: bytes=%lld-", first_byte_position_);
+    range_requested_ = true;
+    offset_ = first_byte_position_;
+  } else if (last_byte_position_ != kPositionNotSpecified) {
+    NOTIMPLEMENTED() << "Suffix length range request not implemented.";
+  }
+
+  bridge_.reset(RenderThread::current()->resource_dispatcher()->CreateBridge(
+      "GET",
+      GURL(url_),
+      GURL(url_),
+      GURL(),         // TODO(hclam): provide referer here.
+      "null",         // TODO(abarth): provide frame_origin
+      "null",         // TODO(abarth): provide main_frame_origin
+      header,
+      net::LOAD_BYPASS_CACHE,
+      base::GetCurrentProcId(),
+      ResourceType::MEDIA,
+      0,
+      // TODO(michaeln): delegate->mediaplayer->frame->
+      //                    app_cache_context()->context_id()
+      // For now don't service media resource requests from the appcache.
+      WebAppCacheContext::kNoAppCacheContextId,
+      route_id_));
+
+  // We may receive stop signal while we are inside this method, it's because
+  // Start() may get called on demuxer thread while Stop() is called on
+  // pipeline thread, so we want to protect the posting of OnStart() task
+  // with a lock.
+  bool task_posted = false;
+  {
+    AutoLock auto_lock(lock_);
+    if (!stopped_) {
+      task_posted =  true;
+      render_loop_->PostTask(FROM_HERE,
+          NewRunnableMethod(this, &BufferedResourceLoader::OnStart));
+    }
+  }
+
+  // Wait for response to arrive we don't have an async start.
+  if (task_posted && !async_start_)
+    buffer_event_.Wait();
+
+  {
+    AutoLock auto_lock(lock_);
+    // We may have stopped because of a bad response from the server.
+    if (stopped_)
+      return net::ERR_ABORTED;
+    else if (completed_)
+      return net::ERR_FAILED;
+    else if (async_start_)
+      return net::ERR_IO_PENDING;
+    return net::OK;
+  }
+}
+
+void BufferedResourceLoader::Stop() {
+  {
+    AutoLock auto_lock(lock_);
+    stopped_ = true;
+  }
+
+  // Wakes up the waiting thread so they can catch the stop signal.
+  render_loop_->PostTask(FROM_HERE,
+      NewRunnableMethod(this, &BufferedResourceLoader::OnDestroy));
+  buffer_event_.Signal();
+}
+
+size_t BufferedResourceLoader::Read(uint8* data, size_t size) {
+  size_t taken = 0;
+  while (taken < size) {
+    Buffer* buffer = NULL;
+    {
+      AutoLock auto_lock(lock_);
+      if (stopped_)
+        break;
+      if (!buffers_.empty())
+        buffer = buffers_.front();
+      else if (completed_)
+        break;
+    }
+    if (buffer) {
+      size_t copy = std::min(size - taken, buffer->size - buffer->taken);
+      memcpy(data + taken, buffer->data.get() + buffer->taken, copy);
+      taken += copy;
+      buffer->taken += copy;
+      if (buffer->taken == buffer->size) {
+        // The buffer has been consumed, remove it.
+        {
+          AutoLock auto_lock(lock_);
+          buffers_.pop_front();
+        }
+        delete buffer;
+      }
+    } else {
+      buffer_event_.Wait();
+    }
+  }
+  if (taken > 0) {
+    offset_ += taken;
+    {
+      AutoLock auto_lock(lock_);
+      buffered_bytes_ -= taken;
+      DCHECK(buffered_bytes_ >= 0);
+    }
+    if (ShouldDisableDefer()) {
+      AutoLock auto_lock(lock_);
+      if (!stopped_) {
+        render_loop_->PostTask(FROM_HERE,
+            NewRunnableMethod(this,
+                              &BufferedResourceLoader::OnDisableDeferLoading));
+      }
+    }
+  }
+  return taken;
+}
+
+bool BufferedResourceLoader::SeekForward(int64 position) {
+  // Use of |offset_| is safe without a lock, because it's modified only in
+  // Read() and this method after Start(). Read() and SeekForward() happens
+  // on the same thread.
+  // Seeking backward.
+  if (position < offset_)
+    return false;
+  // Done seeking forward.
+  else if (position == offset_)
+    return true;
+
+  while(true) {
+    {
+      AutoLock auto_lock(lock_);
+      // Loader has stopped.
+      if (stopped_)
+        return false;
+      // Seek position exceeds bufferable range, buffer_limit_ can be changed.
+      if (position >= offset_ + buffer_limit_)
+        return false;
+      // Response completed and seek position exceeds buffered range.
+      if (completed_ && position >= offset_ + buffered_bytes_)
+        return false;
+
+      if (!buffers_.empty()) {
+        Buffer* buffer = buffers_.front();
+        int64 bytes_to_take = position - offset_;
+        if (!buffers_.empty()) {
+          size_t available_bytes_in_buffer = buffer->size - buffer->taken;
+          size_t taken = 0;
+          if (available_bytes_in_buffer <= bytes_to_take) {
+            taken = available_bytes_in_buffer;
+            buffers_.pop_front();
+            delete buffer;
+          } else {
+            taken = static_cast<size_t>(bytes_to_take);
+            buffer->taken += taken;
+          }
+          offset_ += taken;
+          if (bytes_to_take == taken)
+            return true;
+        }
+        continue;
+      }
+    }
+    buffer_event_.Wait();
+  }
+}
+
+int64 BufferedResourceLoader::GetOffset() {
+  AutoLock auto_lock(lock_);
+  return offset_;
+}
+
+int64 BufferedResourceLoader::GetBufferLimit() {
+  AutoLock auto_lock(lock_);
+  return buffer_limit_;
+}
+
+void BufferedResourceLoader::SetBufferLimit(size_t buffer_limit) {
+  {
+    AutoLock auto_lock(lock_);
+    buffer_limit_ = buffer_limit;
+  }
+
+  if (ShouldDisableDefer()) {
+    AutoLock auto_lock(lock_);
+    if (!stopped_) {
+      render_loop_->PostTask(FROM_HERE,
+          NewRunnableMethod(this,
+                            &BufferedResourceLoader::OnDisableDeferLoading));
+    }
+  }
+  if (ShouldEnableDefer()) {
+    AutoLock auto_lock(lock_);
+    if (!stopped_) {
+      render_loop_->PostTask(FROM_HERE,
+          NewRunnableMethod(this,
+                            &BufferedResourceLoader::OnEnableDeferLoading));
+    }
+  }
+}
+
+size_t BufferedResourceLoader::GetTimeout() {
+  // TODO(hclam): implement.
+  return 0;
+}
+
+void BufferedResourceLoader::SetTimeout(size_t milliseconds) {
+  // TODO(hclam): implement.
+}
+
+/////////////////////////////////////////////////////////////////////////////
+// BufferedResourceLoader,
+//     webkit_glue::ResourceLoaderBridge::Peer implementations
+void BufferedResourceLoader::OnReceivedRedirect(const GURL& new_url) {
+  url_ = new_url;
+
+  // If we got redirected to an unsupported protocol then stop.
+  if (!IsSchemeSupported(new_url))
+    Stop();
+}
+
+void BufferedResourceLoader::OnReceivedResponse(
+    const webkit_glue::ResourceLoaderBridge::ResponseInfo& info,
+    bool content_filtered) {
+  int64 first_byte_position = -1;
+  int64 last_byte_position = -1;
+  int64 instance_size = -1;
+
+  // The file:// protocol should be able to serve any request we want, so we
+  // take an exception for file protocol.
+  if (!url_.SchemeIsFile()) {
+    if (!info.headers) {
+      // We expect to receive headers because this is a HTTP or HTTPS protocol,
+      // if not report failure.
+      InvokeAndResetStartCallback(net::ERR_INVALID_RESPONSE);
+      Stop();
+      return;
+    } else if (range_requested_) {
+      if (info.headers->response_code() != kHttpPartialContent ||
+          !info.headers->GetContentRange(&first_byte_position,
+                                         &last_byte_position,
+                                         &instance_size)) {
+        // We requested a range, but server didn't reply with partial content or
+        // the "Content-Range" header is corrupted.
+        InvokeAndResetStartCallback(net::ERR_INVALID_RESPONSE);
+        Stop();
+        return;
+      }
+    } else if (info.headers->response_code() != kHttpOK) {
+      // We didn't request a range but server didn't reply with "200 OK".
+      InvokeAndResetStartCallback(net::ERR_FAILED);
+      Stop();
+      return;
+    }
+  }
+
+  {
+    AutoLock auto_lock(lock_);
+    // |info.content_length| can be -1, in that case |content_length_| is
+    // not specified and this is a streaming response.
+    content_length_ = info.content_length;
+    // We only care about the first byte position if it's given by the server.
+    if (first_byte_position != kPositionNotSpecified)
+      offset_ = first_byte_position;
+  }
+
+  // If we have started asynchronously we just need to invoke the callback or
+  // we need to signal the Start() method to wake up.
+  if (async_start_)
+    InvokeAndResetStartCallback(net::OK);
+  else
+    buffer_event_.Signal();
+}
+
+void BufferedResourceLoader::OnReceivedData(const char* data, int len) {
+  DCHECK(bridge_.get());
+
+  AppendToBuffer(reinterpret_cast<const uint8*>(data), len);
+  if (ShouldEnableDefer())
+    bridge_->SetDefersLoading(true);
+}
+
+void BufferedResourceLoader::OnCompletedRequest(
+    const URLRequestStatus& status, const std::string& security_info) {
+  SignalComplete();
+  // After the response has completed, we don't need the bridge any more.
+  bridge_.reset();
+
+  if (async_start_)
+    InvokeAndResetStartCallback(status.os_error());
+}
+
+/////////////////////////////////////////////////////////////////////////////
+// BufferedResourceLoader, private
+void BufferedResourceLoader::AppendToBuffer(const uint8* data, size_t size) {
+  Buffer* buffer = new Buffer(size);
+  memcpy(buffer->data.get(), data, size);
+  {
+    AutoLock auto_lock(lock_);
+    buffers_.push_back(buffer);
+    buffered_bytes_ += size;
+  }
+  buffer_event_.Signal();
+}
+
+void BufferedResourceLoader::SignalComplete() {
+  {
+    AutoLock auto_lock(lock_);
+    completed_ = true;
+  }
+  buffer_event_.Signal();
+}
+
+bool BufferedResourceLoader::ShouldEnableDefer() {
+  AutoLock auto_lock(lock_);
+  if (deferred_) {
+    return false;
+  } else if (buffered_bytes_ >= buffer_limit_) {
+    deferred_ = true;
+    return true;
+  }
+  return false;
+}
+
+bool BufferedResourceLoader::ShouldDisableDefer() {
+  AutoLock auto_lock(lock_);
+  if (deferred_ && buffered_bytes_ < buffer_limit_ / 2)
+    return true;
+  else
+    return false;
+}
+
+void BufferedResourceLoader::OnStart() {
+  DCHECK(MessageLoop::current() == render_loop_);
+  DCHECK(bridge_.get());
+  bridge_->Start(this);
+}
+
+void BufferedResourceLoader::OnDestroy() {
+  DCHECK(MessageLoop::current() == render_loop_);
+  bridge_.reset();
+}
+
+void BufferedResourceLoader::OnEnableDeferLoading() {
+  DCHECK(MessageLoop::current() == render_loop_);
+  // This message may arrive after the bridge is destroyed.
+  if (bridge_.get())
+    bridge_->SetDefersLoading(true);
+}
+
+void BufferedResourceLoader::OnDisableDeferLoading() {
+  DCHECK(MessageLoop::current() == render_loop_);
+  // This message may arrive after the bridge is destroyed.
+  if (bridge_.get())
+    bridge_->SetDefersLoading(false);
+}
+
+void BufferedResourceLoader::InvokeAndResetStartCallback(int error) {
+  AutoLock auto_lock(lock_);
+  if (start_callback_.get()) {
+    start_callback_->Run(error);
+    start_callback_.reset();
+  }
+}
+
+//////////////////////////////////////////////////////////////////////////////
+// BufferedDataSource
+BufferedDataSource::BufferedDataSource(WebMediaPlayerDelegateImpl* delegate)
+    : delegate_(delegate),
+      stopped_(false),
+      position_(0),
+      total_bytes_(kPositionNotSpecified),
+      buffered_resource_loader_(NULL),
+      pipeline_loop_(MessageLoop::current()) {
+}
+
+BufferedDataSource::~BufferedDataSource() {
+}
+
+void BufferedDataSource::Stop() {
+  scoped_refptr<BufferedResourceLoader> resource_loader = NULL;
+  // Set the stop signal first.
+  {
+    AutoLock auto_lock(lock_);
+    stopped_ = true;
+    resource_loader = buffered_resource_loader_;
+    // Release the reference to the resource loader.
+    buffered_resource_loader_ = NULL;
+  }
+  // Tell the loader to stop.
+  if (resource_loader)
+    resource_loader->Stop();
+}
+
+bool BufferedDataSource::Initialize(const std::string& url) {
+  // Save the url.
+  url_ = GURL(url);
+
+  // Make sure we support the scheme of the URL.
+  if (!IsSchemeSupported(url_)) {
+    host_->Error(media::PIPELINE_ERROR_NETWORK);
+    return false;
+  }
+
+  media_format_.SetAsString(media::MediaFormat::kMimeType,
+                            media::mime_type::kApplicationOctetStream);
+  media_format_.SetAsString(media::MediaFormat::kURL, url);
+
+  // Setup the BufferedResourceLoader here.
+  scoped_refptr<BufferedResourceLoader> resource_loader = NULL;
+  {
+    AutoLock auto_lock(lock_);
+    if (!stopped_) {
+      buffered_resource_loader_ = new BufferedResourceLoader(
+          delegate_->view()->routing_id(),
+          url_, kPositionNotSpecified, kPositionNotSpecified);
+      resource_loader = buffered_resource_loader_;
+    }
+  }
+
+  // Use the local reference to start the request.
+  if (resource_loader) {
+    if (net::ERR_IO_PENDING != resource_loader->Start(
+            NewCallback(this, &BufferedDataSource::InitialRequestStarted))) {
+      host_->Error(media::PIPELINE_ERROR_NETWORK);
+      return false;
+    }
+    return true;
+  }
+  host_->Error(media::PIPELINE_ERROR_NETWORK);
+  return false;
+}
+
+size_t BufferedDataSource::Read(uint8* data, size_t size) {
+  // We try two times here:
+  // 1. Use the existing resource loader to seek forward and read from it.
+  // 2. If any of the above operations failed, we create a new resource loader
+  //    starting with a new range. Goto 1.
+  // TODO(hclam): change the logic here to do connection recovery and allow a
+  // maximum of trials.
+  for (int trials = 2; trials > 0; --trials) {
+    scoped_refptr<BufferedResourceLoader> resource_loader = NULL;
+    {
+      AutoLock auto_lock(lock_);
+      resource_loader = buffered_resource_loader_;
+    }
+
+    if (resource_loader && resource_loader->SeekForward(position_)) {
+      size_t read = resource_loader->Read(data, size);
+      if (read >= 0) {
+        position_ += read;
+        return read;
+      } else {
+        return DataSource::kReadError;
+      }
+    } else {
+      // We enter here because the current resource loader cannot serve the
+      // range requested, we will need to create a new request for doing it.
+      scoped_refptr<BufferedResourceLoader> old_resource_loader = NULL;
+      {
+        AutoLock auto_lock(lock_);
+        if (stopped_)
+          return DataSource::kReadError;
+        old_resource_loader = buffered_resource_loader_;
+        buffered_resource_loader_ =
+            new BufferedResourceLoader(delegate_->view()->routing_id(),
+                                       url_, position_,
+                                       kPositionNotSpecified);
+        resource_loader = buffered_resource_loader_;
+      }
+      if (old_resource_loader)
+        old_resource_loader->Stop();
+      if (resource_loader) {
+        if (net::OK != resource_loader->Start(NULL)) {
+          // We have started a new request but failed, report that.
+          // TODO(hclam): should allow some retry mechanism here.
+          HandleError(media::PIPELINE_ERROR_NETWORK);
+          return DataSource::kReadError;
+        }
+      }
+    }
+  }
+  return DataSource::kReadError;
+}
+
+bool BufferedDataSource::GetPosition(int64* position_out) {
+  *position_out = position_;
+  return true;
+}
+
+bool BufferedDataSource::SetPosition(int64 position) {
+  position_ = position;
+  return true;
+}
+
+bool BufferedDataSource::GetSize(int64* size_out) {
+  if (total_bytes_ != kPositionNotSpecified) {
+    *size_out = total_bytes_;
+    return true;
+  }
+  *size_out = 0;
+  return false;
+}
+
+bool BufferedDataSource::IsSeekable() {
+  return total_bytes_ != kPositionNotSpecified;
+}
+
+void BufferedDataSource::HandleError(media::PipelineError error) {
+  AutoLock auto_lock(lock_);
+  if (!stopped_) {
+    host_->Error(error);
+  }
+}
+
+void BufferedDataSource::InitialRequestStarted(int error) {
+  // Don't take any lock and call to |host_| here, this method is called from
+  // BufferedResourceLoader after the response has started or failed, it is
+  // very likely we are called within a lock in BufferedResourceLoader.
+  // Acquiring an additional lock here we might have a deadlock situation,
+  // but one thing very sure is that pipeline thread is still alive, so we
+  // just need to post a task on that thread.
+  pipeline_loop_->PostTask(FROM_HERE,
+      NewRunnableMethod(this,
+                        &BufferedDataSource::OnInitialRequestStarted, error));
+}
+
+void BufferedDataSource::OnInitialRequestStarted(int error) {
+  // Acquiring a lock should not be needed because stopped_ is only written
+  // on pipeline thread and we are on pipeline thread but just to be safe.
+  AutoLock auto_lock(lock_);
+  if (!stopped_) {
+    if (error == net::OK) {
+      total_bytes_ = buffered_resource_loader_->content_length();
+      if (IsSeekable()) {
+        host_->SetTotalBytes(total_bytes_);
+        // TODO(hclam): report the amount of bytes buffered accurately.
+        host_->SetBufferedBytes(total_bytes_);
+      }
+      host_->InitializationComplete();
+    } else {
+      host_->Error(media::PIPELINE_ERROR_NETWORK);
+    }
+  }
+}
+
+const media::MediaFormat& BufferedDataSource::media_format() {
+  return media_format_;
+}
diff --git a/chrome/renderer/media/buffered_data_source.h b/chrome/renderer/media/buffered_data_source.h
new file mode 100644
index 0000000..bd80402
--- /dev/null
+++ b/chrome/renderer/media/buffered_data_source.h
@@ -0,0 +1,227 @@
+// Copyright (c) 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.
+
+#ifndef CHROME_RENDERER_MEDIA_BUFFERED_DATA_SOURCE_H_
+#define CHROME_RENDERER_MEDIA_BUFFERED_DATA_SOURCE_H_
+
+#include <deque>
+#include <string>
+
+#include "base/lock.h"
+#include "base/scoped_ptr.h"
+#include "base/waitable_event.h"
+#include "media/base/factory.h"
+#include "media/base/filters.h"
+#include "media/base/media_format.h"
+#include "media/base/pipeline.h"
+#include "net/base/completion_callback.h"
+#include "net/base/file_stream.h"
+#include "webkit/glue/resource_loader_bridge.h"
+#include "googleurl/src/gurl.h"
+
+class WebMediaPlayerDelegateImpl;
+
+/////////////////////////////////////////////////////////////////////////////
+// BufferedResourceLoader
+// This class works inside demuxer thread and render thread. It contains a
+// resource loader bridge and does the actual resource loading. This object
+// does buffering internally, it defers the resource loading if buffer is
+// full and un-defers the resource loading if it is under buffered.
+class BufferedResourceLoader :
+    public base::RefCountedThreadSafe<BufferedResourceLoader>,
+    public webkit_glue::ResourceLoaderBridge::Peer {
+ public:
+  BufferedResourceLoader(int route_id,
+                         const GURL& url,
+                         int64 first_byte_position,
+                         int64 last_byte_position);
+  virtual ~BufferedResourceLoader();
+
+  // Start the resource loading with the specified URL and range.
+  // This method call can operate in two modes, synchronous and asynchronous.
+  // If |start_callback| is NULL, this method operates in synchronous mode and
+  // it returns true if the load has started successfully, false otherwise. It
+  // returns only if a resource is received from the server or this loader is
+  // called to stop.
+  // If |start_callback| is not NULL, this method operates in asynchronous mode
+  // and it returns net::ERR_IO_PENDING if the request is going to start.
+  // Once there's a response from the server, success or fail |start_callback|
+  // is called with the result.
+  // Note that |start_callback| is called within a lock to prevent invoking an
+  // invalid callback method while this object is called to stop.
+  int Start(net::CompletionCallback* callback);
+
+  // Stops this loader. Wakes up all synchronous actions.
+  void Stop();
+
+  // Reads the specified |size| into |buffer| and returns number of bytes copied
+  // into the buffer. Returns 0 if the response has completed and there's no
+  // no buffer left. Returns media::kReadError on error. The read starts from
+  // the current position referred by calling GetOffset(). This method call is
+  // synchronous, it returns only the required amount of bytes is read, the
+  // loader is stopped, this resource loading has completed or the read has
+  // timed out. Read() and SeekForward() cannot be called concurrently.
+  size_t Read(uint8* buffer, size_t size);
+
+  // Seek forward to |position| in bytes in the entire instance of the media
+  // object, returns true if successful. If the seek operation cannot be
+  // performed because it's seeking backward, the loader has been stopped,
+  // the seek |position| exceed bufferable range or the seek operation has
+  // timed out, returns false.
+  // There cannot be SeekForward() while another thread is calling Read().
+  bool SeekForward(int64 position);
+
+  // Returns the position in bytes that this loader is downloading from.
+  int64 GetOffset();
+
+  // Gets and sets the buffering limit of this loader.
+  int64 GetBufferLimit();
+  void SetBufferLimit(size_t buffe_limit);
+
+  // Gets and sets the timeout for the synchronous operations.
+  size_t GetTimeout();
+  void SetTimeout(size_t milliseconds);
+
+  // Gets the content length in bytes of the instance after this loader has been
+  // started.
+  int64 content_length() { return content_length_; }
+
+  /////////////////////////////////////////////////////////////////////////////
+  // webkit_glue::ResourceLoaderBridge::Peer implementations.
+  virtual void OnUploadProgress(uint64 position, uint64 size) {}
+  virtual void OnReceivedRedirect(const GURL& new_url);
+  virtual void OnReceivedResponse(
+      const webkit_glue::ResourceLoaderBridge::ResponseInfo& info,
+      bool content_filtered);
+  virtual void OnReceivedData(const char* data, int len);
+  virtual void OnCompletedRequest(const URLRequestStatus& status,
+      const std::string& security_info);
+  std::string GetURLForDebugging() { return url_.spec(); }
+
+ private:
+  // Append buffer to the queue of buffers.
+  void AppendToBuffer(const uint8* buffer, size_t size);
+  void SignalComplete();
+  bool ShouldEnableDefer();
+  bool ShouldDisableDefer();
+
+  void OnStart();
+  void OnDestroy();
+  void OnDisableDeferLoading();
+  void OnEnableDeferLoading();
+
+  void InvokeAndResetStartCallback(int error);
+
+  struct Buffer {
+    Buffer(size_t len) : taken(0), size(len), data(new uint8[len]) { }
+
+    // The amount of buffer in bytes consumed in this buffer starting from
+    // index 0.
+    size_t taken;
+    size_t size;
+    scoped_array<uint8> data;
+  };
+
+  scoped_ptr<net::CompletionCallback> start_callback_;
+  scoped_ptr<webkit_glue::ResourceLoaderBridge> bridge_;
+  int64 offset_;
+  int64 content_length_;
+
+  std::deque<Buffer*> buffers_;
+  size_t buffered_bytes_;
+  size_t buffer_limit_;
+  base::WaitableEvent buffer_event_;
+
+  bool deferred_;
+  bool stopped_;
+  bool completed_;
+  bool range_requested_;
+  bool async_start_;
+
+  int route_id_;
+  GURL url_;
+  int64 first_byte_position_;
+  int64 last_byte_position_;
+
+  MessageLoop* render_loop_;
+  // A lock that protects usage of the following members:
+  // - buffers_
+  // - buffered_bytes_
+  // - buffered_limit_
+  // - deferred_
+  // - stopped_
+  // - completed_
+  Lock lock_;
+
+  DISALLOW_COPY_AND_ASSIGN(BufferedResourceLoader);
+};
+
+class BufferedDataSource : public media::DataSource {
+ public:
+  // Methods called from pipeline thread
+  // Static methods for creating this class.
+  static media::FilterFactory* CreateFactory(
+      WebMediaPlayerDelegateImpl* delegate) {
+    return new media::FilterFactoryImpl1<BufferedDataSource,
+        WebMediaPlayerDelegateImpl*>(delegate);
+  }
+  virtual bool Initialize(const std::string& url);
+
+  // media::MediaFilter implementation.
+  virtual void Stop();
+
+  // media::DataSource implementation.
+  // Called from demuxer thread.
+  virtual size_t Read(uint8* data, size_t size);
+  virtual bool GetPosition(int64* position_out);
+  virtual bool SetPosition(int64 position);
+  virtual bool GetSize(int64* size_out);
+  virtual bool IsSeekable();
+
+  const media::MediaFormat& media_format();
+
+ private:
+  friend class media::FilterFactoryImpl1<BufferedDataSource,
+                                         WebMediaPlayerDelegateImpl*>;
+  // Call to filter host to trigger an error, be sure not to call this method
+  // while the lock is acquired.
+  void HandleError(media::PipelineError error);
+
+  // Callback method from BufferedResourceLoader for the initial url request.
+  // |error| is net::OK if the request has started successfully or |error| is
+  // a code representing the actual network error.
+  void InitialRequestStarted(int error);
+  void OnInitialRequestStarted(int error);
+
+  explicit BufferedDataSource(WebMediaPlayerDelegateImpl* delegate);
+  virtual ~BufferedDataSource();
+
+  media::MediaFormat media_format_;
+  GURL url_;
+
+  // Pointer to the delegate which provides access to RenderView, this is set
+  // in construction and can be accessed in all threads safely.
+  // TODO(hclam): get rid of this and save the routing id and pointer to
+  // ResourceDispatcher.
+  WebMediaPlayerDelegateImpl* delegate_;
+
+  // A common lock for protecting members accessed by multiple threads.
+  Lock lock_;
+  bool stopped_;
+
+  // Members used for reading.
+  int64 position_;
+  // Members for total bytes of the requested object.
+  int64 total_bytes_;
+
+  // Members related to resource loading with RenderView.
+  scoped_refptr<BufferedResourceLoader> buffered_resource_loader_;
+
+  // The message loop of the pipeline thread.
+  MessageLoop* pipeline_loop_;
+
+  DISALLOW_COPY_AND_ASSIGN(BufferedDataSource);
+};
+
+#endif  // CHROME_RENDERER_MEDIA_BUFFERED_DATA_SOURCE_H_
diff --git a/chrome/renderer/renderer.vcproj b/chrome/renderer/renderer.vcproj
index 5a388fd..da81a40 100644
--- a/chrome/renderer/renderer.vcproj
+++ b/chrome/renderer/renderer.vcproj
@@ -145,6 +145,14 @@
 				>
 			</File>
 			<File
+				RelativePath=".\media\buffered_data_source.cc"
+				>
+			</File>
+			<File
+				RelativePath=".\media\buffered_data_source.h"
+				>
+			</File>
+			<File
 				RelativePath=".\media\data_source_impl.cc"
 				>
 			</File>
diff --git a/chrome/renderer/webmediaplayer_delegate_impl.cc b/chrome/renderer/webmediaplayer_delegate_impl.cc
index 4e811e8..bfece7c 100644
--- a/chrome/renderer/webmediaplayer_delegate_impl.cc
+++ b/chrome/renderer/webmediaplayer_delegate_impl.cc
@@ -8,6 +8,7 @@
 #include "chrome/common/chrome_switches.h"
 #include "chrome/renderer/media/audio_renderer_impl.h"
 #include "chrome/renderer/media/data_source_impl.h"
+#include "chrome/renderer/media/buffered_data_source.h"
 #include "chrome/renderer/media/video_renderer_impl.h"
 #include "chrome/renderer/render_view.h"
 #include "googleurl/src/gurl.h"
@@ -79,7 +80,7 @@ WebMediaPlayerDelegateImpl::WebMediaPlayerDelegateImpl(RenderView* view)
         AudioRendererImpl::CreateFactory(view_->audio_message_filter()));
   }
   filter_factory_->AddFactory(VideoRendererImpl::CreateFactory(this));
-  filter_factory_->AddFactory(DataSourceImpl::CreateFactory(this));
+  filter_factory_->AddFactory(BufferedDataSource::CreateFactory(this));
 }
 
 WebMediaPlayerDelegateImpl::~WebMediaPlayerDelegateImpl() {