media/filters/chunk_demuxer.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "media/filters/chunk_demuxer.h"
   6
   7 #include "base/bind.h"
   8 #include "base/logging.h"
   9 #include "base/message_loop.h"
  10 #include "media/base/audio_decoder_config.h"
  11 #include "media/base/stream_parser_buffer.h"
  12 #include "media/base/video_decoder_config.h"
  13 #include "media/filters/chunk_demuxer_client.h"
  14 #include "media/webm/webm_stream_parser.h"
  15
  16 namespace media {
  17
  18 struct CodecInfo {
  19   const char* name;
  20   DemuxerStream::Type type;
  21 };
  22
  23 typedef StreamParser* (*ParserFactoryFunction)();
  24
  25 struct SupportedTypeInfo {
  26   const char* type;
  27   const ParserFactoryFunction factory_function;
  28   const CodecInfo** codecs;
  29 };
  30
  31 static const CodecInfo kVP8CodecInfo = { "vp8", DemuxerStream::VIDEO };
  32 static const CodecInfo kVorbisCodecInfo = { "vorbis", DemuxerStream::AUDIO };
  33
  34 static const CodecInfo* kVideoWebMCodecs[] = {
  35   &kVP8CodecInfo,
  36   &kVorbisCodecInfo,
  37   NULL
  38 };
  39
  40 static const CodecInfo* kAudioWebMCodecs[] = {
  41   &kVorbisCodecInfo,
  42   NULL
  43 };
  44
  45 static StreamParser* BuildWebMParser() {
  46   return new WebMStreamParser();
  47 }
  48
  49 static const SupportedTypeInfo kSupportedTypeInfo[] = {
  50   { "video/webm", &BuildWebMParser, kVideoWebMCodecs },
  51   { "audio/webm", &BuildWebMParser, kAudioWebMCodecs },
  52 };
  53
  54 // Checks to see if the specified |type| and |codecs| list are supported.
  55 // Returns true if |type| and all codecs listed in |codecs| are supported.
  56 //         |factory_function| contains a function that can build a StreamParser
  57 //                            for this type.
  58 //         |has_audio| is true if an audio codec was specified.
  59 //         |has_video| is true if a video codec was specified.
  60 // Returns false otherwise. The values of |factory_function|, |has_audio|,
  61 //         and |has_video| are undefined.
  62 static bool IsSupported(const std::string& type,
  63                         std::vector<std::string>& codecs,
  64                         ParserFactoryFunction* factory_function,
  65                         bool* has_audio,
  66                         bool* has_video) {
  67   *factory_function = NULL;
  68   *has_audio = false;
  69   *has_video = false;
  70
  71   // Search for the SupportedTypeInfo for |type|
  72   for (size_t i = 0; i < arraysize(kSupportedTypeInfo); ++i) {
  73     const SupportedTypeInfo& type_info = kSupportedTypeInfo[i];
  74     if (type == type_info.type) {
  75       // Make sure all the codecs specified in |codecs| are
  76       // in the supported type info.
  77       for (size_t j = 0; j < codecs.size(); ++j) {
  78         // Search the type info for a match.
  79         bool found_codec = false;
  80         DemuxerStream::Type codec_type = DemuxerStream::UNKNOWN;
  81
  82         for (int k = 0; type_info.codecs[k]; ++k) {
  83           if (codecs[j] == type_info.codecs[k]->name) {
  84             found_codec = true;
  85             codec_type = type_info.codecs[k]->type;
  86             break;
  87           }
  88         }
  89
  90         if (!found_codec)
  91           return false;
  92
  93         switch (codec_type) {
  94           case DemuxerStream::AUDIO:
  95             *has_audio = true;
  96             break;
  97           case DemuxerStream::VIDEO:
  98             *has_video = true;
  99             break;
 100           default:
 101             DVLOG(1) << "Unsupported codec type '"<< codec_type << "' for "
 102                      << codecs[j];
 103             return false;
 104         }
 105       }
 106
 107       *factory_function = type_info.factory_function;
 108
 109       // All codecs were supported by this |type|.
 110       return true;
 111     }
 112   }
 113
 114   // |type| didn't match any of the supported types.
 115   return false;
 116 }
 117
 118 class ChunkDemuxerStream : public DemuxerStream {
 119  public:
 120   typedef std::deque<scoped_refptr<StreamParserBuffer> > BufferQueue;
 121   typedef std::deque<ReadCB> ReadCBQueue;
 122   typedef std::deque<base::Closure> ClosureQueue;
 123
 124   explicit ChunkDemuxerStream(const AudioDecoderConfig& audio_config);
 125   explicit ChunkDemuxerStream(const VideoDecoderConfig& video_config);
 126
 127   void Flush();
 128   void Seek(base::TimeDelta time);
 129
 130   // Checks if it is ok to add the |buffers| to the stream.
 131   bool CanAddBuffers(const BufferQueue& buffers) const;
 132
 133   void AddBuffers(const BufferQueue& buffers);
 134   void Shutdown();
 135
 136   // Gets the time range buffered by this object.
 137   // Returns true if there is buffered data. |start_out| & |end_out| are set to
 138   //    the start and end time of the buffered data respectively.
 139   // Returns false if no data is buffered.
 140   bool GetBufferedRange(base::TimeDelta* start_out,
 141                         base::TimeDelta* end_out) const;
 142
 143   // DemuxerStream methods.
 144   virtual void Read(const ReadCB& read_cb) OVERRIDE;
 145   virtual Type type() OVERRIDE;
 146   virtual void EnableBitstreamConverter() OVERRIDE;
 147   virtual const AudioDecoderConfig& audio_decoder_config() OVERRIDE;
 148   virtual const VideoDecoderConfig& video_decoder_config() OVERRIDE;
 149
 150  protected:
 151   virtual ~ChunkDemuxerStream();
 152
 153  private:
 154   enum State {
 155     RETURNING_DATA_FOR_READS,
 156     WAITING_FOR_SEEK,
 157     RECEIVED_EOS_WHILE_WAITING_FOR_SEEK, // EOS = End of stream.
 158     RECEIVED_EOS,
 159     RETURNING_EOS_FOR_READS,
 160     SHUTDOWN,
 161   };
 162
 163   // Assigns |state_| to |state|
 164   void ChangeState_Locked(State state);
 165
 166   // Adds the callback to |read_cbs_| so it can be called later when we
 167   // have data.
 168   void DeferRead_Locked(const ReadCB& read_cb);
 169
 170   // Creates closures that bind ReadCBs in |read_cbs_| to data in
 171   // |buffers_| and pops the callbacks & buffers from the respecive queues.
 172   void CreateReadDoneClosures_Locked(ClosureQueue* closures);
 173
 174   Type type_;
 175   AudioDecoderConfig audio_config_;
 176   VideoDecoderConfig video_config_;
 177
 178   mutable base::Lock lock_;
 179   State state_;
 180   ReadCBQueue read_cbs_;
 181   BufferQueue buffers_;
 182
 183   // Keeps track of the timestamp of the last buffer we have
 184   // added to |buffers_|. This is used to enforce buffers with strictly
 185   // monotonically increasing timestamps.
 186   base::TimeDelta last_buffer_timestamp_;
 187
 188   DISALLOW_IMPLICIT_CONSTRUCTORS(ChunkDemuxerStream);
 189 };
 190
 191 ChunkDemuxerStream::ChunkDemuxerStream(const AudioDecoderConfig& audio_config)
 192     : type_(AUDIO),
 193       state_(RETURNING_DATA_FOR_READS),
 194       last_buffer_timestamp_(kNoTimestamp()) {
 195   audio_config_.CopyFrom(audio_config);
 196 }
 197
 198
 199 ChunkDemuxerStream::ChunkDemuxerStream(const VideoDecoderConfig& video_config)
 200     : type_(VIDEO),
 201       state_(RETURNING_DATA_FOR_READS),
 202       last_buffer_timestamp_(kNoTimestamp()) {
 203   video_config_.CopyFrom(video_config);
 204 }
 205
 206 void ChunkDemuxerStream::Flush() {
 207   DVLOG(1) << "Flush()";
 208   ReadCBQueue read_cbs;
 209   {
 210     base::AutoLock auto_lock(lock_);
 211     buffers_.clear();
 212     ChangeState_Locked(WAITING_FOR_SEEK);
 213     last_buffer_timestamp_ = kNoTimestamp();
 214
 215     std::swap(read_cbs_, read_cbs);
 216   }
 217
 218   for (ReadCBQueue::iterator it = read_cbs.begin(); it != read_cbs.end(); ++it)
 219     it->Run(scoped_refptr<Buffer>());
 220 }
 221
 222 void ChunkDemuxerStream::Seek(base::TimeDelta time) {
 223   base::AutoLock auto_lock(lock_);
 224
 225   DCHECK(read_cbs_.empty());
 226
 227   if (state_ == WAITING_FOR_SEEK) {
 228     ChangeState_Locked(RETURNING_DATA_FOR_READS);
 229     return;
 230   }
 231
 232   if (state_ == RECEIVED_EOS_WHILE_WAITING_FOR_SEEK) {
 233     ChangeState_Locked(RECEIVED_EOS);
 234     return;
 235   }
 236 }
 237
 238 bool ChunkDemuxerStream::CanAddBuffers(const BufferQueue& buffers) const {
 239   base::AutoLock auto_lock(lock_);
 240
 241   // If we haven't seen any buffers yet, then anything can be added.
 242   if (last_buffer_timestamp_ == kNoTimestamp())
 243     return true;
 244
 245   if (buffers.empty())
 246     return true;
 247
 248   return (buffers.front()->GetTimestamp() > last_buffer_timestamp_);
 249 }
 250
 251 void ChunkDemuxerStream::AddBuffers(const BufferQueue& buffers) {
 252   if (buffers.empty())
 253     return;
 254
 255   ClosureQueue closures;
 256   {
 257     base::AutoLock auto_lock(lock_);
 258
 259     for (BufferQueue::const_iterator itr = buffers.begin();
 260          itr != buffers.end(); itr++) {
 261       // Make sure we aren't trying to add a buffer after we have received and
 262       // "end of stream" buffer.
 263       DCHECK_NE(state_, RECEIVED_EOS_WHILE_WAITING_FOR_SEEK);
 264       DCHECK_NE(state_, RECEIVED_EOS);
 265       DCHECK_NE(state_, RETURNING_EOS_FOR_READS);
 266
 267       if ((*itr)->IsEndOfStream()) {
 268         if (state_ == WAITING_FOR_SEEK)  {
 269           ChangeState_Locked(RECEIVED_EOS_WHILE_WAITING_FOR_SEEK);
 270         } else {
 271           ChangeState_Locked(RECEIVED_EOS);
 272         }
 273       } else {
 274         base::TimeDelta current_ts = (*itr)->GetTimestamp();
 275         if (last_buffer_timestamp_ != kNoTimestamp()) {
 276           DCHECK_GT(current_ts.ToInternalValue(),
 277                     last_buffer_timestamp_.ToInternalValue());
 278         }
 279
 280         last_buffer_timestamp_ = current_ts;
 281         buffers_.push_back(*itr);
 282       }
 283     }
 284
 285     CreateReadDoneClosures_Locked(&closures);
 286   }
 287
 288   for (ClosureQueue::iterator it = closures.begin(); it != closures.end(); ++it)
 289     it->Run();
 290 }
 291
 292 void ChunkDemuxerStream::Shutdown() {
 293   ReadCBQueue read_cbs;
 294   {
 295     base::AutoLock auto_lock(lock_);
 296     ChangeState_Locked(SHUTDOWN);
 297
 298     std::swap(read_cbs_, read_cbs);
 299     buffers_.clear();
 300   }
 301
 302   // Pass end of stream buffers to all callbacks to signal that no more data
 303   // will be sent.
 304   for (ReadCBQueue::iterator it = read_cbs.begin(); it != read_cbs.end(); ++it)
 305     it->Run(StreamParserBuffer::CreateEOSBuffer());
 306 }
 307
 308 bool ChunkDemuxerStream::GetBufferedRange(
 309     base::TimeDelta* start_out, base::TimeDelta* end_out) const {
 310   base::AutoLock auto_lock(lock_);
 311
 312   if (buffers_.empty())
 313     return false;
 314
 315   *start_out = buffers_.front()->GetTimestamp();
 316   *end_out = buffers_.back()->GetTimestamp();
 317
 318   base::TimeDelta end_duration = buffers_.back()->GetDuration();
 319   if (end_duration != kNoTimestamp())
 320     *end_out += end_duration;
 321
 322   return true;
 323 }
 324
 325 // Helper function that makes sure |read_cb| runs on |message_loop|.
 326 static void RunOnMessageLoop(const DemuxerStream::ReadCB& read_cb,
 327                              MessageLoop* message_loop,
 328                              const scoped_refptr<Buffer>& buffer) {
 329   if (MessageLoop::current() != message_loop) {
 330     message_loop->PostTask(FROM_HERE, base::Bind(
 331         &RunOnMessageLoop, read_cb, message_loop, buffer));
 332     return;
 333   }
 334
 335   read_cb.Run(buffer);
 336 }
 337
 338 // DemuxerStream methods.
 339 void ChunkDemuxerStream::Read(const ReadCB& read_cb) {
 340   scoped_refptr<Buffer> buffer;
 341
 342   {
 343     base::AutoLock auto_lock(lock_);
 344
 345     switch (state_) {
 346       case RETURNING_DATA_FOR_READS:
 347         // If we don't have any buffers ready or already have
 348         // pending reads, then defer this read.
 349         if (buffers_.empty() || !read_cbs_.empty()) {
 350           DeferRead_Locked(read_cb);
 351           return;
 352         }
 353
 354         buffer = buffers_.front();
 355         buffers_.pop_front();
 356         break;
 357
 358       case WAITING_FOR_SEEK:
 359       case RECEIVED_EOS_WHILE_WAITING_FOR_SEEK:
 360         // Null buffers should be returned in this state since we are waiting
 361         // for a seek. Any buffers in |buffers_| should NOT be returned because
 362         // they are associated with the seek.
 363         DCHECK(read_cbs_.empty());
 364         break;
 365       case RECEIVED_EOS:
 366         DCHECK(read_cbs_.empty());
 367
 368         if (buffers_.empty()) {
 369           ChangeState_Locked(RETURNING_EOS_FOR_READS);
 370           buffer = StreamParserBuffer::CreateEOSBuffer();
 371         } else {
 372           buffer = buffers_.front();
 373           buffers_.pop_front();
 374         }
 375         break;
 376
 377       case RETURNING_EOS_FOR_READS:
 378       case SHUTDOWN:
 379         DCHECK(buffers_.empty());
 380         DCHECK(read_cbs_.empty());
 381         buffer = StreamParserBuffer::CreateEOSBuffer();
 382     }
 383   }
 384
 385   read_cb.Run(buffer);
 386 }
 387
 388 DemuxerStream::Type ChunkDemuxerStream::type() { return type_; }
 389
 390 void ChunkDemuxerStream::EnableBitstreamConverter() {}
 391
 392 const AudioDecoderConfig& ChunkDemuxerStream::audio_decoder_config() {
 393   CHECK_EQ(type_, AUDIO);
 394   return audio_config_;
 395 }
 396
 397 const VideoDecoderConfig& ChunkDemuxerStream::video_decoder_config() {
 398   CHECK_EQ(type_, VIDEO);
 399   return video_config_;
 400 }
 401
 402 void ChunkDemuxerStream::ChangeState_Locked(State state) {
 403   lock_.AssertAcquired();
 404   state_ = state;
 405 }
 406
 407 ChunkDemuxerStream::~ChunkDemuxerStream() {}
 408
 409 void ChunkDemuxerStream::DeferRead_Locked(const ReadCB& read_cb) {
 410   lock_.AssertAcquired();
 411   // Wrap & store |read_cb| so that it will
 412   // get called on the current MessageLoop.
 413   read_cbs_.push_back(base::Bind(&RunOnMessageLoop, read_cb,
 414                                  MessageLoop::current()));
 415 }
 416
 417 void ChunkDemuxerStream::CreateReadDoneClosures_Locked(ClosureQueue* closures) {
 418   lock_.AssertAcquired();
 419
 420   if (state_ != RETURNING_DATA_FOR_READS && state_ != RECEIVED_EOS)
 421     return;
 422
 423   while (!buffers_.empty() && !read_cbs_.empty()) {
 424     closures->push_back(base::Bind(read_cbs_.front(), buffers_.front()));
 425     buffers_.pop_front();
 426     read_cbs_.pop_front();
 427   }
 428
 429   if (state_ != RECEIVED_EOS || !buffers_.empty() || read_cbs_.empty())
 430     return;
 431
 432   // Push enough EOS buffers to satisfy outstanding Read() requests.
 433   scoped_refptr<Buffer> end_of_stream_buffer =
 434       StreamParserBuffer::CreateEOSBuffer();
 435   while (!read_cbs_.empty()) {
 436     closures->push_back(base::Bind(read_cbs_.front(), end_of_stream_buffer));
 437     read_cbs_.pop_front();
 438   }
 439
 440   ChangeState_Locked(RETURNING_EOS_FOR_READS);
 441 }
 442
 443 ChunkDemuxer::ChunkDemuxer(ChunkDemuxerClient* client)
 444     : state_(WAITING_FOR_INIT),
 445       host_(NULL),
 446       client_(client),
 447       buffered_bytes_(0),
 448       seek_waits_for_data_(true) {
 449   DCHECK(client);
 450 }
 451
 452 void ChunkDemuxer::Initialize(DemuxerHost* host,
 453                               const PipelineStatusCB& cb) {
 454   DVLOG(1) << "Init()";
 455   {
 456     base::AutoLock auto_lock(lock_);
 457     DCHECK_EQ(state_, WAITING_FOR_INIT);
 458     host_ = host;
 459
 460     ChangeState_Locked(INITIALIZING);
 461     init_cb_ = cb;
 462   }
 463
 464   client_->DemuxerOpened(this);
 465 }
 466
 467 void ChunkDemuxer::Stop(const base::Closure& callback) {
 468   DVLOG(1) << "Stop()";
 469   Shutdown();
 470   callback.Run();
 471 }
 472
 473 void ChunkDemuxer::Seek(base::TimeDelta time, const PipelineStatusCB& cb) {
 474   DVLOG(1) << "Seek(" << time.InSecondsF() << ")";
 475
 476   PipelineStatus status = PIPELINE_ERROR_INVALID_STATE;
 477   {
 478     base::AutoLock auto_lock(lock_);
 479
 480     if (state_ == INITIALIZED || state_ == ENDED) {
 481       if (audio_)
 482         audio_->Seek(time);
 483
 484       if (video_)
 485         video_->Seek(time);
 486
 487       if (seek_waits_for_data_) {
 488         DVLOG(1) << "Seek() : waiting for more data to arrive.";
 489         seek_cb_ = cb;
 490         return;
 491       }
 492
 493       status = PIPELINE_OK;
 494     }
 495   }
 496
 497   cb.Run(status);
 498 }
 499
 500 void ChunkDemuxer::OnAudioRendererDisabled() {
 501   base::AutoLock auto_lock(lock_);
 502   audio_ = NULL;
 503 }
 504
 505 int ChunkDemuxer::GetBitrate() {
 506   // TODO(acolwell): Implement bitrate reporting.
 507   return 0;
 508 }
 509
 510 // Demuxer implementation.
 511 scoped_refptr<DemuxerStream> ChunkDemuxer::GetStream(
 512     DemuxerStream::Type type) {
 513   if (type == DemuxerStream::VIDEO)
 514     return video_;
 515
 516   if (type == DemuxerStream::AUDIO)
 517     return audio_;
 518
 519   return NULL;
 520 }
 521
 522 base::TimeDelta ChunkDemuxer::GetStartTime() const {
 523   DVLOG(1) << "GetStartTime()";
 524   // TODO(acolwell) : Fix this so it uses the time on the first packet.
 525   return base::TimeDelta();
 526 }
 527
 528 void ChunkDemuxer::FlushData() {
 529   DVLOG(1) << "FlushData()";
 530   base::AutoLock auto_lock(lock_);
 531   DCHECK(state_ == INITIALIZED || state_ == ENDED || state_ == SHUTDOWN);
 532
 533   if (state_ == SHUTDOWN)
 534     return;
 535
 536   if (audio_.get())
 537     audio_->Flush();
 538
 539   if (video_.get())
 540     video_->Flush();
 541
 542   source_buffer_->Flush();
 543
 544   seek_waits_for_data_ = true;
 545   ChangeState_Locked(INITIALIZED);
 546 }
 547
 548 ChunkDemuxer::Status ChunkDemuxer::AddId(const std::string& id,
 549                                          const std::string& type,
 550                                          std::vector<std::string>& codecs) {
 551   DCHECK_GT(codecs.size(), 0u);
 552
 553   bool has_audio = false;
 554   bool has_video = false;
 555   ParserFactoryFunction factory_function = NULL;
 556   if (!IsSupported(type, codecs, &factory_function, &has_audio, &has_video))
 557     return kNotSupported;
 558
 559   // TODO(acolwell): Support for more than one ID
 560   // will be added as part of http://crbug.com/122909
 561   if (!source_id_.empty())
 562     return kReachedIdLimit;
 563
 564   source_id_ = id;
 565
 566   StreamParser::NewBuffersCB audio_cb;
 567   StreamParser::NewBuffersCB video_cb;
 568
 569   if (has_audio) {
 570     audio_cb = base::Bind(&ChunkDemuxer::OnAudioBuffers,
 571                           base::Unretained(this));
 572   }
 573
 574   if (has_video) {
 575     video_cb = base::Bind(&ChunkDemuxer::OnVideoBuffers,
 576                           base::Unretained(this));
 577   }
 578
 579   scoped_ptr<StreamParser> stream_parser(factory_function());
 580
 581   CHECK(stream_parser.get());
 582
 583   source_buffer_.reset(new SourceBuffer());
 584   source_buffer_->Init(
 585       stream_parser.Pass(),
 586       base::Bind(&ChunkDemuxer::OnSourceBufferInitDone, this),
 587       base::Bind(&ChunkDemuxer::OnNewConfigs, base::Unretained(this)),
 588       audio_cb,
 589       video_cb,
 590       base::Bind(&ChunkDemuxer::OnKeyNeeded, base::Unretained(this)));
 591
 592   return kOk;
 593 }
 594
 595 void ChunkDemuxer::RemoveId(const std::string& id) {
 596   CHECK(!source_id_.empty());
 597   CHECK_EQ(source_id_, id);
 598   source_id_ = "";
 599 }
 600
 601 bool ChunkDemuxer::GetBufferedRanges(const std::string& id,
 602                                      Ranges* ranges_out) const {
 603   DCHECK(!id.empty());
 604   DCHECK_EQ(source_id_, id);
 605   DCHECK(ranges_out);
 606
 607   base::AutoLock auto_lock(lock_);
 608   base::TimeDelta start = kNoTimestamp();
 609   base::TimeDelta end;
 610   base::TimeDelta tmp_start;
 611   base::TimeDelta tmp_end;
 612
 613   if (audio_ && audio_->GetBufferedRange(&tmp_start, &tmp_end)) {
 614     start = tmp_start;
 615     end = tmp_end;
 616   }
 617
 618   if (video_ && video_->GetBufferedRange(&tmp_start, &tmp_end)) {
 619     if (start == kNoTimestamp()) {
 620       start = tmp_start;
 621       end = tmp_end;
 622     } else {
 623       start = std::min(start, tmp_start);
 624       end = std::max(end, tmp_end);
 625     }
 626   }
 627
 628   if (start == kNoTimestamp())
 629     return false;
 630
 631   ranges_out->resize(1);
 632   (*ranges_out)[0].first = start;
 633   (*ranges_out)[0].second = end;
 634   return true;
 635 }
 636
 637 bool ChunkDemuxer::AppendData(const std::string& id,
 638                               const uint8* data,
 639                               size_t length) {
 640   DVLOG(1) << "AppendData(" << id << ", " << length << ")";
 641
 642   // TODO(acolwell): Remove when http://webk.it/83788 fix lands.
 643   if (source_id_.empty()) {
 644     std::vector<std::string> codecs(2);
 645     codecs[0] = "vp8";
 646     codecs[1] = "vorbis";
 647     AddId(id, "video/webm", codecs);
 648   }
 649
 650   DCHECK(!source_id_.empty());
 651   DCHECK_EQ(source_id_, id);
 652   DCHECK(!id.empty());
 653   DCHECK(data);
 654   DCHECK_GT(length, 0u);
 655
 656   int64 buffered_bytes = 0;
 657
 658   PipelineStatusCB cb;
 659   {
 660     base::AutoLock auto_lock(lock_);
 661
 662     // Capture |seek_waits_for_data_| state before we start parsing.
 663     // Its state can be changed by OnAudioBuffers() or OnVideoBuffers()
 664     // calls during the parse.
 665     bool old_seek_waits_for_data = seek_waits_for_data_;
 666
 667     switch (state_) {
 668       case INITIALIZING:
 669         if (!source_buffer_->AppendData(data, length)) {
 670           DCHECK_EQ(state_, INITIALIZING);
 671           ReportError_Locked(DEMUXER_ERROR_COULD_NOT_OPEN);
 672           return true;
 673         }
 674         break;
 675
 676       case INITIALIZED: {
 677         if (!source_buffer_->AppendData(data, length)) {
 678           ReportError_Locked(PIPELINE_ERROR_DECODE);
 679           return true;
 680         }
 681       } break;
 682
 683       case WAITING_FOR_INIT:
 684       case ENDED:
 685       case PARSE_ERROR:
 686       case SHUTDOWN:
 687         DVLOG(1) << "AppendData(): called in unexpected state " << state_;
 688         return false;
 689     }
 690
 691     // Check to see if parsing triggered seek_waits_for_data_ to go from true to
 692     // false. This indicates we have parsed enough data to complete the seek.
 693     if (old_seek_waits_for_data && !seek_waits_for_data_ &&
 694         !seek_cb_.is_null()) {
 695       std::swap(cb, seek_cb_);
 696     }
 697
 698     buffered_bytes_ += length;
 699     buffered_bytes = buffered_bytes_;
 700   }
 701
 702   // Notify the host of 'network activity' because we got data.
 703   host_->SetBufferedBytes(buffered_bytes);
 704
 705   host_->SetNetworkActivity(true);
 706
 707   if (!cb.is_null())
 708     cb.Run(PIPELINE_OK);
 709
 710   return true;
 711 }
 712
 713 void ChunkDemuxer::Abort(const std::string& id) {
 714   DCHECK(!id.empty());
 715   DCHECK_EQ(source_id_, id);
 716
 717   source_buffer_->Flush();
 718 }
 719
 720 void ChunkDemuxer::EndOfStream(PipelineStatus status) {
 721   DVLOG(1) << "EndOfStream(" << status << ")";
 722   base::AutoLock auto_lock(lock_);
 723   DCHECK_NE(state_, WAITING_FOR_INIT);
 724   DCHECK_NE(state_, ENDED);
 725
 726   if (state_ == SHUTDOWN || state_ == PARSE_ERROR)
 727     return;
 728
 729   if (state_ == INITIALIZING) {
 730     ReportError_Locked(DEMUXER_ERROR_COULD_NOT_OPEN);
 731     return;
 732   }
 733
 734   ChangeState_Locked(ENDED);
 735
 736   if (status != PIPELINE_OK) {
 737     ReportError_Locked(status);
 738     return;
 739   }
 740
 741   // Create an end of stream buffer.
 742   ChunkDemuxerStream::BufferQueue buffers;
 743   buffers.push_back(StreamParserBuffer::CreateEOSBuffer());
 744
 745   if (audio_.get())
 746     audio_->AddBuffers(buffers);
 747
 748   if (video_.get())
 749     video_->AddBuffers(buffers);
 750 }
 751
 752 bool ChunkDemuxer::HasEnded() {
 753   base::AutoLock auto_lock(lock_);
 754   return (state_ == ENDED);
 755 }
 756
 757 void ChunkDemuxer::Shutdown() {
 758   DVLOG(1) << "Shutdown()";
 759   PipelineStatusCB cb;
 760   {
 761     base::AutoLock auto_lock(lock_);
 762
 763     if (state_ == SHUTDOWN)
 764       return;
 765
 766     std::swap(cb, seek_cb_);
 767
 768     if (audio_.get())
 769       audio_->Shutdown();
 770
 771     if (video_.get())
 772       video_->Shutdown();
 773
 774     source_buffer_.reset();
 775
 776     ChangeState_Locked(SHUTDOWN);
 777   }
 778
 779   if (!cb.is_null())
 780     cb.Run(PIPELINE_ERROR_ABORT);
 781
 782   client_->DemuxerClosed();
 783 }
 784
 785 void ChunkDemuxer::ChangeState_Locked(State new_state) {
 786   lock_.AssertAcquired();
 787   state_ = new_state;
 788 }
 789
 790 ChunkDemuxer::~ChunkDemuxer() {
 791   DCHECK_NE(state_, INITIALIZED);
 792 }
 793
 794 void ChunkDemuxer::ReportError_Locked(PipelineStatus error) {
 795   lock_.AssertAcquired();
 796   DCHECK_NE(error, PIPELINE_OK);
 797
 798   ChangeState_Locked(PARSE_ERROR);
 799
 800   PipelineStatusCB cb;
 801
 802   if (!init_cb_.is_null()) {
 803     std::swap(cb, init_cb_);
 804   } else {
 805     if (!seek_cb_.is_null())
 806       std::swap(cb, seek_cb_);
 807
 808     if (audio_.get())
 809       audio_->Shutdown();
 810
 811     if (video_.get())
 812       video_->Shutdown();
 813   }
 814
 815   if (!cb.is_null()) {
 816     base::AutoUnlock auto_unlock(lock_);
 817     cb.Run(error);
 818     return;
 819   }
 820
 821   base::AutoUnlock auto_unlock(lock_);
 822   host_->OnDemuxerError(error);
 823 }
 824
 825 void ChunkDemuxer::OnSourceBufferInitDone(bool success,
 826                                           base::TimeDelta duration) {
 827   lock_.AssertAcquired();
 828   DCHECK_EQ(state_, INITIALIZING);
 829   if (!success || (!audio_.get() && !video_.get())) {
 830     ReportError_Locked(DEMUXER_ERROR_COULD_NOT_OPEN);
 831     return;
 832   }
 833
 834   duration_ = duration;
 835   host_->SetDuration(duration_);
 836   host_->SetCurrentReadPosition(0);
 837
 838   ChangeState_Locked(INITIALIZED);
 839   PipelineStatusCB cb;
 840   std::swap(cb, init_cb_);
 841   cb.Run(PIPELINE_OK);
 842 }
 843
 844 bool ChunkDemuxer::OnNewConfigs(const AudioDecoderConfig& audio_config,
 845                                 const VideoDecoderConfig& video_config) {
 846   CHECK(audio_config.IsValidConfig() || video_config.IsValidConfig());
 847   lock_.AssertAcquired();
 848
 849   // Only allow a single audio config for now.
 850   if (audio_config.IsValidConfig()) {
 851     if (audio_.get())
 852       return false;
 853
 854     audio_ = new ChunkDemuxerStream(audio_config);
 855   }
 856
 857   // Only allow a single video config for now.
 858   if (video_config.IsValidConfig()) {
 859     if (video_.get())
 860       return false;
 861
 862     video_ = new ChunkDemuxerStream(video_config);
 863   }
 864
 865   return true;
 866 }
 867
 868 bool ChunkDemuxer::OnAudioBuffers(const StreamParser::BufferQueue& buffers) {
 869   if (!audio_.get())
 870     return false;
 871
 872   if (!audio_->CanAddBuffers(buffers))
 873     return false;
 874
 875   audio_->AddBuffers(buffers);
 876   seek_waits_for_data_ = false;
 877
 878   return true;
 879 }
 880
 881 bool ChunkDemuxer::OnVideoBuffers(const StreamParser::BufferQueue& buffers) {
 882   if (!video_.get())
 883     return false;
 884
 885   if (!video_->CanAddBuffers(buffers))
 886     return false;
 887
 888   video_->AddBuffers(buffers);
 889   seek_waits_for_data_ = false;
 890
 891   return true;
 892 }
 893
 894 bool ChunkDemuxer::OnKeyNeeded(scoped_array<uint8> init_data,
 895                                int init_data_size) {
 896   client_->KeyNeeded(init_data.Pass(), init_data_size);
 897   return true;
 898 }
 899
 900 }  // namespace media