Fix ChunkDemuxer seek and init callback dispatch.

media/filters/chunk_demuxer.cc

 // Copyright (c) 2012 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 "media/filters/chunk_demuxer.h"
 
 #include <algorithm>
 #include <deque>
 #include <limits>
 
 #include "base/bind.h"
 #include "base/callback_helpers.h"
 #include "base/location.h"
 #include "base/message_loop/message_loop_proxy.h"
 #include "media/base/audio_decoder_config.h"
+#include "media/base/bind_to_loop.h"
 #include "media/base/stream_parser_buffer.h"
 #include "media/base/video_decoder_config.h"
 #include "media/filters/stream_parser_factory.h"
 #include "media/webm/webm_webvtt_parser.h"
 
 using base::TimeDelta;
 
 namespace media {
 
 // Contains state belonging to a source id.
@@ skipping 174 matching lines @@
 
   ChunkDemuxerStream(const AudioDecoderConfig& audio_config,
                      const LogCB& log_cb);
   ChunkDemuxerStream(const VideoDecoderConfig& video_config,
                      const LogCB& log_cb);
   virtual ~ChunkDemuxerStream();
 
   void StartWaitingForSeek();
   void Seek(TimeDelta time);
   void CancelPendingSeek();
-  bool IsSeekPending() const;
+  bool IsSeekWaitingForData() const;
 
   // Add buffers to this stream.  Buffers are stored in SourceBufferStreams,
   // which handle ordering and overlap resolution.
   // Returns true if buffers were successfully added.
   bool Append(const StreamParser::BufferQueue& buffers);
 
   // Signal to the stream that duration has changed to |duration|.
-  void OnSetDuration(base::TimeDelta duration);
+  void OnSetDuration(TimeDelta duration);
 
   // Returns the range of buffered data in this stream, capped at |duration|.
-  Ranges<TimeDelta> GetBufferedRanges(base::TimeDelta duration) const;
+  Ranges<TimeDelta> GetBufferedRanges(TimeDelta duration) const;
 
   // Signal to the stream that buffers handed in through subsequent calls to
   // Append() belong to a media segment that starts at |start_timestamp|.
   void OnNewMediaSegment(TimeDelta start_timestamp);
 
   // Called when midstream config updates occur.
   // Returns true if the new config is accepted.
   // Returns false if the new config should trigger an error.
   bool UpdateAudioConfig(const AudioDecoderConfig& config);
   bool UpdateVideoConfig(const VideoDecoderConfig& config);
@@ skipping 95 matching lines @@
   {
     base::AutoLock auto_lock(lock_);
     ChangeState_Locked(CANCELED);
     std::swap(read_cbs_, read_cbs);
   }
 
   for (ReadCBQueue::iterator it = read_cbs.begin(); it != read_cbs.end(); ++it)
     it->Run(kAborted, NULL);
 }
 
-bool ChunkDemuxerStream::IsSeekPending() const {
+bool ChunkDemuxerStream::IsSeekWaitingForData() const {
   base::AutoLock auto_lock(lock_);
   return stream_->IsSeekPending();
 }
 
 void ChunkDemuxerStream::OnNewMediaSegment(TimeDelta start_timestamp) {
   base::AutoLock auto_lock(lock_);
   stream_->OnNewMediaSegment(start_timestamp);
 }
 
 bool ChunkDemuxerStream::Append(const StreamParser::BufferQueue& buffers) {
@@ skipping 10 matching lines @@
     }
     CreateReadDoneClosures_Locked(&closures);
   }
 
   for (ClosureQueue::iterator it = closures.begin(); it != closures.end(); ++it)
     it->Run();
 
   return true;
 }
 
