Handle possible NewSampleEvent before ParserInitEvent

For WebM contents, it is possible that we may receive some NewSampleEvent
before receiving ParserInitEvent. This is because init event is fired
after analyzing a video block which could come after an audio block.

Also modified Flush() function to return a bool to indicate whether the
the flush is successful and whether the samples are handled correctly.

And added macro to ensure Flush() and Parse() results are handled.

Fixes #71

Change-Id: I2294d6f529f54e4578344916559bb1bc116c745a

packager/media/base/demuxer.cc

@@ -22,9 +22,13 @@
 #include "packager/media/formats/wvm/wvm_media_parser.h"
 
 namespace {
-const size_t kInitBufSize = 0x10000;  // 65KB, sufficient to determine the
-                                      // container and likely all init data.
-const size_t kBufSize = 0x200000;     // 2MB
+// 65KB, sufficient to determine the container and likely all init data.
+const size_t kInitBufSize = 0x10000;
+const size_t kBufSize = 0x200000;  // 2MB
+// Maximum number of allowed queued samples. If we are receiving a lot of
+// samples before seeing init_event, something is not right. The number
+// set here is arbitrary though.
+const size_t kQueuedSamplesLimit = 10000;
 }
 
 namespace edash_packager {
@@ -126,14 +130,40 @@ void Demuxer::ParserInitEvent(
   }
 }
 
+Demuxer::QueuedSample::QueuedSample(uint32_t local_track_id,
+                                      scoped_refptr<MediaSample> local_sample)
+    : track_id(local_track_id), sample(local_sample) {}
+Demuxer::QueuedSample::~QueuedSample() {}
+
 bool Demuxer::NewSampleEvent(uint32_t track_id,
                              const scoped_refptr<MediaSample>& sample) {
+  if (!init_event_received_) {
+    if (queued_samples_.size() >= kQueuedSamplesLimit) {
+      LOG(ERROR) << "Queued samples limit reached: " << kQueuedSamplesLimit;
+      return false;
+    }
+    queued_samples_.push_back(QueuedSample(track_id, sample));
+    return true;
+  }
+  while (!queued_samples_.empty()) {
+    if (!PushSample(queued_samples_.front().track_id,
+                    queued_samples_.front().sample)) {
+      return false;
+    }
+    queued_samples_.pop_front();
+  }
+  return PushSample(track_id, sample);
+}
+
+bool Demuxer::PushSample(uint32_t track_id,
+                         const scoped_refptr<MediaSample>& sample) {
   std::vector<MediaStream*>::iterator it = streams_.begin();
   for (; it != streams_.end(); ++it) {
     if (track_id == (*it)->info()->track_id()) {
       return (*it)->PushSample(sample).ok();
     }
   }
+  LOG(ERROR) << "Track " << track_id << " not found.";
   return false;
 }
 
