Refactor ChunkDemuxer code so all source state is in a SourceState class.

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_proxy.h"
 #include "media/base/audio_decoder_config.h"
 #include "media/base/stream_parser_buffer.h"
 #include "media/base/video_decoder_config.h"
 #include "media/filters/stream_parser_factory.h"
 
 using base::TimeDelta;
 
 namespace media {
 
+// Contains state belonging to a source id.
+class SourceState {

[scherkus (not reviewing), 2013/04/05 18:58:26]

does it make any sense to split this class out + unit-test-ify it?

[acolwell GONE FROM CHROMIUM, 2013/04/05 19:14:37]

I'm not sure it quite makes sense yet. I think ChunkDemuxer and SourceState are
still a little too coupled right now for us to have separate unit tests for
each. I expect further refactoring will make these 2 classes a little more
independent and I think I'll be able to refactor the tests then. 

+ public:
+  explicit SourceState(scoped_ptr<StreamParser> stream_parser);
+
+  void Init(const StreamParser::InitCB& init_cb,
+            const StreamParser::NewConfigCB& config_cb,
+            const StreamParser::NewBuffersCB& audio_cb,
+            const StreamParser::NewBuffersCB& video_cb,
+            const StreamParser::NeedKeyCB& need_key_cb,
+            const StreamParser::NewMediaSegmentCB& new_segment_cb,
+            const LogCB& log_cb);
+
+  // Appends new data to the StreamParser.
+  // Returns true if the data was successfully appended. Returns false if an
+  // error occurred.
+  bool Append(const uint8* data, size_t length);
+
+  // Aborts the current append sequence and resets the parser.
+  void Abort();
+
+  // Sets |timestamp_offset_| if possible.
+  // Returns if the offset was set. Returns false if the offset could not be
+  // updated at this time.
+  bool SetTimestampOffset(TimeDelta timestamp_offset);
+
+  TimeDelta timestamp_offset() const { return timestamp_offset_; }
+
+ private:
+  // Called by the |stream_parser_| at the beginning of a new media segment.
+  // |timestamp| is the timestamp on the first buffer in the segment.
+  // It modifies the state of this object and then calls |new_segment_cb| with
+  // modified version of |timestamp|.
+  void OnNewMediaSegment(const StreamParser::NewMediaSegmentCB& new_segment_cb,
+                         TimeDelta timestamp);
+
+  // Called by the |stream_parser_| at the end of a media segment.
+  void OnEndOfMediaSegment();
+
+  // Called by the |stream_parser_| when new buffers have been parsed. It
+  // applies |timestamp_offset_| to all buffers in |buffers| and then calls
+  // |new_buffers_cb| with the modified buffers.
+  // Returns true on a successful call. Returns false if an error occured while
+  // processing the buffers.
+  bool OnBuffers(const StreamParser::NewBuffersCB& new_buffers_cb,
+                 const StreamParser::BufferQueue& buffers);
+
+  // Helper function that adds |timestamp_offset_| to each buffer in |buffers|.
+  void AdjustBufferTimestamps(const StreamParser::BufferQueue& buffers);
+
+  // The offset to apply to media segment timestamps.
+  TimeDelta timestamp_offset_;
+
+  // Keeps track of whether |timestamp_offset_| can be modified.
+  bool can_update_offset_;
+
+  // The object used to parse appended data.
+  scoped_ptr<StreamParser> stream_parser_;
+
+  DISALLOW_COPY_AND_ASSIGN(SourceState);
+};
+
+SourceState::SourceState(scoped_ptr<StreamParser> stream_parser)
+    : can_update_offset_(true),
+      stream_parser_(stream_parser.release()) {
+}
+
+void SourceState::Init(const StreamParser::InitCB& init_cb,
+                       const StreamParser::NewConfigCB& config_cb,
+                       const StreamParser::NewBuffersCB& audio_cb,
+                       const StreamParser::NewBuffersCB& video_cb,
+                       const StreamParser::NeedKeyCB& need_key_cb,
+                       const StreamParser::NewMediaSegmentCB& new_segment_cb,
+                       const LogCB& log_cb) {
+  stream_parser_->Init(init_cb, config_cb,
+                       base::Bind(&SourceState::OnBuffers,
+                                  base::Unretained(this), audio_cb),
+                       base::Bind(&SourceState::OnBuffers,
+                                  base::Unretained(this), video_cb),
+                       need_key_cb,
+                       base::Bind(&SourceState::OnNewMediaSegment,
+                                  base::Unretained(this), new_segment_cb),
+                       base::Bind(&SourceState::OnEndOfMediaSegment,
+                                  base::Unretained(this)),
+                       log_cb);
+}
+
+bool SourceState::SetTimestampOffset(TimeDelta timestamp_offset) {
+  if (!can_update_offset_)
+    return false;
+
+  timestamp_offset_ = timestamp_offset;
+  return true;
+}
+
+bool SourceState::Append(const uint8* data, size_t length) {
+  return stream_parser_->Parse(data, length);
+}
+
+void SourceState::Abort() {
+  stream_parser_->Flush();
+  can_update_offset_ = true;
+}
+
+void SourceState::AdjustBufferTimestamps(
+    const StreamParser::BufferQueue& buffers) {
+  if (timestamp_offset_ == TimeDelta())
+    return;
+
+  for (StreamParser::BufferQueue::const_iterator itr = buffers.begin();
+       itr != buffers.end(); ++itr) {
+    (*itr)->SetDecodeTimestamp(
+        (*itr)->GetDecodeTimestamp() + timestamp_offset_);
+    (*itr)->SetTimestamp((*itr)->GetTimestamp() + timestamp_offset_);
+  }
+}
+
+void SourceState::OnNewMediaSegment(
+    const StreamParser::NewMediaSegmentCB& new_segment_cb,
+    TimeDelta timestamp) {
+  DCHECK(timestamp != kNoTimestamp());
+  DVLOG(2) << "OnNewMediaSegment(" << timestamp.InSecondsF() << ")";
+
+  can_update_offset_ = false;
+  new_segment_cb.Run(timestamp + timestamp_offset_);
+}
+
+void SourceState::OnEndOfMediaSegment() {
+  DVLOG(2) << "OnEndOfMediaSegment()";
+  can_update_offset_ = true;
+}
+
+bool SourceState::OnBuffers(const StreamParser::NewBuffersCB& new_buffers_cb,
+                            const StreamParser::BufferQueue& buffers) {
+  if (new_buffers_cb.is_null())
+    return false;
+
+  AdjustBufferTimestamps(buffers);
+
+  return new_buffers_cb.Run(buffers);
+}
 
 class ChunkDemuxerStream : public DemuxerStream {
  public:
   typedef std::deque<scoped_refptr<StreamParserBuffer> > BufferQueue;
   typedef std::deque<ReadCB> ReadCBQueue;
   typedef std::deque<base::Closure> ClosureQueue;
 
   ChunkDemuxerStream(const AudioDecoderConfig& audio_config,
                      const LogCB& log_cb);
   ChunkDemuxerStream(const VideoDecoderConfig& video_config,
@@ skipping 492 matching lines @@
   if (!cb.is_null())
     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) ||
-      stream_parser_map_.count(id) > 0u)
+  if ((state_ != WAITING_FOR_INIT && state_ != INITIALIZING) || IsValidId(id))
     return kReachedIdLimit;
 
   bool has_audio = false;
   bool has_video = false;
   scoped_ptr<media::StreamParser> stream_parser(
       StreamParserFactory::Create(type, codecs, log_cb_,
                                   &has_audio, &has_video));
 
   if (!stream_parser)
     return ChunkDemuxer::kNotSupported;
@@ skipping 10 matching lines @@
     audio_cb = base::Bind(&ChunkDemuxer::OnAudioBuffers,
                           base::Unretained(this));
   }
 
   if (has_video) {
     source_id_video_ = id;
     video_cb = base::Bind(&ChunkDemuxer::OnVideoBuffers,
                           base::Unretained(this));
   }
 
