Initial implementation of stream switching in AdaptiveDemuxer.

media/filters/adaptive_demuxer.cc

 // Copyright (c) 2011 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/bind.h"
 #include "base/logging.h"
+#include "base/message_loop.h"
 #include "base/string_number_conversions.h"
 #include "base/string_split.h"
 #include "base/string_util.h"
+#include "media/ffmpeg/ffmpeg_common.h"
 #include "media/filters/adaptive_demuxer.h"
 
 namespace media {
 
+static const int64 kSwitchTimerPeriod = 5000; // In milliseconds.
+
+// Object that decides when to switch streams.
+class StreamSwitchManager
+    : public base::RefCountedThreadSafe<StreamSwitchManager> {
+ public:
+  StreamSwitchManager();
+  virtual ~StreamSwitchManager();
+
+  void Init(AdaptiveDemuxer* demuxer);
+
+  // Playback events. These methods are called when playback starts, pauses, or
+  // stops.
+  void Play();
+  void Pause();
+  void Stop();
+
+ private:
+  // Method called periodically to determine if we should switch
+  // streams.
+  void OnSwitchTimer();
+
+  // Called when the demuxer completes a stream switch.
+  void OnSwitchDone(PipelineStatus status);
+
+  // Helper method that schedules OnSwitchTimer() to be called.
+  void StartSwitchTimer();
+
+  // The demuxer that owns this object.
+  AdaptiveDemuxer* demuxer_;
+
+  // Message loop this object runs on.
+  MessageLoop* message_loop_;
+
+  // Is clip playing or not?
+  bool playing_;
+
+  // Is a stream switch in progress?
+  bool switch_pending_;
+
+  // Has an OnSwitchTimer() call been scheduled?
+  bool switch_timer_running_;
+
+  // The stream IDs for the video streams.
+  AdaptiveDemuxer::StreamIdVector video_ids_;
+
+  // An index into |video_ids_| for the current video stream.
+  int current_id_index_;
+
+  DISALLOW_COPY_AND_ASSIGN(StreamSwitchManager);
+};
+
+StreamSwitchManager::StreamSwitchManager()
+    : demuxer_(NULL),
+      playing_(false),
+      switch_pending_(false),
+      switch_timer_running_(false),
+      current_id_index_(-1) {
+}
+
+StreamSwitchManager::~StreamSwitchManager() {}
+
+void StreamSwitchManager::Init(AdaptiveDemuxer* demuxer) {
+  DCHECK(demuxer);
+  message_loop_ = MessageLoop::current();
+  demuxer_ = demuxer;
+  video_ids_ = demuxer->GetVideoIds();
+  current_id_index_ = -1;
+
+  if (video_ids_.size() > 0) {
+    // Find the index in video_ids_ for the current video ID.
+    int current_id = demuxer->GetCurrentVideoId();
+    current_id_index_ = 0;
+    for (int i = 0; i < video_ids_.size(); i++) {
+      if (current_id == video_ids_[i]) {
+        current_id_index_ = i;
+        break;
+      }
+    }
+  }
+}
+
+void StreamSwitchManager::Play() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(!playing_);
+  playing_ = true;
+
+  if (video_ids_.size() > 1 && !switch_timer_running_) {
+    StartSwitchTimer();
+  }
+}
+
+void StreamSwitchManager::Pause() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(playing_);
+  playing_ = false;
+}
+
+void StreamSwitchManager::Stop() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  DCHECK(!playing_);
+  demuxer_ = NULL;
+}
+
+void StreamSwitchManager::OnSwitchTimer() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+
+  switch_timer_running_ = false;
+
+  if (!playing_)
+    return;
+
+  if (!switch_pending_) {
+    // Select the stream to switch to. For now we are just rotating
+    // through the available streams.
+    int new_id_index = (current_id_index_ + 1) % video_ids_.size();
+
+    // Did we select a new stream?
+    if (new_id_index != current_id_index_) {
+      current_id_index_ = new_id_index;
+      switch_pending_ = true;
+      demuxer_->ChangeVideoStream(video_ids_[new_id_index],
+                                  base::Bind(&StreamSwitchManager::OnSwitchDone,
+                                             this));
+    }
+  }
+
+  StartSwitchTimer();
+}
+
+void StreamSwitchManager::OnSwitchDone(PipelineStatus status) {
+  if (MessageLoop::current() != message_loop_) {

[scherkus (not reviewing), 2011/05/23 23:17:06]

nothing actionable right now but some food for thought: ajwong and I have been
talking about is how to remove boilerplate code such as this

Imagine a world where every method starts with:
DCHECK_EQ(MessageLoop::current(), message_loop_);

...and reposting is forbidden (i.e., callee is responsible for making sure the
method was called on the right thread)

it'll work as long as we don't expect our callbacks to by synchronous -- which I
think is alright since most of the callbacks today do repost!

[acolwell GONE FROM CHROMIUM, 2011/05/23 23:46:40]

I was going to comment on this when I made the change. It is annoying right now
that I have to know that another object has a different thread and can call me
on it. It would be nice if say FFmpeg demuxer would wrap callbacks I hand it so
that it would always call me back on the thread that I called Read() from. It
would change the semantics from "I could call you on a different thread" to
"I'll call you back on the thread you called me on." I think the later is WAY
better. It is also much easier to do with the new callback system.

+    message_loop_->PostTask(
+        FROM_HERE,
+        NewRunnableMethod(this,&StreamSwitchManager::OnSwitchDone, status));
+  }
+  switch_pending_ = false;
+}
+
+void StreamSwitchManager::StartSwitchTimer() {
+  DCHECK_EQ(MessageLoop::current(), message_loop_);
+  switch_timer_running_ = true;
+  message_loop_->PostDelayedTask(
+      FROM_HERE, NewRunnableMethod(this, &StreamSwitchManager::OnSwitchTimer),
+      kSwitchTimerPeriod);
+}
+
 //
 // AdaptiveDemuxerStream
 //
 AdaptiveDemuxerStream::AdaptiveDemuxerStream(
     StreamVector const& streams, int initial_stream)
     : streams_(streams), current_stream_index_(initial_stream),