@@ -183,7 +213,8 @@ Status Demuxer::Parse() {
 
   int64_t bytes_read = media_file_->Read(buffer_.get(), kBufSize);
   if (bytes_read == 0) {
-    parser_->Flush();
+    if (!parser_->Flush())
+      return Status(error::PARSER_FAILURE, "Failed to flush.");
     return Status(error::END_OF_STREAM, "");
   } else if (bytes_read < 0) {
     return Status(error::FILE_FAILURE, "Cannot read file " + file_name_);

packager/media/base/demuxer.h

@@ -7,8 +7,10 @@
 #ifndef MEDIA_BASE_DEMUXER_H_
 #define MEDIA_BASE_DEMUXER_H_
 
+#include <deque>
 #include <vector>
 
+#include "packager/base/compiler_specific.h"
 #include "packager/base/memory/ref_counted.h"
 #include "packager/base/memory/scoped_ptr.h"
 #include "packager/media/base/container_names.h"
@@ -70,15 +72,30 @@ class Demuxer {
   MediaContainerName container_name() { return container_name_; }
 
  private:
-  // Parser event handlers.
+  struct QueuedSample {
+    QueuedSample(uint32_t track_id, scoped_refptr<MediaSample> sample);
+    ~QueuedSample();
+
+    uint32_t track_id;
+    scoped_refptr<MediaSample> sample;
+  };
+
+  // Parser init event.
   void ParserInitEvent(const std::vector<scoped_refptr<StreamInfo> >& streams);
+  // Parser new sample event handler. Queues the samples if init event has not
+  // been received, otherwise calls PushSample() to push the sample to
+  // corresponding stream.
   bool NewSampleEvent(uint32_t track_id,
                       const scoped_refptr<MediaSample>& sample);
+  // Helper function to push the sample to corresponding stream.
+  bool PushSample(uint32_t track_id, const scoped_refptr<MediaSample>& sample);
 
   std::string file_name_;
   File* media_file_;
   bool init_event_received_;
   Status init_parsing_status_;
+  // Queued samples received in NewSampleEvent() before ParserInitEvent().
+  std::deque<QueuedSample> queued_samples_;
   scoped_ptr<MediaParser> parser_;
   std::vector<MediaStream*> streams_;
   MediaContainerName container_name_;

packager/media/base/media_parser.h

@@ -11,6 +11,7 @@
 #include <vector>
 
 #include "packager/base/callback.h"
+#include "packager/base/compiler_specific.h"
 #include "packager/base/memory/ref_counted.h"
 #include "packager/base/memory/scoped_ptr.h"
 #include "packager/media/base/container_names.h"
@@ -55,11 +56,12 @@ class MediaParser {
 
   /// Flush data currently in the parser and put the parser in a state where it
   /// can receive data for a new seek point.
-  virtual void Flush() = 0;
+  /// @return true if successful, false otherwise.
+  virtual bool Flush() WARN_UNUSED_RESULT = 0;
 
   /// Should be called when there is new data to parse.
   /// @return true if successful.
-  virtual bool Parse(const uint8_t* buf, int size) = 0;
+  virtual bool Parse(const uint8_t* buf, int size) WARN_UNUSED_RESULT = 0;
 
  private:
   DISALLOW_COPY_AND_ASSIGN(MediaParser);

packager/media/formats/mp2t/mp2t_media_parser.cc

@@ -166,7 +166,7 @@ void Mp2tMediaParser::Init(
   new_sample_cb_ = new_sample_cb;
 }
 
-void Mp2tMediaParser::Flush() {
+bool Mp2tMediaParser::Flush() {
   DVLOG(1) << "Mp2tMediaParser::Flush";
 
   // Flush the buffers and reset the pids.
@@ -176,12 +176,13 @@ void Mp2tMediaParser::Flush() {
     PidState* pid_state = it->second;
     pid_state->Flush();
   }
-  EmitRemainingSamples();
+  bool result = EmitRemainingSamples();
   STLDeleteValues(&pids_);
 
   // Remove any bytes left in the TS buffer.
   // (i.e. any partial TS packet => less than 188 bytes).
   ts_byte_queue_.Reset();
+  return result;
 }
 
 bool Mp2tMediaParser::Parse(const uint8_t* buf, int size) {

packager/media/formats/mp2t/mp2t_media_parser.h

@@ -8,6 +8,7 @@
 #include <deque>
 #include <map>
 
+#include "packager/base/compiler_specific.h"
 #include "packager/base/memory/ref_counted.h"
 #include "packager/base/memory/scoped_ptr.h"
 #include "packager/media/base/byte_queue.h"
@@ -32,14 +33,14 @@ class Mp2tMediaParser : public MediaParser {
   Mp2tMediaParser();
   ~Mp2tMediaParser() override;
 
-  // MediaParser implementation overrides.
+  /// @name MediaParser implementation overrides.
+  /// @{
   void Init(const InitCB& init_cb,
             const NewSampleCB& new_sample_cb,
             KeySource* decryption_key_source) override;
-
-  void Flush() override;
-
-  bool Parse(const uint8_t* buf, int size) override;
+  bool Flush() override WARN_UNUSED_RESULT;
+  bool Parse(const uint8_t* buf, int size) override WARN_UNUSED_RESULT;
+  /// @}
 
  private:
   typedef std::map<int, PidState*> PidMap;

packager/media/formats/mp2t/mp2t_media_parser_unittest.cc

@@ -125,15 +125,15 @@ TEST_F(Mp2tMediaParserTest, UnalignedAppend17) {
   // Test small, non-segment-aligned appends.
   ParseMpeg2TsFile("bear-1280x720.ts", 17);
   EXPECT_EQ(video_frame_count_, 80);
-  parser_->Flush();
+  EXPECT_TRUE(parser_->Flush());
   EXPECT_EQ(video_frame_count_, 82);
 }
 
 TEST_F(Mp2tMediaParserTest, UnalignedAppend512) {
   // Test small, non-segment-aligned appends.
   ParseMpeg2TsFile("bear-1280x720.ts", 512);
   EXPECT_EQ(video_frame_count_, 80);
-  parser_->Flush();
+  EXPECT_TRUE(parser_->Flush());
   EXPECT_EQ(video_frame_count_, 82);
 }
 
@@ -143,7 +143,7 @@ TEST_F(Mp2tMediaParserTest, TimestampWrapAround) {
   // (close to 2^33 / 90000) which results in timestamps wrap around
   // in the Mpeg2 TS stream.
   ParseMpeg2TsFile("bear-1280x720_ptswraparound.ts", 512);
-  parser_->Flush();
+  EXPECT_TRUE(parser_->Flush());
   EXPECT_EQ(video_frame_count_, 82);
   EXPECT_GE(video_min_dts_, static_cast<int64_t>(95443 - 1) * kMpeg2Timescale);
   EXPECT_LE(video_max_dts_, static_cast<int64_t>(95443 + 4) * kMpeg2Timescale);

packager/media/formats/mp4/mp4_media_parser.cc

@@ -119,10 +119,11 @@ void MP4MediaParser::Reset() {
   mdat_tail_ = 0;
 }
 
-void MP4MediaParser::Flush() {
+bool MP4MediaParser::Flush() {
   DCHECK_NE(state_, kWaitingForInit);
   Reset();
   ChangeState(kParsingBoxes);
+  return true;
 }
 
 bool MP4MediaParser::Parse(const uint8_t* buf, int size) {

packager/media/formats/mp4/mp4_media_parser.h

@@ -40,8 +40,8 @@ class MP4MediaParser : public MediaParser {
   void Init(const InitCB& init_cb,
             const NewSampleCB& new_sample_cb,
             KeySource* decryption_key_source) override;
-  void Flush() override;
-  bool Parse(const uint8_t* buf, int size) override;
+  bool Flush() override WARN_UNUSED_RESULT;
+  bool Parse(const uint8_t* buf, int size) override WARN_UNUSED_RESULT;
   /// @}
 
   /// Handles ISO-BMFF containers which have the 'moov' box trailing the

packager/media/formats/mp4/mp4_media_parser_unittest.cc

@@ -194,7 +194,7 @@ TEST_F(MP4MediaParserTest, Flush) {
 
   std::vector<uint8_t> buffer = ReadTestDataFile("bear-640x360-av_frag.mp4");
   EXPECT_TRUE(AppendDataInPieces(buffer.data(), 65536, 512));
-  parser_->Flush();
+  EXPECT_TRUE(parser_->Flush());
   EXPECT_EQ(2u, num_streams_);
   EXPECT_NE(0u, num_samples_);
   num_samples_ = 0;
@@ -214,7 +214,7 @@ TEST_F(MP4MediaParserTest, NoMoovAfterFlush) {
 
   std::vector<uint8_t> buffer = ReadTestDataFile("bear-640x360-av_frag.mp4");
   EXPECT_TRUE(AppendDataInPieces(buffer.data(), buffer.size(), 512));
-  parser_->Flush();
+  EXPECT_TRUE(parser_->Flush());
 
   const int kFirstMoofOffset = 1308;
   EXPECT_TRUE(AppendDataInPieces(

packager/media/formats/webm/webm_cluster_parser.cc

@@ -103,11 +103,12 @@ void WebMClusterParser::Reset() {
   ResetTextTracks();
 }
 
-void WebMClusterParser::Flush() {
+bool WebMClusterParser::Flush() {
   // Estimate the duration of the last frame if necessary.
-  audio_.ApplyDurationEstimateIfNeeded();
-  video_.ApplyDurationEstimateIfNeeded();
+  bool audio_result = audio_.ApplyDurationEstimateIfNeeded();
+  bool video_result = video_.ApplyDurationEstimateIfNeeded();
   Reset();
+  return audio_result && video_result;
 }
 
 int WebMClusterParser::Parse(const uint8_t* buf, int size) {
@@ -524,9 +525,9 @@ bool WebMClusterParser::Track::EmitBuffer(
   return EmitBufferHelp(buffer);
 }
 
-void WebMClusterParser::Track::ApplyDurationEstimateIfNeeded() {
+bool WebMClusterParser::Track::ApplyDurationEstimateIfNeeded() {
   if (!last_added_buffer_missing_duration_.get())
-    return;
+    return true;
 
   int64_t estimated_duration = GetDurationEstimate();
   last_added_buffer_missing_duration_->set_duration(estimated_duration);
@@ -544,8 +545,10 @@ void WebMClusterParser::Track::ApplyDurationEstimateIfNeeded() {
 
   // Don't use the applied duration as a future estimation (don't use
   // EmitBufferHelp() here.)
-  new_sample_cb_.Run(track_num_, last_added_buffer_missing_duration_);
+  if (!new_sample_cb_.Run(track_num_, last_added_buffer_missing_duration_))
+    return false;
   last_added_buffer_missing_duration_ = NULL;
+  return true;
 }
 
 void WebMClusterParser::Track::Reset() {
@@ -583,8 +586,7 @@ bool WebMClusterParser::Track::EmitBufferHelp(
     }
   }
 
-  new_sample_cb_.Run(track_num_, buffer);
-  return true;
+  return new_sample_cb_.Run(track_num_, buffer);
 }
 
 int64_t WebMClusterParser::Track::GetDurationEstimate() {

packager/media/formats/webm/webm_cluster_parser.h

@@ -10,6 +10,7 @@
 #include <set>
 #include <string>
 
+#include "packager/base/compiler_specific.h"
 #include "packager/base/memory/scoped_ptr.h"
 #include "packager/media/base/decryptor_source.h"
 #include "packager/media/base/media_parser.h"
@@ -55,7 +56,7 @@ class WebMClusterParser : public WebMParserClient {
     // for this buffer using helper function GetDurationEstimate() then emits it
     // and unsets |last_added_buffer_missing_duration_| (This method helps
     // stream parser emit all buffers in a media segment).
-    void ApplyDurationEstimateIfNeeded();
+    bool ApplyDurationEstimateIfNeeded();
 
     // Clears all buffer state, including any possibly held-aside buffer that
     // was missing duration.
@@ -135,7 +136,8 @@ class WebMClusterParser : public WebMParserClient {
 
   /// Flush data currently in the parser and reset the parser so it can accept a
   /// new cluster.
-  void Flush();
+  /// @return true on success, false otherwise.
+  bool Flush() WARN_UNUSED_RESULT;
 
   /// Parses a WebM cluster element in |buf|.
   /// @return -1 if the parse fails.