Adding ChunkDemuxer implementation.

media/filters/chunk_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.
+
+// Implements a Demuxer that can switch among different data sources mid-stream.
+// Uses FFmpegDemuxer under the covers, so see the caveats at the top of
+// ffmpeg_demuxer.h.
+
+#include "media/filters/chunk_demuxer.h"
+
+#include "base/bind.h"
+#include "base/logging.h"
+#include "base/message_loop.h"
+#include "media/base/filter_host.h"
+#include "media/base/data_buffer.h"
+#include "media/ffmpeg/ffmpeg_common.h"
+#include "media/filters/buffer_url_protocol.h"
+#include "media/filters/ffmpeg_glue.h"
+#include "media/webm/webm_constants.h"
+
+namespace media {
+
+// WebM File Header. This is prepended to the INFO & TRACKS
+// data passed to Init() before handing it to FFmpeg. Essentially
+// we are making the INFO & TRACKS data look like a small WebM
+// file so we can use FFmpeg to initialize the AVFormatContext.
+static const uint8 kWebMHeader[] = {

[scherkus (not reviewing), 2011/06/23 18:50:18]

should this go in media/webm or do we plan to delete this once we figure out how
to get rid of AVFormatContext

if so.. do we have a TODO for getting rid of the workaround?

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

Added a TODO. I think it should stay here because it is specific to this Demuxer
implementation.

+  0x1A, 0x45, 0xDF, 0xA3, 0x9F, // EBML (size = 0x1f)
+  0x42, 0x86, 0x81, 0x01, // EBMLVersion = 1
+  0x42, 0xF7, 0x81, 0x01, // EBMLReadVersion = 1
+  0x42, 0xF2, 0x81, 0x04, // EBMLMaxIDLength = 4
+  0x42, 0xF3, 0x81, 0x08, // EBMLMaxSizeLength = 8
+  0x42, 0x82, 0x84, 0x77, 0x65, 0x62, 0x6D, // DocType = "webm"
+  0x42, 0x87, 0x81, 0x02, // DocTypeVersion = 2
+  0x42, 0x85, 0x81, 0x02, // DocTypeReadVersion = 2
+  // EBML end
+  0x18, 0x53, 0x80, 0x67, // Segment
+  0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // segment(size = 0)
+  // INFO goes here.
+};
+
+// Offset of the segment size field in kWebMHeader. Used to update
+// the segment size field before handing the buffer to FFmpeg.
+static const int kSegmentSizeOffset = sizeof(kWebMHeader) - 8;
+
+static const uint8 kEmptyCluster[] = {
+  0x1F, 0x43, 0xB6, 0x75, 0x80 // CLUSTER (size = 0)
+};
+
+MediaDataSink::MediaDataSink(const scoped_refptr<ChunkDemuxer>& demuxer)
+    : demuxer_(demuxer) {
+}
+
+MediaDataSink::~MediaDataSink() {}
+
+void MediaDataSink::Flush() {
+  demuxer_->FlushData();
+}
+
+bool MediaDataSink::AddData(const uint8* data, unsigned length) {
+  return demuxer_->AddData(data, length);
+}
+
+void MediaDataSink::Shutdown() {
+  demuxer_->Shutdown();
+}
+
+class ChunkDemuxerStream : public DemuxerStream {
+ public:
+  typedef std::deque<scoped_refptr<Buffer> > BufferQueue;
+  typedef std::deque<ReadCallback> ReadCBQueue;
+
+  ChunkDemuxerStream(Type type, AVStream* stream);
+  virtual ~ChunkDemuxerStream();
+
+  void Flush();
+  void AddBuffers(const BufferQueue& buffers);
+  void Shutdown();
+
+  bool GetLastBufferTimestamp(base::TimeDelta* timestamp) const;
+
+  // DemuxerStream methods.
+  virtual void Read(const ReadCallback& read_callback);
+  virtual Type type();
+  virtual const MediaFormat& media_format();
+  virtual void EnableBitstreamConverter();
+  virtual AVStream* GetAVStream();
+
+ private:
+  static void RunCallback(ReadCallback cb, scoped_refptr<Buffer> buffer);
+
+  Type type_;
+  MediaFormat media_format_;
+  AVStream* av_stream_;
+
+  mutable base::Lock lock_;
+  ReadCBQueue read_cbs_;
+  BufferQueue buffers_;
+  bool shutdown_called_;
+
+  DISALLOW_IMPLICIT_CONSTRUCTORS(ChunkDemuxerStream);
+};
+
+// Helper class used to parse WebM clusters.
+class ClusterWebMParserClient : public WebMParserClient {

[scherkus (not reviewing), 2011/06/23 18:50:18]

could this also move to media/webm ?

this class could simply queue all of the buffers as member variables and then
after a successful parse have the client transfer the buffers to the
ChunkDemuxer buffer queue

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

The implementation is specific to this demuxer. It only allows 2 streams &
errors on any elements other than simple blocks and cluster timecode. There are
other WebM elements that can appear in a cluster, but we don't need to support
them for this use case.
This class already does this.

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

Ahh sorry I had read the code using ChunkDemuxerStream::BufferQueue as a
dependency on ChunkDemuxerStream class + was thinking of a way to break that
dependency

+ public:
+  ClusterWebMParserClient(int64 timecode_scale,
+                          int audio_track_num,
+                          int audio_default_duration,
+                          int video_track_num,
+                          int video_default_duration)
+      : timecode_multiplier_(timecode_scale / 1000.0), // will convert timecodes

[scherkus (not reviewing), 2011/06/23 18:50:18]

what about putting this comment in the .h file?

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

This class doesn't appear in the .h file. I've moved the comment down to the
timecode_multiplier_ declaration though.

+                                                       // to microseconds.
+        audio_track_num_(audio_track_num),
+        audio_default_duration_(audio_default_duration),
+        video_track_num_(video_track_num),
+        video_default_duration_(video_default_duration),
+        cluster_timecode_(-1) {
+    VLOG(1) << "tc_s " << timecode_multiplier_
+               << " atn " << audio_track_num_

[scherkus (not reviewing), 2011/06/23 18:50:18]

indentation

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

Done.

+               << " add " << audio_default_duration_
+               << " vtn " << video_track_num_
+               << " vdd " << video_default_duration_;
+  }
+
+  const ChunkDemuxerStream::BufferQueue& audio_buffers() const {
+    return audio_buffers_;
+  };
+
+  const ChunkDemuxerStream::BufferQueue& video_buffers() const {
+    return video_buffers_;
+  };
+
+  virtual bool OnListStart(int id) {
+    if (id == kWebMIdCluster)
+      cluster_timecode_ = -1;
+
+    return true;
+  }
+
+  virtual bool OnListEnd(int id) {
+    if (id == kWebMIdCluster)
+      cluster_timecode_ = -1;
+
+    return true;
+  }
+
+  virtual bool OnUInt(int id, int64 val) {
+    if (id == kWebMIdTimecode) {
+      if (cluster_timecode_ != -1)
+        return false;
+
+      cluster_timecode_ = val;
+    }
+
+    return true;
+  }
+
+  virtual bool OnFloat(int id, double val) {
+    VLOG(1) << "Unexpected float element with ID " << std::hex << id;
+    return false;
+  }
+
+  virtual bool OnBinary(int id, const uint8* data, int size) {
+    VLOG(1) << "Unexpected binary element with ID " << std::hex << id;
+    return false;
+  }
+
+  virtual bool OnString(int id, const std::string& str) {
+    VLOG(1) << "Unexpected string element with ID " << std::hex << id;
+    return false;
+  }
+
+  virtual bool OnSimpleBlock(int track_num, int timecode,
+                             int flags,
+                             const uint8* data, int size) {
+    if (cluster_timecode_ == -1) {
+      VLOG(1) << "Got SimpleBlock before cluster timecode.";
+      return false;
+    }
+
+    //VLOG(1) << "SimpleBlock : tn " << track_num
+    //           << " tc " << (cluster_timecode_ + timecode)

[scherkus (not reviewing), 2011/06/23 18:50:18]

?

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

removed

+    //           << "(" << timecode << ")"
+    //           << " flags " << std::hex << flags << std::dec
+    //           << " size " << size;
+
+    base::TimeDelta timestamp = base::TimeDelta::FromMicroseconds(
+        (cluster_timecode_ + timecode) * timecode_multiplier_);
+
+    scoped_refptr<DataBuffer> buffer(new DataBuffer(const_cast<uint8*>(data),
+                                                    size, true));
+    buffer->SetTimestamp(timestamp);
+
+    if (track_num == audio_track_num_) {
+      base::TimeDelta audio_duration = base::TimeDelta::FromMicroseconds(
+          audio_default_duration_ / 1000.0);
+      buffer->SetDuration(audio_duration);
+      audio_buffers_.push_back(buffer);
+      return true;
+    }
+
+    if (track_num == video_track_num_) {
+      base::TimeDelta video_duration = base::TimeDelta::FromMicroseconds(
+          video_default_duration_ / 1000.0);
+      buffer->SetDuration(video_duration);
+      video_buffers_.push_back(buffer);
+      return true;
+    }
+
+    VLOG(1) << "Unexpected track number " << track_num;
+    return false;
+  }
+
+ private:
+  double timecode_multiplier_;
+  int audio_track_num_;
+  int audio_default_duration_;
+  int video_track_num_;

[scherkus (not reviewing), 2011/06/23 18:50:18]

ditto for TimeDelta here

we should do the conversion during the parse instead of forcing all clients to
do the conversion

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

Done.

+  int video_default_duration_;
+
+  int64 cluster_timecode_;
+
+  ChunkDemuxerStream::BufferQueue audio_buffers_;
+  ChunkDemuxerStream::BufferQueue video_buffers_;
+};
+
+ChunkDemuxerStream::ChunkDemuxerStream(Type type, AVStream* stream)
+    : type_(type),
+      av_stream_(stream),
+      shutdown_called_(false) {
+}
+
+ChunkDemuxerStream::~ChunkDemuxerStream() {}
+
+void ChunkDemuxerStream::Flush() {
+  VLOG(1) << "Flush()";
+  base::AutoLock auto_lock(lock_);
+  buffers_.clear();
+}
+
+void ChunkDemuxerStream::AddBuffers(const BufferQueue& buffers) {
+  std::deque<base::Closure> callbacks;
+  {
+    base::AutoLock auto_lock(lock_);
+
+    for (BufferQueue::const_iterator itr = buffers.begin();
+         itr != buffers.end(); itr++) {
+      buffers_.push_back(*itr);
+    }
+
+    while (!buffers_.empty() && !read_cbs_.empty()) {
+      callbacks.push_back(base::Bind(&ChunkDemuxerStream::RunCallback,
+                                     read_cbs_.front(),
+                                     buffers_.front()));
+      buffers_.pop_front();
+      read_cbs_.pop_front();
+    }
+
+    if (!buffers_.empty()) {
+      base::TimeDelta fts = buffers_.front()->GetTimestamp();

[scherkus (not reviewing), 2011/06/23 18:50:18]

holy hard to understand variables!!

why not "front_ts", "back_ts" and "delta_ts"

...also if this is only for logging purposes do we really need to kepe this code
around?

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

removed

+      base::TimeDelta bts = buffers_.back()->GetTimestamp();
+      base::TimeDelta dlt = bts - fts;
+      VLOG(1) << "AddBuffers() : ct " << buffers_.size()
+                 << " fts " << fts.InSecondsF()
+                 << " bts " << bts.InSecondsF()
+                 << " dlt "  << dlt.InSecondsF();
+    }
+  }
+
+  while (!callbacks.empty()) {
+    callbacks.front().Run();
+    callbacks.pop_front();
+  }
+}
+
+void ChunkDemuxerStream::Shutdown() {
+  std::deque<ReadCallback> callbacks;
+  {
+    base::AutoLock auto_lock(lock_);
+    shutdown_called_ = true;
+
+    // Collect all the pending Read() callbacks.
+    while (!read_cbs_.empty()) {
+      callbacks.push_back(read_cbs_.front());
+      read_cbs_.pop_front();
+    }
+  }
+
+  // Pass NULL to all callbacks to signify read failure.
+  while (!callbacks.empty()) {
+    callbacks.front().Run(NULL);
+    callbacks.pop_front();
+  }
+}
+
+bool ChunkDemuxerStream::GetLastBufferTimestamp(
+    base::TimeDelta* timestamp) const {
+  base::AutoLock auto_lock(lock_);
+
+  if (buffers_.empty())
+    return false;
+
+  *timestamp = buffers_.back()->GetTimestamp();
+  return true;
+}
+
+// Helper function used to make Closures for ReadCallbacks.
+//static
+void ChunkDemuxerStream::RunCallback(ReadCallback cb,
+                                     scoped_refptr<Buffer> buffer) {
+  cb.Run(buffer);
+}
+
+// Helper function that makes sure |read_callback| runs on |message_loop|.
+static void RunOnMessageLoop(const DemuxerStream::ReadCallback& read_callback,
+                             MessageLoop* message_loop,
+                             scoped_refptr<Buffer> buffer) {
+  if (MessageLoop::current() != message_loop) {
+    message_loop->PostTask(FROM_HERE,
+                           NewRunnableFunction(&RunOnMessageLoop,
+                                               read_callback,
+                                               message_loop,
+                                               buffer));
+    return;
+  }
+
+  read_callback.Run(buffer);
+}
+
+// DemuxerStream methods.
+void ChunkDemuxerStream::Read(const ReadCallback& read_callback) {
+

[scherkus (not reviewing), 2011/06/23 18:50:18]

nit: get rid of blank line

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

Done.

+  scoped_refptr<Buffer> buffer;
+
+  {
+    base::AutoLock auto_lock(lock_);
+
+    if (!shutdown_called_) {
+      if (buffers_.empty()) {
+        // Wrap & store |read_callback| so that it will
+        // get called on the current MessageLoop.
+        read_cbs_.push_back(base::Bind(&RunOnMessageLoop,
+                                       read_callback,
+                                       MessageLoop::current()));
+        return;
+      }
+
+      if (!read_cbs_.empty()) {
+        // Wrap & store |read_callback| so that it will
+        // get called on the current MessageLoop.
+        read_cbs_.push_back(base::Bind(&RunOnMessageLoop,
+                                       read_callback,
+                                       MessageLoop::current()));
+        return;
+      }
+
+      buffer = buffers_.front();
+      buffers_.pop_front();
+    }
+  }
+
+  read_callback.Run(buffer);
+}
+
+DemuxerStream::Type ChunkDemuxerStream::type() { return type_; }
+
+const MediaFormat& ChunkDemuxerStream::media_format() { return media_format_; }
+
+void ChunkDemuxerStream::EnableBitstreamConverter() {}
+
+AVStream* ChunkDemuxerStream::GetAVStream() { return av_stream_; }
+
+ChunkDemuxer::ChunkDemuxer()
+    : state_(WAITING_FOR_INIT),
+      format_context_(NULL),
+      buffered_bytes_(0),
+      first_seek_(true) {
+}
+
+ChunkDemuxer::~ChunkDemuxer() {
+  DCHECK_NE(state_, INITIALIZED);
+
+  if (!format_context_)
+    return;
+
+  DestroyAVFormatContext(format_context_);
+  format_context_ = NULL;
+}
+
+bool ChunkDemuxer::Init(const uint8* data, int size) {
+  DCHECK(data);
+  DCHECK_GT(size, 0);
+  VLOG(1) << "Init(" << size << ")";
+
+  base::AutoLock auto_lock(lock_);
+  DCHECK_EQ(state_, WAITING_FOR_INIT);
+
+  int res = ParseWebMHeaders(&info_, data, size);
+
+  if (res < 0) {
+    ChangeState(INIT_ERROR);
+    return false;
+  }
+
+  format_context_ = CreateFormatContext(data, size);
+
+  if (!format_context_ || !SetupStreams() || !ParsePendingBuffers()) {
+    ChangeState(INIT_ERROR);
+    return false;
+  }
+
+  ChangeState(INITIALIZED);
+  return true;
+}
+
+// Filter implementation.
+void ChunkDemuxer::set_host(FilterHost* filter_host) {
+  Demuxer::set_host(filter_host);
+  double mult = info_.timecode_scale() / 1000.0;
+
+  base::TimeDelta duration =
+      base::TimeDelta::FromMicroseconds(info_.duration() * mult);
+
+  filter_host->SetDuration(duration);
+  filter_host->SetCurrentReadPosition(0);
+}
+
+void ChunkDemuxer::Stop(FilterCallback* callback) {
+  VLOG(1) << "Stop()";
+
+  callback->Run();
+  delete callback;
+}
+
+void ChunkDemuxer::Seek(base::TimeDelta time, const FilterStatusCB&  cb) {
+  VLOG(1) << "Seek(" << time.InSecondsF() << ")";
+
+  bool run_callback = false;
+  {
+    base::AutoLock auto_lock(lock_);
+
+    if (first_seek_) {
+      first_seek_ = false;
+      run_callback = true;
+    } else {
+      seek_cb_ = cb;
+      seek_time_ = time;
+    }
+  }
+
+  if (run_callback)
+    cb.Run(PIPELINE_OK);
+}
+
+void ChunkDemuxer::OnAudioRendererDisabled() {
+  base::AutoLock auto_lock(lock_);
+  audio_ = NULL;
+}
+
+void ChunkDemuxer::SetPreload(Preload preload) {}
+
+// Demuxer implementation.
+scoped_refptr<DemuxerStream> ChunkDemuxer::GetStream(
+    DemuxerStream::Type type) {
+  VLOG(1) << "GetStream(" << type << ")";
+
+  if (type == DemuxerStream::VIDEO)
+    return video_;
+
+  if (type == DemuxerStream::AUDIO)
+    return audio_;
+
+  return NULL;
+}
+
+base::TimeDelta ChunkDemuxer::GetStartTime() const {
+  VLOG(1) << "GetStartTime()";
+  // TODO(acolwell) : Fix this so it uses the time on the first packet.
+  return base::TimeDelta();
+}
+
+void ChunkDemuxer::FlushData() {
+  base::AutoLock auto_lock(lock_);
+  if (audio_.get())
+    audio_->Flush();
+
+  if (video_.get())
+    video_->Flush();
+
+  pending_buffers_.clear();
+}
+
+bool ChunkDemuxer::AddData(const uint8* data, unsigned length) {
+  VLOG(1) << "AddData(" << length << ")";
+
+  int64 buffered_bytes = 0;
+  base::TimeDelta buffered_ts = base::TimeDelta::FromSeconds(-1);
+
+  FilterStatusCB cb;
+  {
+    base::AutoLock auto_lock(lock_);
+
+    switch(state_) {
+      case WAITING_FOR_INIT:
+        pending_buffers_.push_back(new DataBuffer(const_cast<uint8*>(data),
+                                                  length, true));
+        return false;
+        break;
+
+      case INITIALIZED:
+        if (!ParseAndAddData_Locked(data, length)) {
+          VLOG(1) << "AddData(): parsing data failed";
+          return false;
+        }
+        break;
+
+      case INIT_ERROR:
+      case SHUTDOWN:
+        VLOG(1) << "AddData(): called in unexpected state " << state_;
+        return false;
+        break;
+    }
+
+    base::TimeDelta tmp;
+    if (audio_.get() && audio_->GetLastBufferTimestamp(&tmp) &&
+        tmp > buffered_ts) {
+      buffered_ts = tmp;
+    }
+
+    if (video_.get() && video_->GetLastBufferTimestamp(&tmp) &&
+        tmp > buffered_ts) {
+      buffered_ts = tmp;
+    }
+
+    buffered_bytes = buffered_bytes_;
+
+    if (!seek_cb_.is_null())
+      std::swap(cb, seek_cb_);
+  }
+
+  // Notify the host of 'network activity' because we got data.
+  if (host()) {
+    host()->SetBufferedBytes(buffered_bytes);
+
+    if (buffered_ts.InSeconds() >= 0) {
+      host()->SetBufferedTime(buffered_ts);
+    }
+
+    host()->SetNetworkActivity(true);
+  }
+
+  if (!cb.is_null())
+    cb.Run(PIPELINE_OK);
+
+  return true;
+}
+
+void ChunkDemuxer::Shutdown() {
+  FilterStatusCB cb;
+  {
+    base::AutoLock auto_lock(lock_);
+
+    std::swap(cb, seek_cb_);
+
+    if (audio_.get())
+      audio_->Shutdown();
+
+    if (video_.get())
+      video_->Shutdown();
+
+    ChangeState(SHUTDOWN);
+  }
+
+  if (!cb.is_null())
+    cb.Run(PIPELINE_OK);
+}
+
+void ChunkDemuxer::ChangeState(State new_state) {

[scherkus (not reviewing), 2011/06/23 18:50:18]

nit: should this be ChangeState_Locked and/or have a lock_.AssertAcquired ?

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

Done.

+  state_ = new_state;
+}
+
+AVFormatContext* ChunkDemuxer::CreateFormatContext(const uint8* data,
+                                                   int size) const {
+  int segment_size = size + sizeof(kEmptyCluster);
+  int buf_size = sizeof(kWebMHeader) + segment_size;
+  scoped_array<uint8> buf(new uint8[buf_size]);
+  memcpy(buf.get(), kWebMHeader, sizeof(kWebMHeader));
+  memcpy(buf.get() + sizeof(kWebMHeader), data, size);
+  memcpy(buf.get() + sizeof(kWebMHeader) + size, kEmptyCluster,
+         sizeof(kEmptyCluster));
+
+  // Update the segment size in the buffer.
+  int64 tmp = (segment_size & GG_LONGLONG(0x00FFFFFFFFFFFFFF)) |
+      GG_LONGLONG(0x0100000000000000);
+  for (int i = 0; i < 8; i++) {
+    buf[kSegmentSizeOffset + i] = (tmp >> (8 * (7 - i))) & 0xff;
+  }
+
+  BufferUrlProtocol bup(buf.get(), buf_size, true);
+  std::string key = FFmpegGlue::GetInstance()->AddProtocol(&bup);
+
+  // Open FFmpeg AVFormatContext.
+  AVFormatContext* context = NULL;
+  int result = av_open_input_file(&context, key.c_str(), NULL, 0, NULL);
+
+  // Remove ourself from protocol list.
+  FFmpegGlue::GetInstance()->RemoveProtocol(&bup);
+
+  if (result < 0)
+    return NULL;
+
+  return context;
+}
+
+bool ChunkDemuxer::SetupStreams() {
+  int result = av_find_stream_info(format_context_);
+
+  if (result < 0)
+    return false;
+
+  bool no_supported_streams = true;
+  for (size_t i = 0; i < format_context_->nb_streams; ++i) {
+    AVStream* stream = format_context_->streams[i];
+    AVCodecContext* codec_context = stream->codec;
+    CodecType codec_type = codec_context->codec_type;
+
+    if (codec_type == CODEC_TYPE_AUDIO &&
+        stream->codec->codec_id == CODEC_ID_VORBIS &&
+        !audio_.get()) {
+      audio_ = new ChunkDemuxerStream(DemuxerStream::AUDIO, stream);
+      no_supported_streams = false;
+      continue;
+    }
+
+    if (codec_type == CODEC_TYPE_VIDEO &&
+        stream->codec->codec_id == CODEC_ID_VP8 &&
+        !video_.get()) {
+      video_ = new ChunkDemuxerStream(DemuxerStream::VIDEO, stream);
+      no_supported_streams = false;
+      continue;
+    }
+  }
+
+  return !no_supported_streams;
+}
+
+bool ChunkDemuxer::ParsePendingBuffers() {
+  // Handle any buffers that came in between the time the pipeline was
+  // started and Init() was called.
+  while(!pending_buffers_.empty()) {
+    scoped_refptr<media::Buffer> buf = pending_buffers_.front();
+    pending_buffers_.pop_front();
+
+    if (!ParseAndAddData_Locked(buf->GetData(), buf->GetDataSize())) {
+      pending_buffers_.clear();
+      ChangeState(INIT_ERROR);
+      return false;
+    }
+  }
+
+  return true;
+}
+
+bool ChunkDemuxer::ParseAndAddData_Locked(const uint8* data, int length) {
+  ClusterWebMParserClient client(info_.timecode_scale(),
+                                 info_.audio_track_num(),
+                                 info_.audio_default_duration(),
+                                 info_.video_track_num(),
+                                 info_.video_default_duration());
+  int res = ParseWebMCluster(&client, data, length);
+
+  if (res <= 0)
+    return false;
+
+  if (audio_.get())
+    audio_->AddBuffers(client.audio_buffers());
+
+  if (video_.get())
+    video_->AddBuffers(client.video_buffers());
+
+  // TODO(acolwell) : make this more representative of what is actually
+  // buffered.
+  buffered_bytes_ += length;
+
+  return true;
+}
+
+ChunkDemuxer::InfoTrackWebMParserClient::InfoTrackWebMParserClient()
+    : timecode_scale_(-1),
+      duration_(-1),
+      track_type_(-1),
+      track_num_(-1),
+      track_default_duration_(-1),
+      audio_track_num_(-1),
+      audio_default_duration_(-1),
+      video_track_num_(-1),
+      video_default_duration_(-1) {
+}
+
+int64 ChunkDemuxer::InfoTrackWebMParserClient::timecode_scale() const {
+  return timecode_scale_;
+}
+
+double ChunkDemuxer::InfoTrackWebMParserClient::duration() const {
+  return duration_;
+}
+
+int64 ChunkDemuxer::InfoTrackWebMParserClient::audio_track_num() const {
+  return audio_track_num_;
+}
+
+int64 ChunkDemuxer::InfoTrackWebMParserClient::audio_default_duration() const {
+  return audio_default_duration_;
+}
+
+int64 ChunkDemuxer::InfoTrackWebMParserClient::video_track_num() const {
+  return video_track_num_;
+}
+
+int64 ChunkDemuxer::InfoTrackWebMParserClient::video_default_duration() const {
+  return video_default_duration_;
+}
+
+bool ChunkDemuxer::InfoTrackWebMParserClient::OnListStart(int id) {
+  if (id == kWebMIdTrackEntry) {
+    track_type_ = -1;
+    track_num_ = -1;
+  }
+
+  return true;
+}
+
+bool ChunkDemuxer::InfoTrackWebMParserClient::OnListEnd(int id) {
+  if (id == kWebMIdTrackEntry) {
+    if (track_type_ == -1 || track_num_ == -1) {
+      VLOG(1) << "Missing TrackEntry data"
+                 << " TrackType " << track_type_
+                 << " TrackNum " << track_num_;
+      return false;
+    }
+
+    if (track_type_ == 1) {

[scherkus (not reviewing), 2011/06/23 18:50:18]

can these be webm_constants.h somehow?

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

Done.

+      video_track_num_ = track_num_;
+      video_default_duration_ = track_default_duration_;
+    } else if (track_type_ == 2) {
+      audio_track_num_ = track_num_;
+      audio_default_duration_ = track_default_duration_;
+    } else {
+      VLOG(1) << "Unexpected TrackType " << track_type_;
+      return false;
+    }
+
+    track_type_ = -1;
+    track_num_ = -1;
+  } else if (id == kWebMIdInfo && timecode_scale_ == -1) {
+    // Set timecode scale to default value if it isn't present in
+    // the Info element.
+    timecode_scale_ = 1000000;

[scherkus (not reviewing), 2011/06/23 18:50:18]

constant?

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

Done.

+  }
+  return true;
+}
+
+bool ChunkDemuxer::InfoTrackWebMParserClient::OnUInt(int id, int64 val) {
+  int64* dst = NULL;
+
+  switch (id) {
+    case kWebMIdTimecodeScale:
+      dst = &timecode_scale_;
+      break;
+    case kWebMIdTrackNumber:
+      dst = &track_num_;
+      break;
+    case kWebMIdTrackType:
+      dst = &track_type_;
+      break;
+    case  kWebMIdDefaultDuration:
+      dst = &track_default_duration_;
+      break;
+    default:
+      return true;
+  }
+
+  if (*dst != -1) {
+    VLOG(1) << "Multiple values for id " << std::hex << id << " specified";
+    return false;
+  }
+
+  *dst = val;
+  return true;
+}
+
+bool ChunkDemuxer::InfoTrackWebMParserClient::OnFloat(int id, double val) {
+  if (id != kWebMIdDuration) {
+    VLOG(1) << "Unexpected float for id" << std::hex << id;
+    return false;
+  }
+
+  if (duration_ != -1) {
+    VLOG(1) << "Multiple values for duration.";
+    return false;
+  }
+
+  duration_ = val;
+  return true;
+}
+
+bool ChunkDemuxer::InfoTrackWebMParserClient::OnBinary(int id, const
+                                                       uint8* data,
+                                                       int size) {
+  return true;
+}
+
+bool ChunkDemuxer::InfoTrackWebMParserClient::OnString(int id,
+                                                       const std::string& str) {
+  if (id != kWebMIdCodecID)
+    return false;
+
+  if (str != "A_VORBIS" && str != "V_VP8") {
+    VLOG(1) << "Unexpected CodecID " << str;
+    return false;
+  }
+
+  return true;
+}
+
+bool ChunkDemuxer::InfoTrackWebMParserClient::OnSimpleBlock(
+    int track_num, int timecode, int flags, const uint8* data, int size) {
+  return false;
+}
+
+const char ChunkDemuxerFactory::kURLPrefix[] = "x-media-chunks:";
+
+class ChunkDemuxerFactory::BuildState
+    : public base::RefCountedThreadSafe<BuildState> {
+ public:
+  static const int64 kMaxInfoSize = 32678;
+
+  BuildState(const std::string& url, BuildCallback* cb,
+             const scoped_refptr<ChunkDemuxer>& demuxer)
+      : url_(url),
+        cb_(cb),
+        demuxer_(demuxer),
+        read_buffer_(NULL),
+        read_size_(0),
+        message_loop_(MessageLoop::current()) {
+    AddRef();

[scherkus (not reviewing), 2011/06/23 18:50:18]

I'm guessing we have to do this because BuildState lives its life on various
message loops?

:(

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

This is needed because BuildState manages it's own lifetime. The factory doesn't
keep track of the object once it has been created. BuildState is responsible for
deleting itself once the build completes.

+  }
+
+  ~BuildState() { delete read_buffer_; }

[scherkus (not reviewing), 2011/06/23 18:50:18]

virtual

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

Done.

+
+  void BuildDone(PipelineStatus status, DataSource* data_source) {
+    if (message_loop_ != MessageLoop::current()) {

[scherkus (not reviewing), 2011/06/23 18:50:18]

as I've been discussing w/ other folks, this pattern has to die

Sure it saves a few LOC but it's rather maddening to outside viewers (not to
mention makes debugging a bit painful).  Instead:
  1) BuildDone should always post a task to message_loop_
  2) You can name the other function BuildDoneOnDemuxerThread, or something to
your liking (DoBuildDone ?)

BuildDoneTask is something I helped propagate but is a really poor name for
tasks since it reads as a noun as opposed to a verb.

The reasoning for (1) is that if/when this code get refactored / moved to
different threads, we don't want to have completely different run-time
characteristics for the callee executing the callback depending on the executing
thread.

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

Done. I wish there was a better way to do this. It is annoying having to create
2 methods all the time for this. I wish there was some variant of NewCallback()
or base::Bind() that allowed you to specify a MessageLoop along with the
function signature and it would auto generate the PostTask() method.

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

In Chrome land I believe the solution is to essentially hardcode the threads
things are supposed to be run on.

So you see code where it DCHECK()s that certain things are on IO/UI threads, and
then callees post directly to those threads.

In media code we inject/abstract the thread that it runs on, which means we
can't use a similar trick and instead would have to carry around the
message_loop we want to have callees execute callbacks on.

...but yes I agree :)

+      message_loop_->PostTask(
+          FROM_HERE,
+          NewRunnableMethod(this,
+                            &BuildState::BuildDone,
+                            status,
+                            scoped_refptr<DataSource>(data_source)));
+      return;
+    }
+
+    if (status != PIPELINE_OK) {
+      cb_->Run(status, static_cast<Demuxer*>(NULL));
+      Release();
+      return;
+    }
+
+    data_source_ = data_source;
+
+    int64 size = 0;
+
+    if (!data_source_->GetSize(&size) || size >= kMaxInfoSize) {
+      cb_->Run(DEMUXER_ERROR_COULD_NOT_OPEN,
+               static_cast<Demuxer*>(NULL));
+      data_source_->Stop(NewCallback(this, &BuildState::StopDone));
+      return;
+    }
+
+    DCHECK(!read_buffer_);
+    read_size_ = size;
+    read_buffer_ = new uint8[read_size_];