-  stream_parser->Init(
-      base::Bind(&ChunkDemuxer::OnStreamParserInitDone, base::Unretained(this)),
+  scoped_ptr<SourceState> source_state(new SourceState(stream_parser.Pass()));
+  source_state->Init(
+      base::Bind(&ChunkDemuxer::OnSourceInitDone, base::Unretained(this)),
       base::Bind(&ChunkDemuxer::OnNewConfigs, base::Unretained(this),
                  has_audio, has_video),
       audio_cb,
       video_cb,
       base::Bind(&ChunkDemuxer::OnNeedKey, base::Unretained(this)),
       base::Bind(&ChunkDemuxer::OnNewMediaSegment, base::Unretained(this), id),
-      base::Bind(&ChunkDemuxer::OnEndOfMediaSegment,
-                 base::Unretained(this), id),
       log_cb_);
 
-  stream_parser_map_[id] = stream_parser.release();
-  SourceInfo info = { base::TimeDelta(), true };
-  source_info_map_[id] = info;
-
+  source_state_map_[id] = source_state.release();
   return kOk;
 }
 
 void ChunkDemuxer::RemoveId(const std::string& id) {
   base::AutoLock auto_lock(lock_);
   CHECK(IsValidId(id));
 
-  delete stream_parser_map_[id];
-  stream_parser_map_.erase(id);
-  source_info_map_.erase(id);
+  delete source_state_map_[id];
+  source_state_map_.erase(id);
 
   if (source_id_audio_ == id) {
     if (audio_)
       audio_->Shutdown();
     source_id_audio_.clear();
   }
 
   if (source_id_video_ == id) {
     if (video_)
       video_->Shutdown();
@@ skipping 78 matching lines @@
     }
 
     if (length == 0u)
       return;
 
     DCHECK(data);
 
     switch (state_) {
       case INITIALIZING:
         DCHECK(IsValidId(id));
-        if (!stream_parser_map_[id]->Parse(data, length)) {
+        if (!source_state_map_[id]->Append(data, length)) {
           ReportError_Locked(DEMUXER_ERROR_COULD_NOT_OPEN);
           return;
         }
         break;
 
       case INITIALIZED: {
         DCHECK(IsValidId(id));
-        if (!stream_parser_map_[id]->Parse(data, length)) {
+        if (!source_state_map_[id]->Append(data, length)) {
           ReportError_Locked(PIPELINE_ERROR_DECODE);
           return;
         }
       } break;
 
       case PARSE_ERROR:
         DVLOG(1) << "AppendData(): Ignoring data after a parse error.";
         return;
 
       case WAITING_FOR_INIT:
@@ skipping 17 matching lines @@
 
   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));
-  stream_parser_map_[id]->Flush();
-  source_info_map_[id].can_update_offset = true;
+  source_state_map_[id]->Abort();
 }
 
 double ChunkDemuxer::GetDuration() {
   base::AutoLock auto_lock(lock_);
   return GetDuration_Locked();
 }
 
 double ChunkDemuxer::GetDuration_Locked() {
   lock_.AssertAcquired();
   if (duration_ == kNoTimestamp())
@@ skipping 50 matching lines @@
 
   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));
 