-      bitstream_converter_enabled_(false) {
+      bitstream_converter_enabled_(false),
+      pending_reads_(0),
+      switch_index_(-1),
+      last_buffer_timestamp_(kNoTimestamp) {
   DCheckSanity();
 }
 
 void AdaptiveDemuxerStream::DCheckSanity() {
   // We only carry out sanity checks in debug mode.
   if (!logging::DEBUG_MODE)
     return;
   DCHECK(streams_[current_stream_index_].get());
 
   bool non_null_stream_seen = false;
@@ skipping 19 matching lines @@
   current_stream_index_ = -1;
   streams_.clear();
 }
 
 DemuxerStream* AdaptiveDemuxerStream::current_stream() {
   base::AutoLock auto_lock(lock_);
   return streams_[current_stream_index_];
 }
 
 void AdaptiveDemuxerStream::Read(const ReadCallback& read_callback) {
-  current_stream()->Read(read_callback);
+  DemuxerStream* stream = NULL;
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    read_cb_queue_.push_back(read_callback);
+
+    // Check to make sure we aren't doing a stream switch. We only want to
+    // make calls on |streams_[current_stream_index_]| when we aren't
+    // in the middle of a stream switch. Since the callback is stored in
+    // |read_cb_queue_| we will issue the Read() on the new stream once
+    // the switch has completed.
+    if (!IsSwitchPending_Locked()) {
+      stream = streams_[current_stream_index_];
+      pending_reads_++;
+    }
+  }
+
+  if (stream)
+    stream->Read(base::Bind(&AdaptiveDemuxerStream::OnReadDone, this));
 }
 
 AVStream* AdaptiveDemuxerStream::GetAVStream() {
   return current_stream()->GetAVStream();
 }
 
 DemuxerStream::Type AdaptiveDemuxerStream::type() {
   return current_stream()->type();
 }
 
 const MediaFormat& AdaptiveDemuxerStream::media_format() {
   return current_stream()->media_format();
 }
 
 void AdaptiveDemuxerStream::EnableBitstreamConverter() {
   {
     base::AutoLock auto_lock(lock_);
     bitstream_converter_enabled_ = true;
   }
   current_stream()->EnableBitstreamConverter();
 }
 