[scherkus (not reviewing), 2011/06/23 18:50:18]

who deallocates this?

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

It was deleted in the destructor, but I've converted this to a
scoped_array<uint8> so I don't have to explicitly delete it.

+    data_source_->Read(0, read_size_, read_buffer_,
+                       NewCallback(this, &BuildState::ReadDone));
+  }
+
+  void ReadDone(size_t size) {
+    if (message_loop_ != MessageLoop::current()) {

[scherkus (not reviewing), 2011/06/23 18:50:18]

ditto

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

Done.

+      message_loop_->PostTask(FROM_HERE,
+                              NewRunnableMethod(this,
+                                                &BuildState::ReadDone,
+                                                size));
+      return;
+    }
+
+    if (size == DataSource::kReadError) {
+      cb_->Run(PIPELINE_ERROR_READ, static_cast<Demuxer*>(NULL));
+      data_source_->Stop(NewCallback(this, &BuildState::StopDone));
+      return;
+    }
+
+    if (demuxer_->Init(read_buffer_, size)) {
+      cb_->Run(PIPELINE_OK, demuxer_.get());
+    } else {
+      cb_->Run(DEMUXER_ERROR_COULD_NOT_OPEN, static_cast<Demuxer*>(NULL));
+    }
+
+    data_source_->Stop(NewCallback(this, &BuildState::StopDone));
+  }
+
+  void StopDone() { Release(); }
+
+ private:
+  std::string url_;
+  scoped_ptr<BuildCallback> cb_;
+  scoped_refptr<ChunkDemuxer> demuxer_;
+
+  scoped_refptr<DataSource> data_source_;
+  uint8* read_buffer_;
+  int read_size_;
+  MessageLoop* message_loop_;
+};

