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

[scherkus (not reviewing), 2011/05/23 18:26:29]

nit: remove blank line

[acolwell GONE FROM CHROMIUM, 2011/05/23 22:45:45]

Done.

+ public:
+  StreamSwitchManager();
+  ~StreamSwitchManager();

[scherkus (not reviewing), 2011/05/23 18:26:29]

virtual

[acolwell GONE FROM CHROMIUM, 2011/05/23 22:45:45]

Done.

+
+  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_Locked();
+
+  // The demuxer that owns this object.
+  AdaptiveDemuxer* demuxer_;
+
+  // Guards the members below.  Only held for simple variable reads/writes, not
+  // during async operation.
+  base::Lock lock_;
+
+  // 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);
+  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() {
+  base::AutoLock auto_lock(lock_);

[scherkus (not reviewing), 2011/05/23 18:26:29]

what are we locking ourselves from?

which threads are running through these objects?

[acolwell GONE FROM CHROMIUM, 2011/05/23 22:45:45]

The pipeline & ffmpeg_demuxer threads were accessing this code. I've made
modifications so that the lock isn't needed and this class always runs on the
pipeline thread.

+
+  DCHECK(!playing_);
+
+  playing_ = true;
+
+  if (video_ids_.size() > 1 && !switch_timer_running_) {
+    StartSwitchTimer_Locked();
+  }
+}
+
+void StreamSwitchManager::Pause() {
+  base::AutoLock auto_lock(lock_);
+  DCHECK(playing_);
+  playing_ = false;
+}
+
+void StreamSwitchManager::Stop() {
+  base::AutoLock auto_lock(lock_);
+
+  DCHECK(!playing_);
+  demuxer_ = NULL;
+}
+
+void StreamSwitchManager::OnSwitchTimer() {
+  int new_stream_id = -1;
+  AdaptiveDemuxer* demuxer = NULL;
+
+  {
+    base::AutoLock auto_lock(lock_);
+    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 = demuxer_;
+        new_stream_id = video_ids_[new_id_index];
+      }
+    }
+
+    StartSwitchTimer_Locked();
+  }
+
+  if (demuxer && new_stream_id != -1) {
+    demuxer->ChangeVideoStream(new_stream_id,
+                               base::Bind(&StreamSwitchManager::OnSwitchDone,
+                                          this));
+  }
+}
+
+void StreamSwitchManager::OnSwitchDone(PipelineStatus status) {
+  base::AutoLock auto_lock(lock_);
+  switch_pending_ = false;
+}
+
+void StreamSwitchManager::StartSwitchTimer_Locked() {
+  lock_.AssertAcquired();
+
+  switch_timer_running_ = true;
+  MessageLoop::current()->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);

[scherkus (not reviewing), 2011/05/23 18:26:29]

want to raise it again since it seems like it might solve a lot of pain...

what if we had a push model for demuxer -> decoder?

it'd do away with a lot of this silly queueing stuff

alternatively, what if we only allowed one outstanding read at any time?

could simplify things!

[acolwell GONE FROM CHROMIUM, 2011/05/23 22:45:45]

It wouldn't remove queuing. It would just move the queue to the decoder. The
push vs pull discussion should happen outside of this forum.
That could simplify this code slightly. I'm not sure why the multiple Read()s
were allowed in the first place. I could attack that in a follow-up CL. Doing it
before hand wouldn't significantly change this CL. It would just change
pending_reads to a bool and change the CB queue to a single CB. I don't think
that is enough to defer this CL.

+
+    // 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;
+
+  // XXAJC Figure out what to do if this happens during a stream switch.

[scherkus (not reviewing), 2011/05/23 18:26:29]

should these be TODOs / NOTIMPLEMENTED ?

[acolwell GONE FROM CHROMIUM, 2011/05/23 22:45:45]

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());
+
+    // XXAJC - 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_;
+
+    // XXAJC 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