-void AdaptiveDemuxerStream::ChangeCurrentStream(int index) {
-  bool needs_bitstream_converter_enabled;
-  {
-    base::AutoLock auto_lock(lock_);
-    current_stream_index_ = index;
-    DCHECK(streams_[current_stream_index_]);
-    needs_bitstream_converter_enabled = bitstream_converter_enabled_;
-  }
+void AdaptiveDemuxerStream::OnAdaptiveDemuxerSeek() {
+  base::AutoLock auto_lock(lock_);
+
+  last_buffer_timestamp_ = kNoTimestamp;
+
+  // TODO Figure out what to do if this happens during a stream switch.

[scherkus (not reviewing), 2011/05/23 23:17:06]

TODO -> TODO(acolwell):

[acolwell GONE FROM CHROMIUM, 2011/05/23 23:46:40]

Done.

+}
+
+// This method initiates a stream switch. The diagram below shows the steps
+// involved.
+//
+//                           +-----------------------+
+// ChangeCurrentStream() ->  | Store stream switch   |
+//                           | index, seek_function, |
+//                           |    and switch_cb.     |
+//                           +-----------------------+
+//                                   |
+//                                  \|/
+//                          +-------------------+ Yes
+//                          | Are there pending | -----> (Wait for OnReadDone())
+//                          | Read()s on the    |                |
+//                          | current stream?   | <-----+       \|/
+//                          +-------------------+       +--- OnReadDone()
+//                                   | No
+//                                  \|/
+//                              StartSwitch()
+//                                   |
+//                                  \|/
+//                        +------------------------+ No
+//                        | Have buffer timestamp? |-----+
+//                        +------------------------+     |
+//                                   | Yes               |
+//                                  \|/                  |
+//                     +-----------------------------+   |
+//                     | Seek stream to timestamp    |   |
+//                     | using switch_seek_function_ |   |
+//                     +-----------------------------+   |
+//                                   |                   |
+//                                  \|/                  |
+//                              OnSwitchSeekDone() <-----+
+//                                   |
+//                                  \|/
+//                         +------------------+ No
+//                         | Seek successful? |----------+
+//                         +------------------+          |
+//                                   | Yes               |
+//                                  \|/                  |
+//                     +------------------------------+  |
+//                     | Update current_stream_index_ |  |
+//                     +------------------------------+  |
+//                                   |                   |
+//                                  \|/                  |
+//                      +---------------------------+    |
+//                      | Call Read() on new stream |<---+
+//                      |    for deferred reads.    |
+//                      +---------------------------+
+//                                   |
+//                                  \|/
+//                            switch_cb(OK | ERROR)
+//
+// NOTE: Any AdaptiveDemuxerStream::Read() calls that occur during the stream
+//       switch will be deferred until the switch has completed. The callbacks
+//       will be queued on |read_cb_queue_|, but no Read() will be issued on the
+//       current stream.
+void AdaptiveDemuxerStream::ChangeCurrentStream(
+    int index,
+    const SeekFunction& seek_function,
+    const PipelineStatusCB& switch_cb) {
+  DCHECK_GE(index, 0);
+
+  PipelineStatusCB error_cb;
+  PipelineStatus error_status = PIPELINE_ERROR_INVALID_STATE;
+
+  bool start_switch = false;
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    DCHECK_LE(index, streams_.size());
+    DCHECK(streams_[index].get());
+    DCHECK(!IsSwitchPending_Locked());
+
+    // TODO - Still need to handle the case where the stream has ended.
+    if (index == current_stream_index_) {
+      error_cb = switch_cb;
+      error_status = PIPELINE_OK;
+    } else {
+      switch_cb_ = switch_cb;
+      switch_index_ = index;
+      switch_seek_function_ = seek_function;
+
+      start_switch = CanStartSwitch_Locked();
+    }
+  }
+
+  if (!error_cb.is_null()) {
+    error_cb.Run(error_status);
+    return;
+  }
+
+  if (start_switch)
+    StartSwitch();
+}
+
+void AdaptiveDemuxerStream::OnReadDone(Buffer* buffer) {
+  ReadCallback read_cb;
+  bool start_switch = false;
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    pending_reads_--;
+
+    DCHECK_GE(pending_reads_, 0);
+    DCHECK_GE(read_cb_queue_.size(), 0);
+
+    read_cb = read_cb_queue_.front();
+    read_cb_queue_.pop_front();
+
+    if (buffer && !buffer->IsEndOfStream())
+      last_buffer_timestamp_ = buffer->GetTimestamp();
+
+    start_switch = IsSwitchPending_Locked() && CanStartSwitch_Locked();
+  }
+
+  if (!read_cb.is_null())
+    read_cb.Run(buffer);
+
+  if (start_switch)
+    StartSwitch();
+}
+
+bool AdaptiveDemuxerStream::IsSwitchPending_Locked() const {
+  lock_.AssertAcquired();
+  return !switch_cb_.is_null();
+}
+
+bool AdaptiveDemuxerStream::CanStartSwitch_Locked() const {
+  lock_.AssertAcquired();
+  return (pending_reads_ == 0);
+}
+
+void AdaptiveDemuxerStream::StartSwitch() {
+  SeekFunction seek_function;
+  base::TimeDelta seek_point;
+
+  {
+    base::AutoLock auto_lock(lock_);
+    DCHECK(IsSwitchPending_Locked());
+    DCHECK_EQ(pending_reads_, 0);
+    DCHECK_GE(switch_index_, 0);
+
+    seek_point = last_buffer_timestamp_;
+    seek_function = switch_seek_function_;
+
+    // TODO add code to call switch_cb_ if we are at the end of the stream now.
+  }
+
+  if (seek_point == kNoTimestamp) {
+    // We haven't seen a buffer so there is no need to seek. Just move on to
+    // the next stage in the switch process.
+    OnSwitchSeekDone(PIPELINE_OK, kNoTimestamp);
+    return;
+  }
+
+  DCHECK(!seek_function.is_null());
+  seek_function.Run(seek_point,
+                    base::Bind(&AdaptiveDemuxerStream::OnSwitchSeekDone, this));
+}
+
+void AdaptiveDemuxerStream::OnSwitchSeekDone(PipelineStatus status,
+                                             base::TimeDelta seek_time) {
+  DemuxerStream* stream = NULL;
+  PipelineStatusCB switch_cb;
+  int reads_to_request = 0;
+  bool needs_bitstream_converter_enabled = false;
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    if (status == PIPELINE_OK) {
+      DCHECK(streams_[switch_index_]);
+
+      current_stream_index_ = switch_index_;
+      needs_bitstream_converter_enabled = bitstream_converter_enabled_;
+    }
+
+    // Clear stream switch state.
+    switch_cb = switch_cb_;
+    switch_cb_.Reset();
+    switch_index_ = -1;
+    switch_seek_function_.Reset();
+
+    // Get the number of outstanding Read()s on this object.
+    reads_to_request = read_cb_queue_.size();
+
+    DCHECK_EQ(pending_reads_, 0);
+    pending_reads_ = reads_to_request;
+
+    stream = streams_[current_stream_index_];
+  }
+
   if (needs_bitstream_converter_enabled)
     EnableBitstreamConverter();