-void ChunkDemuxerStream::OnSetDuration(base::TimeDelta duration) {
+void ChunkDemuxerStream::OnSetDuration(TimeDelta duration) {
   base::AutoLock auto_lock(lock_);
   stream_->OnSetDuration(duration);
 }
 
 Ranges<TimeDelta> ChunkDemuxerStream::GetBufferedRanges(
-    base::TimeDelta duration) const {
+    TimeDelta duration) const {
   base::AutoLock auto_lock(lock_);
   Ranges<TimeDelta> range = stream_->GetBufferedTime();
 
   if (range.size() == 0u)
     return range;
 
   // Clamp the end of the stream's buffered ranges to fit within the duration.
   // This can be done by intersecting the stream's range with the valid time
   // range.
   Ranges<TimeDelta> valid_time_range;
@@ skipping 186 matching lines @@
       user_specified_duration_(-1) {
   DCHECK(!open_cb_.is_null());
   DCHECK(!need_key_cb_.is_null());
 }
 
 void ChunkDemuxer::Initialize(DemuxerHost* host, const PipelineStatusCB& cb) {
   DVLOG(1) << "Init()";
 
   base::AutoLock auto_lock(lock_);
 
+  init_cb_ = BindToCurrentLoop(cb);
   if (state_ == SHUTDOWN) {
-    base::MessageLoopProxy::current()->PostTask(FROM_HERE, base::Bind(
-        cb, DEMUXER_ERROR_COULD_NOT_OPEN));
+    base::ResetAndReturn(&init_cb_).Run(DEMUXER_ERROR_COULD_NOT_OPEN);
     return;
   }
   DCHECK_EQ(state_, WAITING_FOR_INIT);
   host_ = host;
 
   ChangeState_Locked(INITIALIZING);
-  init_cb_ = cb;
 
   base::ResetAndReturn(&open_cb_).Run();
 }
 
 void ChunkDemuxer::Stop(const base::Closure& callback) {
   DVLOG(1) << "Stop()";
   Shutdown();
   callback.Run();
 }
 
 void ChunkDemuxer::Seek(TimeDelta time, const PipelineStatusCB& cb) {
   DVLOG(1) << "Seek(" << time.InSecondsF() << ")";
   DCHECK(time >= TimeDelta());
   DCHECK(seek_cb_.is_null());
 
   PipelineStatus status = PIPELINE_ERROR_INVALID_STATE;
-  {
-    base::AutoLock auto_lock(lock_);
+  base::AutoLock auto_lock(lock_);
 
-    if (state_ == INITIALIZED || state_ == ENDED) {
-      if (audio_)
-        audio_->Seek(time);
+  seek_cb_ = BindToCurrentLoop(cb);
+  if (state_ == INITIALIZED || state_ == ENDED) {
+    if (audio_)
+      audio_->Seek(time);
 
-      if (video_)
-        video_->Seek(time);
+    if (video_)
+      video_->Seek(time);
 
-      if (IsSeekPending_Locked()) {
-        DVLOG(1) << "Seek() : waiting for more data to arrive.";
-        seek_cb_ = cb;
-        return;
-      }
+    if (IsSeekWaitingForData_Locked()) {
+      DVLOG(1) << "Seek() : waiting for more data to arrive.";
+      return;
+    }
 
-      status = PIPELINE_OK;
-    }
+    status = PIPELINE_OK;
   }
 
-  cb.Run(status);
+  base::ResetAndReturn(&seek_cb_).Run(status);
 }
 
 void ChunkDemuxer::OnAudioRendererDisabled() {
   base::AutoLock auto_lock(lock_);
   audio_->Shutdown();
   disabled_audio_ = audio_.Pass();
 }
 
 // Demuxer implementation.
 DemuxerStream* ChunkDemuxer::GetStream(DemuxerStream::Type type) {
@@ skipping 20 matching lines @@
     return;
 
   if (audio_)
     audio_->StartWaitingForSeek();
 
   if (video_)
     video_->StartWaitingForSeek();
 }
 
 void ChunkDemuxer::CancelPendingSeek() {
-  PipelineStatusCB cb;
-  {
-    base::AutoLock auto_lock(lock_);
-    if (IsSeekPending_Locked() && !seek_cb_.is_null()) {
-      std::swap(cb, seek_cb_);
-    }
-    if (audio_)
-      audio_->CancelPendingSeek();
+  base::AutoLock auto_lock(lock_);
+  DCHECK(seek_cb_.is_null() != IsSeekWaitingForData_Locked());
 
-    if (video_)
-      video_->CancelPendingSeek();
-  }
+  if (audio_)
+    audio_->CancelPendingSeek();
 
-  if (!cb.is_null())
-    cb.Run(PIPELINE_OK);
+  if (video_)
+    video_->CancelPendingSeek();
+
+  if (!seek_cb_.is_null())
+    base::ResetAndReturn(&seek_cb_).Run(PIPELINE_OK);
 }
 
 ChunkDemuxer::Status ChunkDemuxer::AddId(const std::string& id,
                                          const std::string& type,
                                          std::vector<std::string>& codecs) {
   DCHECK_GT(codecs.size(), 0u);
   base::AutoLock auto_lock(lock_);
 
   if ((state_ != WAITING_FOR_INIT && state_ != INITIALIZING) || IsValidId(id))
     return kReachedIdLimit;
@@ skipping 115 matching lines @@
 
 void ChunkDemuxer::AppendData(const std::string& id,
                               const uint8* data,
                               size_t length) {
   DVLOG(1) << "AppendData(" << id << ", " << length << ")";
 
   DCHECK(!id.empty());
 
   Ranges<TimeDelta> ranges;
 
-  PipelineStatusCB cb;
   {
     base::AutoLock auto_lock(lock_);
 
-    // Capture if the SourceBuffer has a pending seek before we start parsing.
-    bool old_seek_pending = IsSeekPending_Locked();
+    // Capture if any of the SourceBuffers are waiting for data before we start
+    // parsing.
+    bool old_waiting_for_data = IsSeekWaitingForData_Locked();
 
     if (state_ == ENDED) {
       ChangeState_Locked(INITIALIZED);
 
       if (audio_)
         audio_->CancelEndOfStream();
 
       if (video_)
         video_->CancelEndOfStream();
     }
@@ skipping 26 matching lines @@
 
       case WAITING_FOR_INIT:
       case ENDED:
       case SHUTDOWN:
         DVLOG(1) << "AppendData(): called in unexpected state " << state_;
         return;
     }
 
     // Check to see if data was appended at the pending seek point. This
     // indicates we have parsed enough data to complete the seek.
-    if (old_seek_pending && !IsSeekPending_Locked() && !seek_cb_.is_null()) {
-      std::swap(cb, seek_cb_);
+    if (old_waiting_for_data && !IsSeekWaitingForData_Locked() &&
+        !seek_cb_.is_null()) {
+      base::ResetAndReturn(&seek_cb_).Run(PIPELINE_OK);
     }
 
     ranges = GetBufferedRanges();
   }
 
   for (size_t i = 0; i < ranges.size(); ++i)
     host_->AddBufferedTimeRange(ranges.start(i), ranges.end(i));
-
-  if (!cb.is_null())
-    cb.Run(PIPELINE_OK);
 }
 
 void ChunkDemuxer::Abort(const std::string& id) {
   DVLOG(1) << "Abort(" << id << ")";
   base::AutoLock auto_lock(lock_);
   DCHECK(!id.empty());
   CHECK(IsValidId(id));
   source_state_map_[id]->Abort();
 }
 
@@ skipping 21 matching lines @@
 void ChunkDemuxer::SetDuration(double duration) {
   base::AutoLock auto_lock(lock_);
   DVLOG(1) << "SetDuration(" << duration << ")";
   DCHECK_GE(duration, 0);
 
   if (duration == GetDuration_Locked())
     return;
 
   // Compute & bounds check the TimeDelta representation of duration.
   // This can be different if the value of |duration| doesn't fit the range or
-  // precision of base::TimeDelta.
-  base::TimeDelta min_duration = base::TimeDelta::FromInternalValue(1);
-  base::TimeDelta max_duration =
-      base::TimeDelta::FromInternalValue(kint64max - 1);
+  // precision of TimeDelta.
+  TimeDelta min_duration = TimeDelta::FromInternalValue(1);
+  TimeDelta max_duration = TimeDelta::FromInternalValue(kint64max - 1);
   double min_duration_in_seconds = min_duration.InSecondsF();
   double max_duration_in_seconds = max_duration.InSecondsF();
 
-  base::TimeDelta duration_td;
+  TimeDelta duration_td;
   if (duration == std::numeric_limits<double>::infinity()) {
     duration_td = media::kInfiniteDuration();
   } else if (duration < min_duration_in_seconds) {
     duration_td = min_duration;
   } else if (duration > max_duration_in_seconds) {
     duration_td = max_duration;
   } else {
-    duration_td = base::TimeDelta::FromMicroseconds(
+    duration_td = TimeDelta::FromMicroseconds(
         duration * base::Time::kMicrosecondsPerSecond);
   }
 
-  DCHECK(duration_td > base::TimeDelta());
+  DCHECK(duration_td > TimeDelta());
 
   user_specified_duration_ = duration;
   duration_ = duration_td;
   host_->SetDuration(duration_);
 
   if (audio_)
     audio_->OnSetDuration(duration_);
 
   if (video_)
     video_->OnSetDuration(duration_);
 }
 
 bool ChunkDemuxer::SetTimestampOffset(const std::string& id, TimeDelta offset) {
   base::AutoLock auto_lock(lock_);
   DVLOG(1) << "SetTimestampOffset(" << id << ", " << offset.InSecondsF() << ")";
   CHECK(IsValidId(id));
 
   return source_state_map_[id]->SetTimestampOffset(offset);
 }
 
 void ChunkDemuxer::EndOfStream(PipelineStatus status) {
   DVLOG(1) << "EndOfStream(" << status << ")";
-  PipelineStatusCB cb;
-  {
-    base::AutoLock auto_lock(lock_);
-    DCHECK_NE(state_, WAITING_FOR_INIT);
-    DCHECK_NE(state_, ENDED);
+  base::AutoLock auto_lock(lock_);
+  DCHECK_NE(state_, WAITING_FOR_INIT);
+  DCHECK_NE(state_, ENDED);
 
-    if (state_ == SHUTDOWN || state_ == PARSE_ERROR)
-      return;
+  if (state_ == SHUTDOWN || state_ == PARSE_ERROR)
+    return;
 
-    if (state_ == INITIALIZING) {
-      ReportError_Locked(DEMUXER_ERROR_COULD_NOT_OPEN);
-      return;
-    }
-
-    bool old_seek_pending = IsSeekPending_Locked();
-    if (audio_)
-      audio_->EndOfStream();
-
-    if (video_)
-      video_->EndOfStream();
-
-    // Give a chance to resume the pending seek process.
-    if (status != PIPELINE_OK) {
-      ReportError_Locked(status);
-      return;
-    }
-
-    ChangeState_Locked(ENDED);
-    DecreaseDurationIfNecessary();
-
-    if (old_seek_pending && !IsSeekPending_Locked() && !seek_cb_.is_null())
-      std::swap(cb, seek_cb_);
+  if (state_ == INITIALIZING) {
+    ReportError_Locked(DEMUXER_ERROR_COULD_NOT_OPEN);
+    return;
   }
 
-  if (!cb.is_null())
-    cb.Run(PIPELINE_OK);
+  bool old_waiting_for_data = IsSeekWaitingForData_Locked();
+  if (audio_)
+    audio_->EndOfStream();
+
+  if (video_)
+    video_->EndOfStream();
+
+  // Give a chance to resume the pending seek process.
+  if (status != PIPELINE_OK) {
+    ReportError_Locked(status);
+    return;
+  }
+
+  ChangeState_Locked(ENDED);
+  DecreaseDurationIfNecessary();
+
+  if (old_waiting_for_data && !IsSeekWaitingForData_Locked() &&
+      !seek_cb_.is_null()) {
+    base::ResetAndReturn(&seek_cb_).Run(PIPELINE_OK);
+  }
 }
 
 void ChunkDemuxer::Shutdown() {
   DVLOG(1) << "Shutdown()";
-  PipelineStatusCB cb;
-  {
-    base::AutoLock auto_lock(lock_);
+  base::AutoLock auto_lock(lock_);
 
-    if (state_ == SHUTDOWN)
-      return;
+  if (state_ == SHUTDOWN)
+    return;
 
-    std::swap(cb, seek_cb_);
+  if (audio_)
+    audio_->Shutdown();
 
-    if (audio_)
-      audio_->Shutdown();
+  if (video_)
+    video_->Shutdown();
 
-    if (video_)
-      video_->Shutdown();
+  ChangeState_Locked(SHUTDOWN);
 
-    ChangeState_Locked(SHUTDOWN);
-  }
-
-  if (!cb.is_null())
-    cb.Run(PIPELINE_ERROR_ABORT);
+  if(!seek_cb_.is_null())
+    base::ResetAndReturn(&seek_cb_).Run(PIPELINE_ERROR_ABORT);
 }
 
 void ChunkDemuxer::ChangeState_Locked(State new_state) {
   lock_.AssertAcquired();
   DVLOG(1) << "ChunkDemuxer::ChangeState_Locked() : "
            << state_ << " -> " << new_state;
   state_ = new_state;
 }
 
 ChunkDemuxer::~ChunkDemuxer() {
@@ skipping 21 matching lines @@
       std::swap(cb, seek_cb_);
 
     if (audio_)
       audio_->Shutdown();
 
     if (video_)
       video_->Shutdown();
   }
 
   if (!cb.is_null()) {
-    base::AutoUnlock auto_unlock(lock_);
     cb.Run(error);
     return;
   }
 
   base::AutoUnlock auto_unlock(lock_);
   host_->OnDemuxerError(error);
 }
 
-bool ChunkDemuxer::IsSeekPending_Locked() const {
+bool ChunkDemuxer::IsSeekWaitingForData_Locked() const {
   lock_.AssertAcquired();
-  bool seek_pending = false;
+  bool waiting_for_data = false;
 
   if (audio_)
-    seek_pending = audio_->IsSeekPending();
+    waiting_for_data = audio_->IsSeekWaitingForData();
 
-  if (!seek_pending && video_)
-    seek_pending = video_->IsSeekPending();
+  if (!waiting_for_data && video_)
+    waiting_for_data = video_->IsSeekWaitingForData();
 
-  return seek_pending;
+  return waiting_for_data;
 }
 
 void ChunkDemuxer::OnSourceInitDone(bool success, TimeDelta duration) {
   DVLOG(1) << "OnSourceInitDone(" << success << ", "
            << duration.InSecondsF() << ")";
   lock_.AssertAcquired();
   DCHECK_EQ(state_, INITIALIZING);
   if (!success || (!audio_ && !video_)) {
     ReportError_Locked(DEMUXER_ERROR_COULD_NOT_OPEN);
     return;
   }
 
-  if (duration != base::TimeDelta() && duration_ == kNoTimestamp())
+  if (duration != TimeDelta() && duration_ == kNoTimestamp())
     UpdateDuration(duration);
 
   // Wait until all streams have initialized.
   if ((!source_id_audio_.empty() && !audio_) ||
       (!source_id_video_.empty() && !video_))
     return;
 
   if (audio_)
     audio_->Seek(TimeDelta());
 
@@ skipping 100 matching lines @@
   TextTrack* text_track,
   const StreamParser::BufferQueue& buffers) {
   lock_.AssertAcquired();
   DCHECK_NE(state_, SHUTDOWN);
 
   // TODO(matthewjheaney): IncreaseDurationIfNecessary
 
   for (StreamParser::BufferQueue::const_iterator itr = buffers.begin();
        itr != buffers.end(); ++itr) {
     const StreamParserBuffer* const buffer = itr->get();
-    const base::TimeDelta start = buffer->GetTimestamp();
-    const base::TimeDelta end = start + buffer->GetDuration();
+    const TimeDelta start = buffer->GetTimestamp();
+    const TimeDelta end = start + buffer->GetDuration();
 
     std::string id, settings, content;
 
     WebMWebVTTParser::Parse(buffer->GetData(),
                             buffer->GetDataSize(),
                             &id, &settings, &content);
 
     text_track->addWebVTTCue(start, end, id, content, settings);
   }
 
@@ skipping 23 matching lines @@
     audio_->OnNewMediaSegment(timestamp);
   if (video_ && source_id == source_id_video_)
     video_->OnNewMediaSegment(timestamp);
 }
 
 bool ChunkDemuxer::IsValidId(const std::string& source_id) const {
   lock_.AssertAcquired();
   return source_state_map_.count(source_id) > 0u;
 }
 
-void ChunkDemuxer::UpdateDuration(base::TimeDelta new_duration) {
+void ChunkDemuxer::UpdateDuration(TimeDelta new_duration) {
   DCHECK(duration_ != new_duration);
   user_specified_duration_ = -1;
   duration_ = new_duration;
   host_->SetDuration(new_duration);
 }
 
 void ChunkDemuxer::IncreaseDurationIfNecessary(
     const StreamParser::BufferQueue& buffers,
     ChunkDemuxerStream* stream) {
   DCHECK(!buffers.empty());
   if (buffers.back()->GetTimestamp() <= duration_)
     return;
 
   Ranges<TimeDelta> ranges = stream->GetBufferedRanges(kInfiniteDuration());
   DCHECK_GT(ranges.size(), 0u);
 
-  base::TimeDelta last_timestamp_buffered = ranges.end(ranges.size() - 1);
+  TimeDelta last_timestamp_buffered = ranges.end(ranges.size() - 1);
   if (last_timestamp_buffered > duration_)
     UpdateDuration(last_timestamp_buffered);
 }
 
 void ChunkDemuxer::DecreaseDurationIfNecessary() {
   Ranges<TimeDelta> ranges = GetBufferedRanges();
   if (ranges.size() == 0u)
     return;
 
-  base::TimeDelta last_timestamp_buffered = ranges.end(ranges.size() - 1);
+  TimeDelta last_timestamp_buffered = ranges.end(ranges.size() - 1);
   if (last_timestamp_buffered < duration_)
     UpdateDuration(last_timestamp_buffered);
 }
 
 Ranges<TimeDelta> ChunkDemuxer::GetBufferedRanges() const {
   if (audio_ && !video_)
     return audio_->GetBufferedRanges(duration_);
   else if (!audio_ && video_)
     return video_->GetBufferedRanges(duration_);
   return ComputeIntersection();
 }
 
 }  // namespace media