-  if (!source_info_map_[id].can_update_offset)
-    return false;
-
-  source_info_map_[id].timestamp_offset = offset;
-  return true;
+  return source_state_map_[id]->SetTimestampOffset(offset);
 }
 
 bool ChunkDemuxer::EndOfStream(PipelineStatus status) {
   DVLOG(1) << "EndOfStream(" << status << ")";
   base::AutoLock auto_lock(lock_);
   DCHECK_NE(state_, WAITING_FOR_INIT);
   DCHECK_NE(state_, ENDED);
 
   if (state_ == SHUTDOWN || state_ == PARSE_ERROR)
     return true;
@@ skipping 48 matching lines @@
 
 void ChunkDemuxer::ChangeState_Locked(State new_state) {
   lock_.AssertAcquired();
   DVLOG(1) << "ChunkDemuxer::ChangeState_Locked() : "
            << state_ << " -> " << new_state;
   state_ = new_state;
 }
 
 ChunkDemuxer::~ChunkDemuxer() {
   DCHECK_NE(state_, INITIALIZED);
-  for (StreamParserMap::iterator it = stream_parser_map_.begin();
-       it != stream_parser_map_.end(); ++it) {
+  for (SourceStateMap::iterator it = source_state_map_.begin();
+       it != source_state_map_.end(); ++it) {
     delete it->second;
   }
-  stream_parser_map_.clear();
+  source_state_map_.clear();
 }
 
 void ChunkDemuxer::ReportError_Locked(PipelineStatus error) {
   DVLOG(1) << "ReportError_Locked(" << error << ")";
   lock_.AssertAcquired();
   DCHECK_NE(error, PIPELINE_OK);
 
   ChangeState_Locked(PARSE_ERROR);
 
   PipelineStatusCB cb;
@@ skipping 33 matching lines @@
 
   return seek_pending;
 }
 
 bool ChunkDemuxer::CanEndOfStream_Locked() const {
   lock_.AssertAcquired();
   return (!audio_ || audio_->CanEndOfStream()) &&
          (!video_ || video_->CanEndOfStream());
 }
 
-void ChunkDemuxer::OnStreamParserInitDone(bool success, TimeDelta duration) {
-  DVLOG(1) << "OnStreamParserInitDone(" << success << ", "
+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())
     UpdateDuration(duration);
@@ skipping 76 matching lines @@
 }
 
 bool ChunkDemuxer::OnAudioBuffers(const StreamParser::BufferQueue& buffers) {
   lock_.AssertAcquired();
   DCHECK_NE(state_, SHUTDOWN);
 
   if (!audio_)
     return false;
 
   CHECK(IsValidId(source_id_audio_));
-  AdjustBufferTimestamps(
-      buffers, source_info_map_[source_id_audio_].timestamp_offset);
-
   if (!audio_->Append(buffers))
     return false;
 
   IncreaseDurationIfNecessary(buffers, audio_);
   return true;
 }
 
 bool ChunkDemuxer::OnVideoBuffers(const StreamParser::BufferQueue& buffers) {
   lock_.AssertAcquired();
   DCHECK_NE(state_, SHUTDOWN);
 
   if (!video_)
     return false;
 
   CHECK(IsValidId(source_id_video_));
-  AdjustBufferTimestamps(
-      buffers, source_info_map_[source_id_video_].timestamp_offset);
-
   if (!video_->Append(buffers))
     return false;
 
   IncreaseDurationIfNecessary(buffers, video_);
   return true;
 }
 
 // TODO(acolwell): Remove bool from StreamParser::NeedKeyCB so that
 // this method can be removed and need_key_cb_ can be passed directly
 // to the parser.
 bool ChunkDemuxer::OnNeedKey(const std::string& type,
                              scoped_array<uint8> init_data,
                              int init_data_size) {
   lock_.AssertAcquired();
   need_key_cb_.Run(type, init_data.Pass(), init_data_size);
   return true;
 }
 
 void ChunkDemuxer::OnNewMediaSegment(const std::string& source_id,
                                      TimeDelta timestamp) {
   DCHECK(timestamp != kNoTimestamp());
   DVLOG(2) << "OnNewMediaSegment(" << source_id << ", "
            << timestamp.InSecondsF() << ")";
   lock_.AssertAcquired();
 
   CHECK(IsValidId(source_id));
-  source_info_map_[source_id].can_update_offset = false;
-  base::TimeDelta start_timestamp =
-      timestamp + source_info_map_[source_id].timestamp_offset;
-
   if (audio_ && source_id == source_id_audio_)
-    audio_->OnNewMediaSegment(start_timestamp);
+    audio_->OnNewMediaSegment(timestamp);
   if (video_ && source_id == source_id_video_)
-    video_->OnNewMediaSegment(start_timestamp);
-}
-
-void ChunkDemuxer::OnEndOfMediaSegment(const std::string& source_id) {
-  DVLOG(2) << "OnEndOfMediaSegment(" << source_id << ")";
-  CHECK(IsValidId(source_id));
-  source_info_map_[source_id].can_update_offset = true;
-}
-
-void ChunkDemuxer::AdjustBufferTimestamps(
-    const StreamParser::BufferQueue& buffers,
-    base::TimeDelta timestamp_offset) {
-  if (timestamp_offset == base::TimeDelta())
-    return;
-
-  for (StreamParser::BufferQueue::const_iterator itr = buffers.begin();
-       itr != buffers.end(); ++itr) {
-    (*itr)->SetDecodeTimestamp(
-        (*itr)->GetDecodeTimestamp() + timestamp_offset);
-    (*itr)->SetTimestamp((*itr)->GetTimestamp() + timestamp_offset);
-  }
+    video_->OnNewMediaSegment(timestamp);
 }
 
 bool ChunkDemuxer::IsValidId(const std::string& source_id) const {
   lock_.AssertAcquired();
-  return source_info_map_.count(source_id) > 0u &&
-      stream_parser_map_.count(source_id) > 0u;
+  return source_state_map_.count(source_id) > 0u;
 }
 
 void ChunkDemuxer::UpdateDuration(base::TimeDelta new_duration) {
   DCHECK(duration_ != new_duration);
   user_specified_duration_ = -1;
   duration_ = new_duration;
   host_->SetDuration(new_duration);
 }
 
 void ChunkDemuxer::IncreaseDurationIfNecessary(
@@ skipping 23 matching lines @@
 
 Ranges<TimeDelta> ChunkDemuxer::GetBufferedRanges() const {
   if (audio_ && !video_)
     return audio_->GetBufferedRanges(duration_);
   else if (!audio_ && video_)
     return video_->GetBufferedRanges(duration_);
   return ComputeIntersection();
 }
 
 }  // namespace media