[scherkus (not reviewing), 2011/06/23 18:50:18]

DISALLOW etc

[acolwell GONE FROM CHROMIUM, 2011/06/23 22:42:17]

Done.

+
+ChunkDemuxerFactory::ChunkDemuxerFactory(
+    DataSourceFactory* data_source_factory)
+    : data_source_factory_(data_source_factory) {
+}
+
+ChunkDemuxerFactory::~ChunkDemuxerFactory() {}
+
+bool ChunkDemuxerFactory::IsUrlSupported(const std::string& url) const {
+  return (url.find(kURLPrefix) == 0);
+}
+
+MediaDataSink* ChunkDemuxerFactory::CreateMediaDataSink() {
+  demuxer_ = new ChunkDemuxer();
+  return new MediaDataSink(demuxer_);
+}
+
+void ChunkDemuxerFactory::Build(const std::string& url, BuildCallback* cb) {
+  if (!IsUrlSupported(url) || !demuxer_.get()) {
+    cb->Run(DEMUXER_ERROR_COULD_NOT_OPEN, static_cast<Demuxer*>(NULL));
+    delete cb;
+    return;
+  }
+
+  std::string info_url = url.substr(strlen(kURLPrefix));
+
+  data_source_factory_->Build(
+      info_url,
+      NewCallback(new BuildState(info_url, cb, demuxer_),
+                  &ChunkDemuxerFactory::BuildState::BuildDone));
+  demuxer_ = NULL;
+}
+
+DemuxerFactory* ChunkDemuxerFactory::Clone() const {
+  return new ChunkDemuxerFactory(data_source_factory_->Clone());
+}
+
+}  // namespace media