+
+  if (stream) {
+    // Make the Read() calls that were deferred during the stream switch.
+    for(;reads_to_request > 0; --reads_to_request)
+      stream->Read(base::Bind(&AdaptiveDemuxerStream::OnReadDone, this));
+  }
+
+  // Signal that the stream switch has completed.
+  if (!switch_cb.is_null())
+    switch_cb.Run(status);
 }
 
 //
 // AdaptiveDemuxer
 //
 
 AdaptiveDemuxer::AdaptiveDemuxer(DemuxerVector const& demuxers,
                                  int initial_audio_demuxer_index,
                                  int initial_video_demuxer_index)
     : demuxers_(demuxers),
       current_audio_demuxer_index_(initial_audio_demuxer_index),
-      current_video_demuxer_index_(initial_video_demuxer_index) {
+      current_video_demuxer_index_(initial_video_demuxer_index),
+      playback_rate_(0),
+      switch_pending_(false),
+      stream_switch_manager_(new StreamSwitchManager()){
   DCHECK(!demuxers_.empty());
   DCHECK_GE(current_audio_demuxer_index_, -1);
   DCHECK_GE(current_video_demuxer_index_, -1);
   DCHECK_LT(current_audio_demuxer_index_, static_cast<int>(demuxers_.size()));
   DCHECK_LT(current_video_demuxer_index_, static_cast<int>(demuxers_.size()));
   DCHECK(current_audio_demuxer_index_ != -1 ||
          current_video_demuxer_index_ != -1);
   AdaptiveDemuxerStream::StreamVector audio_streams, video_streams;
   for (size_t i = 0; i < demuxers_.size(); ++i) {
     audio_streams.push_back(demuxers_[i]->GetStream(DemuxerStream::AUDIO));
     video_streams.push_back(demuxers_[i]->GetStream(DemuxerStream::VIDEO));
   }
   if (current_audio_demuxer_index_ >= 0) {
     audio_stream_ = new AdaptiveDemuxerStream(
         audio_streams, current_audio_demuxer_index_);
   }
   if (current_video_demuxer_index_ >= 0) {
     video_stream_ = new AdaptiveDemuxerStream(
         video_streams, current_video_demuxer_index_);
   }
 
+  stream_switch_manager_->Init(this);
+
   // TODO(fischman): any streams in the underlying demuxers that aren't being
   // consumed currently need to be sent to /dev/null or else FFmpegDemuxer will
   // hold data for them in memory, waiting for them to get drained by a
   // non-existent decoder.
 }
 
 AdaptiveDemuxer::~AdaptiveDemuxer() {}
 
-void AdaptiveDemuxer::ChangeCurrentDemuxer(int audio_index, int video_index) {
+// Switches the current video stream. The diagram below describes the switch
+// process.
+//                          +-------------------------------+
+// ChangeVideoStream() ---> |          video_index          |
+//                          |              ==               | Yes.
+//                          | current_video_demuxer_index_? |--> done_cb(OK)
+//                          +-------------------------------+
+//                                          | No.
+//                                         \|/
+//                         Call video_stream_->ChangeCurrentStream()
+//                                          |
+//                                         \|/
+//                          (Wait for ChangeVideoStreamDone())
+//                                          |
+//                                         \|/
+//                                ChangeVideoStreamDone()
+//                                          |
+//                                         \|/
+//                             +------------------------+ No
+//                             | Was switch successful? |---> done_cb(ERROR)
+//                             +------------------------+
+//                                          |
+//                                         \|/
+//                            Update current_video_demuxer_index_
+//                            & SetPlaybackRate() on new stream.
+//                                          |
+//                                         \|/
+//                                      done_cb(OK).
+//
+void AdaptiveDemuxer::ChangeVideoStream(int video_index,
+                                        const PipelineStatusCB& done_cb) {
   // TODO(fischman): this is currently broken because when a new Demuxer is to
   // be used we need to set_host(host()) it, and we need to set_host(NULL) the
   // current Demuxer if it's no longer being used.
+  AdaptiveDemuxerStream* stream = NULL;
+  bool switching_to_current_stream = false;
+  AdaptiveDemuxerStream::SeekFunction seek_function;
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    DCHECK(video_stream_);
+    DCHECK(!switch_pending_);
+    if (current_video_demuxer_index_ == video_index) {
+      switching_to_current_stream = true;
+    } else {
+      stream = video_stream_;
+      switch_pending_ = true;
+      seek_function = base::Bind(&AdaptiveDemuxer::StartStreamSwitchSeek,
+                                 this,
+                                 DemuxerStream::VIDEO,
+                                 video_index);
+    }
+  }
+
+  if (switching_to_current_stream) {
+    done_cb.Run(PIPELINE_OK);
+    return;
+  }
+
+  DCHECK(stream);
+  stream->ChangeCurrentStream(
+      video_index, seek_function,
+      base::Bind(&AdaptiveDemuxer::ChangeVideoStreamDone, this, video_index,
+                 done_cb));
+}
+
+int AdaptiveDemuxer::GetCurrentVideoId() const {
   base::AutoLock auto_lock(lock_);
-  current_audio_demuxer_index_ = audio_index;
-  current_video_demuxer_index_ = video_index;
-  if (audio_stream_)
-    audio_stream_->ChangeCurrentStream(audio_index);
-  if (video_stream_)
-    video_stream_->ChangeCurrentStream(video_index);
+  return current_video_demuxer_index_;
+}
+
+AdaptiveDemuxer::StreamIdVector AdaptiveDemuxer::GetVideoIds() const {
+  StreamIdVector ids;
+  base::AutoLock auto_lock(lock_);
+
+  for (int i = 0; i < demuxers_.size(); i++) {
+    if (demuxers_[i]->GetStream(DemuxerStream::VIDEO))
+      ids.push_back(i);
+  }
+
+  return ids;
+}
+
+void AdaptiveDemuxer::ChangeVideoStreamDone(int new_stream_index,
+                                            const PipelineStatusCB& done_cb,
+                                            PipelineStatus status) {
+  {
+    base::AutoLock auto_lock(lock_);
+
+    switch_pending_ = false;
+
+    if (status == PIPELINE_OK) {
+      demuxers_[current_video_demuxer_index_]->SetPlaybackRate(0);
+      current_video_demuxer_index_ = new_stream_index;
+      demuxers_[current_video_demuxer_index_]->SetPlaybackRate(playback_rate_);
+    }
+  }
+
+  done_cb.Run(status);
 }
 
 Demuxer* AdaptiveDemuxer::current_demuxer(DemuxerStream::Type type) {
   base::AutoLock auto_lock(lock_);
   switch (type) {
     case DemuxerStream::AUDIO:
       return (current_audio_demuxer_index_ < 0) ? NULL :
           demuxers_[current_audio_demuxer_index_];
     case DemuxerStream::VIDEO:
       return (current_video_demuxer_index_ < 0) ? NULL :
@@ skipping 79 matching lines @@
 
   base::Lock lock_;
   int remaining_count_;
   FilterStatusCB orig_cb_;
   PipelineStatus overall_status_;
 
   DISALLOW_COPY_AND_ASSIGN(CountingStatusCB);
 };
 
 void AdaptiveDemuxer::Stop(FilterCallback* callback) {
+  stream_switch_manager_->Stop();
+
   // Stop() must be called on all of the demuxers even though only one demuxer
   // is  actively delivering audio and another one is delivering video. This
   // just satisfies the contract that all demuxers must have Stop() called on
   // them before they are destroyed.
   //
   // TODO: Remove CountingCallback once Stop() is converted to FilterStatusCB.
   CountingCallback* wrapper = new CountingCallback(demuxers_.size(), callback);
   for (size_t i = 0; i < demuxers_.size(); ++i)
     demuxers_[i]->Stop(wrapper->GetACallback());
 }
 
 void AdaptiveDemuxer::Seek(base::TimeDelta time, const FilterStatusCB& cb) {
+
+  if (audio_stream_)
+    audio_stream_->OnAdaptiveDemuxerSeek();
+
+  if (video_stream_)
+    video_stream_->OnAdaptiveDemuxerSeek();
+
   Demuxer* audio = current_demuxer(DemuxerStream::AUDIO);
   Demuxer* video = current_demuxer(DemuxerStream::VIDEO);
   int count = (audio ? 1 : 0) + (video && audio != video ? 1 : 0);
   CountingStatusCB* wrapper = new CountingStatusCB(count, cb);
   if (audio)
     audio->Seek(time, wrapper->GetACallback());
   if (video && audio != video)
     video->Seek(time, wrapper->GetACallback());
 }
 
 void AdaptiveDemuxer::OnAudioRendererDisabled() {
   for (size_t i = 0; i < demuxers_.size(); ++i)
     demuxers_[i]->OnAudioRendererDisabled();
 }
 
 void AdaptiveDemuxer::set_host(FilterHost* filter_host) {
   Demuxer* audio = current_demuxer(DemuxerStream::AUDIO);
   Demuxer* video = current_demuxer(DemuxerStream::VIDEO);
   if (audio) audio->set_host(filter_host);
   if (video && audio != video) video->set_host(filter_host);
 }
 
 void AdaptiveDemuxer::SetPlaybackRate(float playback_rate) {
+  {
+    base::AutoLock auto_lock(lock_);
+    if (playback_rate_ == 0 && playback_rate > 0) {
+      stream_switch_manager_->Play();
+    } else if (playback_rate_ > 0 && playback_rate == 0) {
+      stream_switch_manager_->Pause();
+    }
+
+    playback_rate_ = playback_rate;
+  }
+
   Demuxer* audio = current_demuxer(DemuxerStream::AUDIO);
   Demuxer* video = current_demuxer(DemuxerStream::VIDEO);
   if (audio) audio->SetPlaybackRate(playback_rate);
   if (video && audio != video) video->SetPlaybackRate(playback_rate);
 }
 
 void AdaptiveDemuxer::SetPreload(Preload preload) {
   for (size_t i = 0; i < demuxers_.size(); ++i)
     demuxers_[i]->SetPreload(preload);
 }
 
 scoped_refptr<DemuxerStream> AdaptiveDemuxer::GetStream(
     DemuxerStream::Type type) {
   switch (type) {
     case DemuxerStream::AUDIO:
       return audio_stream_;
     case DemuxerStream::VIDEO:
       return video_stream_;
     default:
       LOG(DFATAL) << "Unexpected type " << type;
       return NULL;
   }
 }
 
+void AdaptiveDemuxer::StartStreamSwitchSeek(
+    DemuxerStream::Type type,
+    int stream_index,
+    base::TimeDelta seek_time,
+    const AdaptiveDemuxerStream::SeekFunctionCB& seek_cb) {
+  DCHECK_GE(stream_index, 0);
+  DCHECK(!seek_cb.is_null());
+
+  Demuxer* demuxer = NULL;
+  base::TimeDelta seek_point;
+  FilterStatusCB filter_cb;
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    demuxer = demuxers_[stream_index];
+
+    if (GetSeekTimeAfter(demuxer->GetStream(type)->GetAVStream(), seek_time,
+                         &seek_point)) {
+      // We found a seek point.
+      filter_cb = base::Bind(&AdaptiveDemuxer::OnStreamSeekDone, this,
+                             seek_point, seek_cb);
+    } else {
+      // We didn't find a seek point. Assume we don't have index data for it
+      // yet. Seek to the specified time to force index data to be loaded.
+      seek_point = seek_time;
+      filter_cb = base::Bind(&AdaptiveDemuxer::OnIndexSeekDone, this,
+                             type, stream_index, seek_time, seek_cb);
+    }
+  }
+
+  DCHECK(demuxer);
+  demuxer->Seek(seek_point, filter_cb);
+}
+
+void AdaptiveDemuxer::OnIndexSeekDone(
+    DemuxerStream::Type type,
+    int stream_index,
+    base::TimeDelta seek_time,
+    const AdaptiveDemuxerStream::SeekFunctionCB& seek_cb,
+    PipelineStatus status) {
+  base::TimeDelta seek_point;
+  FilterStatusCB filter_cb;
+
+  Demuxer* demuxer = NULL;
+
+  if (status != PIPELINE_OK) {
+    seek_cb.Run(status, base::TimeDelta());
+    return;
+  }
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    demuxer = demuxers_[stream_index];
+    DCHECK(demuxer);
+
+    // Look for a seek point now that we have index data.
+    if (GetSeekTimeAfter(demuxer->GetStream(type)->GetAVStream(), seek_time,
+                         &seek_point)) {
+      filter_cb = base::Bind(&AdaptiveDemuxer::OnStreamSeekDone, this,
+                             seek_point, seek_cb);
+    } else {
+      // Failed again to find a seek point. Clear the demuxer so that
+      // a seek error will be reported.
+      demuxer = NULL;
+    }
+  }
+
+  if (!demuxer) {
+    seek_cb.Run(PIPELINE_ERROR_INITIALIZATION_FAILED, base::TimeDelta());
+    return;
+  }
+
+  demuxer->Seek(seek_point, filter_cb);
+}
+
+void AdaptiveDemuxer::OnStreamSeekDone(
+    base::TimeDelta seek_point,
+    const AdaptiveDemuxerStream::SeekFunctionCB& seek_cb,
+    PipelineStatus status) {
+  if (status != PIPELINE_OK) {
+    seek_cb.Run(status, base::TimeDelta());
+    return;
+  }
+
+  seek_cb.Run(PIPELINE_OK, seek_point);
+}
+
+
 //
 // AdaptiveDemuxerFactory
 //
 
 AdaptiveDemuxerFactory::AdaptiveDemuxerFactory(
     DemuxerFactory* delegate_factory)
     : delegate_factory_(delegate_factory) {
   DCHECK(delegate_factory);
 }
 
@@ skipping 172 matching lines @@
     delete cb;
     return;
   }
   DemuxerAccumulator* accumulator = new DemuxerAccumulator(
       audio_index, video_index, urls.size(), cb);
   for (size_t i = 0; i < urls.size(); ++i)
     delegate_factory_->Build(urls[i], accumulator->GetNthCallback(i));
 }
 
 }  // namespace media