Mpeg2 TS stream parser for media source.

media/mpeg2/mpeg2ts_stream_parser.cc

Index: media/mpeg2/mpeg2ts_stream_parser.cc
diff --git a/media/mpeg2/mpeg2ts_stream_parser.cc b/media/mpeg2/mpeg2ts_stream_parser.cc
new file mode 100644
index 0000000000000000000000000000000000000000..ae446b5e621514bb98db4ecc3275f9e9b54bcf1d
--- /dev/null
+++ b/media/mpeg2/mpeg2ts_stream_parser.cc
@@ -0,0 +1,589 @@
+// Copyright (c) 2013 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/mpeg2/mpeg2ts_stream_parser.h"
+
+#include "base/bind.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/stl_util.h"
+#include "media/base/audio_decoder_config.h"
+#include "media/base/buffers.h"
+#include "media/base/stream_parser_buffer.h"
+#include "media/base/video_decoder_config.h"
+#include "media/mpeg2/es_parser.h"
+#include "media/mpeg2/es_parser_adts.h"
+#include "media/mpeg2/es_parser_h264.h"
+#include "media/mpeg2/mpeg2ts_common.h"
+#include "media/mpeg2/mpeg2ts_pat.h"
+#include "media/mpeg2/mpeg2ts_pes.h"
+#include "media/mpeg2/mpeg2ts_pmt.h"
+#include "media/mpeg2/mpeg2ts_section_parser.h"
+#include "media/mpeg2/ts_packet.h"
+
+namespace {
+
+enum StreamType {
+  // ISO-13818.1 / ITU H.222 Table 2.34 "Stream type assignments"
+  kStreamTypeMpeg1Audio = 0x3,
+  kStreamTypeAAC = 0xf,
+  kStreamTypeAVC = 0x1b,
+};
+
+}
+
+namespace media {
+namespace mpeg2ts {
+
+class PidState {
+ public:
+  enum PidType {
+    kPidPat,
+    kPidPmt,
+    kPidAudioPes,
+    kPidVideoPes,
+  };
+
+  // Take ownership of |section_parser|.
+  PidState(int pid, PidType pid_tyoe,
+           Mpeg2TsSectionParser* section_parser);

[acolwell GONE FROM CHROMIUM, 2013/09/05 18:29:10]

nit: Use scoped_ptr<Mpeg2TsSectionParser> here so the comment is unnecessary.

[damienv1, 2013/09/09 23:29:45]

Done.

+
+  // Extract the content of the TS packet and parse it.
+  // Return true if successful.
+  bool PushTsPacket(TsPacket* ts_packet);
+
+  // Flush the PID state (possibly emitting some pending frames)
+  // and reset its state.
+  void Flush();

[acolwell GONE FROM CHROMIUM, 2013/09/05 18:29:10]

nit: EmitPendingBuffersAndReset() seems like a more appropriate name.

+
+  // Enable/disable the PID.
+  // Disabling a PID will reset its state and ignore any further incoming TS
+  // packets.
+  void Enable();
+  void Disable();
+  bool IsEnabled();

[acolwell GONE FROM CHROMIUM, 2013/09/05 18:29:10]

nit: add const

[damienv1, 2013/09/09 23:29:45]

Done.

+
+  PidType pid_type() { return pid_type_; }

[acolwell GONE FROM CHROMIUM, 2013/09/05 18:29:10]

nit: add const here and in any other similar accessors that I may have missed.

[damienv1, 2013/09/09 23:29:45]

Done.

+
+ private:
+  void ResetState();
+
+  int pid_;
+  PidType pid_type_;
+  scoped_ptr<Mpeg2TsSectionParser> section_parser_;
+
+  bool enable_;
+
+  int continuity_counter_;
+};
+
+PidState::PidState(int pid, PidType pid_type,
+                   Mpeg2TsSectionParser* section_parser)
+  : pid_(pid),
+    pid_type_(pid_type),
+    section_parser_(section_parser),
+    enable_(false),
+    continuity_counter_(-1) {
+  DCHECK(section_parser);
+}
+
+bool PidState::PushTsPacket(TsPacket* ts_packet) {
+  DCHECK_EQ(ts_packet->pid(), pid_);
+
+  // The current PID is not part of the PID filter,
+  // just discard the incoming TS packet.
+  if (!enable_)
+    return true;
+
+  int expected_continuity_counter = (continuity_counter_ + 1) % 16;
+  if (continuity_counter_ >= 0 &&
+      ts_packet->continuity_counter() != expected_continuity_counter) {
+    LOG(WARNING) << "TS discontinuity detected for pid: " << pid_;
+    return false;
+  }
+
+  bool parse_result = section_parser_->Parse(

[acolwell GONE FROM CHROMIUM, 2013/09/05 18:29:10]

nit: just return here since nothing else uses parse_result

[damienv1, 2013/09/09 23:29:45]

Done.

+      ts_packet->payload_unit_start_indicator(),
+      ts_packet->GetPayload(),
+      ts_packet->GetPayloadSize());
+  return parse_result;
+}
+
+void PidState::Flush() {
+  section_parser_->Flush();
+  ResetState();
+}
+
+void PidState::Enable() {
+  enable_ = true;
+}
+
+void PidState::Disable() {
+  if (!enable_)
+    return;
+
+  ResetState();
+  enable_ = false;
+}
+
+bool PidState::IsEnabled() {
+  return enable_;
+}
+
+void PidState::ResetState() {
+  // TODO(damienv)
+  //section_parser_->ResetState();

[acolwell GONE FROM CHROMIUM, 2013/09/05 18:29:10]

Remove if this really isn't needed.

[damienv1, 2013/09/09 23:29:45]

Done: implemented.

+  continuity_counter_ = -1;
+}
+
+class Mpeg2TsStreamParser::AudioBufferWithConfig {
+ public:
+  scoped_refptr<StreamParserBuffer> buffer;
+  AudioDecoderConfig config;
+};
+
+class Mpeg2TsStreamParser::VideoBufferWithConfig {
+ public:
+  scoped_refptr<StreamParserBuffer> buffer;
+  VideoDecoderConfig config;
+};
+
+
+Mpeg2TsStreamParser::Mpeg2TsStreamParser()
+  : selected_audio_pid_(-1),
+    selected_video_pid_(-1),
+    is_initialized_(false),
+    segment_started_(false) {
+}
+
+Mpeg2TsStreamParser::~Mpeg2TsStreamParser() {
+  STLDeleteValues(&pids_);
+}
+
+void Mpeg2TsStreamParser::Init(
+    const InitCB& init_cb,
+    const NewConfigCB& config_cb,
+    const NewBuffersCB& new_buffers_cb,
+    const NewTextBuffersCB& text_cb,
+    const NeedKeyCB& need_key_cb,
+    const AddTextTrackCB& add_text_track_cb,
+    const NewMediaSegmentCB& new_segment_cb,
+    const base::Closure& end_of_segment_cb,
+    const LogCB& log_cb) {
+  DCHECK(!is_initialized_);
+  DCHECK(init_cb_.is_null());
+  DCHECK(!init_cb.is_null());
+  DCHECK(!config_cb.is_null());
+  DCHECK(!new_buffers_cb.is_null());
+  DCHECK(!need_key_cb.is_null());
+  DCHECK(!end_of_segment_cb.is_null());
+
+  init_cb_ = init_cb;
+  config_cb_ = config_cb;
+  new_buffers_cb_ = new_buffers_cb;
+  need_key_cb_ = need_key_cb;
+  new_segment_cb_ = new_segment_cb;
+  end_of_segment_cb_ = end_of_segment_cb;
+  log_cb_ = log_cb;
+}
+
+void Mpeg2TsStreamParser::Flush() {
+  DVLOG(1) << "Mpeg2TsStreamParser::Flush";
+
+  // Flush the buffers and reset the pids.
+  for (std::map<int, PidState*>::iterator it = pids_.begin();
+       it != pids_.end(); ++it) {
+    DVLOG(1) << "Flushing PID: " << it->first;
+    PidState* pid_state = it->second;
+    pid_state->Flush();
+    delete pid_state;
+  }
+  pids_.clear();
+  EmitRemainingBuffers();
+  DCHECK(audio_buffer_queue_.empty());
+  DCHECK(video_buffer_queue_.empty());
+  audio_buffer_queue_.clear();

[acolwell GONE FROM CHROMIUM, 2013/09/05 18:29:10]

nit: If we are DCHECKing that these queues are empty, then we shouldn't need to
clear() them here.

[damienv1, 2013/09/09 23:29:45]

Should now be clearer. Whether buffers are emitted or cleared depend on
|is_initialized|.

+  video_buffer_queue_.clear();
+
+  // End of the segment.
+  // Note: does not need to invoke |end_of_segment_cb_| since flushing the
+  // stream parser already involves the end of the current segment.
+  segment_started_ = false;
+
+  // Remove any bytes left in the TS buffer.
+  // (i.e. any partial TS packet => less than 188 bytes).
+  ts_byte_queue_.Reset();
+
+  // Reset the selected PIDs.
+  selected_audio_pid_ = -1;
+  selected_video_pid_ = -1;
+
+  // Reset the audio and video configs.
+  audio_config_ = AudioDecoderConfig();
+  video_config_ = VideoDecoderConfig();
+  last_audio_config_ = AudioDecoderConfig();
+  last_video_config_ = VideoDecoderConfig();
+}
+
+bool Mpeg2TsStreamParser::Parse(const uint8* buf, int size) {
+  DVLOG(1) << "Mpeg2TsStreamParser::Parse size=" << size;
+
+  // Add the data to the parser state.
+  ts_byte_queue_.Push(buf, size);
+  const uint8* ts_buffer = NULL;

[acolwell GONE FROM CHROMIUM, 2013/09/05 18:29:10]

nit: init not needed

[damienv1, 2013/09/09 23:29:45]

Done.

+  int ts_buffer_size = 0;
+  ts_byte_queue_.Peek(&ts_buffer, &ts_buffer_size);
+
+  int pos = 0;
+  int remaining_size = ts_buffer_size;
+  while (remaining_size >= TsPacket::kPacketSize) {
+    // Synchronization.
+    int skipped_bytes = TsPacket::Sync(&ts_buffer[pos], remaining_size);
+    if (skipped_bytes > 0) {
+      LOG(WARNING) << "Packet not aligned on a TS syncword:"

[acolwell GONE FROM CHROMIUM, 2013/09/05 18:29:10]

DVLOG here and everywhere else

[damienv1, 2013/09/09 23:29:45]

Done.

+                   << " skipped_bytes=" << skipped_bytes;
+      pos += skipped_bytes;
+      remaining_size -= skipped_bytes;
+      continue;
+    }
+
+    // Parse the TS header.
+    scoped_ptr<TsPacket> ts_packet(
+        TsPacket::Parse(&ts_buffer[pos], remaining_size));
+    if (!ts_packet) {
+      LOG(WARNING) << "Error: invalid TS packet";
+      pos += 1;
+      remaining_size -= 1;
+      continue;
+    }
+
+    DVLOG(LOG_LEVEL_TS)
+        << "Processing PID=" << ts_packet->pid()
+        << " start_unit=" << ts_packet->payload_unit_start_indicator();
+
+    // Parse the section.
+    std::map<int, PidState*>::iterator it = pids_.find(ts_packet->pid());
+    if (it == pids_.end() &&
+        ts_packet->pid() == Mpeg2TsSectionParser::kPidPat) {
+      // Create the PAT state here if needed.
+      scoped_ptr<Mpeg2TsPatParser> pat_section_parser(
+          new Mpeg2TsPatParser(
+              base::Bind(&Mpeg2TsStreamParser::RegisterPmt,
+                         base::Unretained(this))));
+      scoped_ptr<PidState> pat_pid_state(
+          new PidState(ts_packet->pid(), PidState::kPidPat,
+                       pat_section_parser.release()));
+      pat_pid_state->Enable();
+      it = pids_.insert(
+          std::pair<int, PidState*>(ts_packet->pid(),
+                                    pat_pid_state.release())).first;
+    }
+
+    if (it != pids_.end()) {
+      it->second->PushTsPacket(ts_packet.get());
+    } else {
+      LOG(WARNING) << "Ignoring TS packet for pid: " << ts_packet->pid();
+    }
+
+    // Go to the next packet.
+    pos += TsPacket::kPacketSize;
+    remaining_size -= TsPacket::kPacketSize;
+  }
+
+  // Keep only the possible incomplete trailing TS packet.
+  ts_byte_queue_.Pop(ts_buffer_size - remaining_size);
+
+  // Emit the A/V buffers that kept accumulating during TS parsing.
+  EmitRemainingBuffers();
+
+  return true;
+}
+
+void Mpeg2TsStreamParser::RegisterPmt(int program_number, int pmt_pid) {
+  DVLOG(1) << "RegisterPmt:"
+           << " program_number=" << program_number
+           << " pmt_pid=" << pmt_pid;
+
+  // Only one TS program is allowed. Ignore the incoming program map table,
+  // if there is already one registered.
+  for (std::map<int, PidState*>::iterator it = pids_.begin();
+       it != pids_.end(); ++it) {
+    PidState* pid_state = it->second;
+    if (pid_state->pid_type() == PidState::kPidPmt) {
+      int pid = it->first;
+      LOG_IF(WARNING, pmt_pid != pid) << "More than one program is defined";
+      return;
+    }
+  }
+
+  // Create the PMT state here if needed.
+  DVLOG(1) << "Create a new PMT parser";
+  scoped_ptr<Mpeg2TsPmtParser> pmt_section_parser(
+      new Mpeg2TsPmtParser(
+          base::Bind(&Mpeg2TsStreamParser::RegisterPes,
+                     base::Unretained(this), pmt_pid)));
+  scoped_ptr<PidState> pmt_pid_state(
+      new PidState(pmt_pid, PidState::kPidPmt, pmt_section_parser.release()));
+  pmt_pid_state->Enable();
+  pids_.insert(std::pair<int, PidState*>(pmt_pid, pmt_pid_state.release()));
+}
+
+void Mpeg2TsStreamParser::RegisterPes(int pmt_pid,
+                                      int pes_pid,
+                                      int stream_type) {
+  // TODO(damienv): check there is no mismatch if the entry already exists.
+  LOG(INFO) << "RegisterPes:"
+            << " pes_pid=" << pes_pid
+            << " stream_type=" << std::hex << stream_type << std::dec;
+  std::map<int, PidState*>::iterator it = pids_.find(pes_pid);
+  if (it != pids_.end())
+    return;
+
+  // Create a stream parser corresponding to the stream type.
+  bool is_audio = false;
+  scoped_ptr<EsParser> es_parser;
+  if (stream_type == kStreamTypeAVC) {
+    es_parser.reset(
+        new EsParserH264(
+            base::Bind(&Mpeg2TsStreamParser::OnVideoConfigChanged,
+                       base::Unretained(this),
+                       pes_pid),
+            base::Bind(&Mpeg2TsStreamParser::OnEmitVideoBuffer,
+                       base::Unretained(this),
+                       pes_pid)));
+  } else if (stream_type == kStreamTypeAAC) {
+    es_parser.reset(
+        new EsParserAdts(
+            base::Bind(&Mpeg2TsStreamParser::OnAudioConfigChanged,
+                       base::Unretained(this),
+                       pes_pid),
+            base::Bind(&Mpeg2TsStreamParser::OnEmitAudioBuffer,
+                       base::Unretained(this),
+                       pes_pid)));
+    is_audio = true;
+  } else {
+    return;
+  }
+
+  // Create the PES state here.
+  DVLOG(1) << "Create a new PES state";
+  scoped_ptr<Mpeg2TsPesParser> pes_section_parser(
+      new Mpeg2TsPesParser(es_parser.release()));
+  PidState::PidType pid_type =
+      is_audio ? PidState::kPidAudioPes : PidState::kPidVideoPes;
+  scoped_ptr<PidState> pes_pid_state(
+      new PidState(pes_pid, pid_type, pes_section_parser.release()));
+  pids_.insert(std::pair<int, PidState*>(pes_pid, pes_pid_state.release()));
+
+  // The pid filter must be updated.
+  UpdatePidFilter();
+}
+
+void Mpeg2TsStreamParser::UpdatePidFilter() {
+  // Applies the HLS rule to select the default audio/video PIDs:
+  // select the pid with the lowest PID.
+  // TODO(damienv): this can be changed when the StreamParser interface
+  // supports multiple audio/video streams.
+  std::map<int, PidState*>::iterator lowest_audio_pid = pids_.end();
+  std::map<int, PidState*>::iterator lowest_video_pid = pids_.end();
+  for (std::map<int, PidState*>::iterator it = pids_.begin();
+       it != pids_.end(); ++it) {
+    int pid = it->first;
+    PidState* pid_state = it->second;
+    if (pid_state->pid_type() == PidState::kPidAudioPes &&
+        ((lowest_audio_pid == pids_.end() || pid < lowest_audio_pid->first)))
+      lowest_audio_pid = it;
+    if (pid_state->pid_type() == PidState::kPidVideoPes &&
+        ((lowest_video_pid == pids_.end() || pid < lowest_video_pid->first)))
+      lowest_video_pid = it;
+  }
+
+  // Enable both the lowest audio and video PIDs.
+  if (lowest_audio_pid != pids_.end()) {
+    DVLOG(1) << "Enable audio pid: " << lowest_audio_pid->first;
+    lowest_audio_pid->second->Enable();
+    selected_audio_pid_ = lowest_audio_pid->first;
+  }
+  if (lowest_video_pid != pids_.end()) {
+    DVLOG(1) << "Enable video pid: " << lowest_audio_pid->first;
+    lowest_video_pid->second->Enable();
+    selected_video_pid_ = lowest_video_pid->first;
+  }
+
+  // Disable all the other audio and video PIDs.
+  for (std::map<int, PidState*>::iterator it = pids_.begin();
+       it != pids_.end(); ++it) {
+    PidState* pid_state = it->second;
+    if (it != lowest_audio_pid && it != lowest_video_pid &&
+        (pid_state->pid_type() == PidState::kPidAudioPes ||
+         pid_state->pid_type() == PidState::kPidVideoPes))
+      pid_state->Disable();
+  }
+}
+
+void Mpeg2TsStreamParser::OnVideoConfigChanged(
+    int pes_pid,
+    const VideoDecoderConfig& video_decoder_config) {
+  DVLOG(1) << "OnVideoConfigChanged for pid=" << pes_pid;
+  DCHECK_EQ(pes_pid, selected_video_pid_);
+
+  video_config_ = video_decoder_config;
+  FinishInitializationIfNeeded();
+}
+
+void Mpeg2TsStreamParser::OnAudioConfigChanged(
+    int pes_pid,
+    const AudioDecoderConfig& audio_decoder_config) {
+  DVLOG(1) << "OnAudioConfigChanged for pid=" << pes_pid;
+  DCHECK_EQ(pes_pid, selected_audio_pid_);
+
+  audio_config_ = audio_decoder_config;
+  FinishInitializationIfNeeded();
+}
+
+void Mpeg2TsStreamParser::FinishInitializationIfNeeded() {
+  // Nothing to be done if already initialized.
+  if (is_initialized_)
+    return;
+
+  // Initialization is done when both the audio decoder config
+  // and the video decoder config are known
+  // (for a stream with both audio and video).
+  if (selected_audio_pid_ > 0 && !audio_config_.IsValidConfig())
+    return;
+  if (selected_video_pid_ > 0 && !video_config_.IsValidConfig())
+    return;
+
+  // The audio and video decoder configs passed in the callback
+  // are the latest audio and video decoder configs.
+  // This might be different from the configs of the first audio and video
+  // buffer if we have a sequence like this one in the Mpeg2 TS stream:
+  // VConfigA VBuffer0 VBuffer1 VConfigB VBuffer2 AConfigA ABuffer0
+  // In this case, |audio_config_| corresponds to AConfigA
+  // and |video_config_| corresponds to VConfigB and not VConfigA.
+  // This does not matter since the callback will be invoked later before
+  // emitting any buffers and will thus overwrite the audio/video config.
+  config_cb_.Run(audio_config_, video_config_);
+
+  // For Mpeg2 TS, the duration is not known.
+  DVLOG(1) << "Mpeg2TS stream parser initialization done";
+  init_cb_.Run(true, kInfiniteDuration());
+  is_initialized_ = true;
+}
+
+void Mpeg2TsStreamParser::OnEmitAudioBuffer(
+    int pes_pid,
+    scoped_refptr<StreamParserBuffer> stream_parser_buffer) {
+  DCHECK_EQ(pes_pid, selected_audio_pid_);
+
+  DVLOG(LOG_LEVEL_ES)
+      << "OnEmitAudioBuffer: "
+      << " size="
+      << stream_parser_buffer->data_size()
+      << " dts="
+      << stream_parser_buffer->GetDecodeTimestamp().InMilliseconds()
+      << " pts="
+      << stream_parser_buffer->timestamp().InMilliseconds();
+  stream_parser_buffer->set_timestamp(
+      stream_parser_buffer->timestamp() - time_offset_);
+  stream_parser_buffer->SetDecodeTimestamp(
+      stream_parser_buffer->GetDecodeTimestamp() - time_offset_);
+
+  AudioBufferWithConfig audio_buffer_with_config;
+  audio_buffer_with_config.buffer = stream_parser_buffer;
+  audio_buffer_with_config.config = audio_config_;

[acolwell GONE FROM CHROMIUM, 2013/09/05 18:29:10]

You shouldn't need to do this. Just emit the pending buffers when the new config
arrives. Then you don't have to make a copy of the config for every buffer or
keep track of which configs are associated with what buffers.

[damienv1, 2013/09/09 23:29:45]

That was my initial implementation. However, I switched to keeping a
configuration for each buffer because you can have some special cases before
initialization. 
You can't emit any buffer before initialization is done. However, nothing is
preventing several video configuration changes before initialization is
completed.

+  audio_buffer_queue_.push_back(audio_buffer_with_config);
+}
+
+void Mpeg2TsStreamParser::OnEmitVideoBuffer(
+    int pes_pid,
+    scoped_refptr<StreamParserBuffer> stream_parser_buffer) {
+  DCHECK_EQ(pes_pid, selected_video_pid_);
+
+  DVLOG(LOG_LEVEL_ES)
+      << "OnEmitVideoBuffer"
+      << " size="
+      << stream_parser_buffer->data_size()
+      << " dts="
+      << stream_parser_buffer->GetDecodeTimestamp().InMilliseconds()
+      << " pts="
+      << stream_parser_buffer->timestamp().InMilliseconds()
+      << " IsKeyframe="
+      << stream_parser_buffer->IsKeyframe();
+  stream_parser_buffer->set_timestamp(
+      stream_parser_buffer->timestamp() - time_offset_);
+  stream_parser_buffer->SetDecodeTimestamp(
+      stream_parser_buffer->GetDecodeTimestamp() - time_offset_);
+
+  VideoBufferWithConfig video_buffer_with_config;
+  video_buffer_with_config.buffer = stream_parser_buffer;
+  video_buffer_with_config.config = video_config_;
+  video_buffer_queue_.push_back(video_buffer_with_config);
+}
+
+void Mpeg2TsStreamParser::EmitRemainingBuffers() {
+  DVLOG(LOG_LEVEL_ES) << "Mpeg2TsStreamParser::EmitRemainingBuffers";
+  if (!is_initialized_)
+    return;
+
+  while (!audio_buffer_queue_.empty())
+    EmitAudioBuffers();
+
+  while (!video_buffer_queue_.empty())
+    EmitVideoBuffers();
+}
+
+void Mpeg2TsStreamParser::EmitAudioBuffers() {
+  DCHECK(!audio_buffer_queue_.empty());
+
+  AudioDecoderConfig audio_config = audio_buffer_queue_.front().config;
+
+  if (!segment_started_) {
+    DVLOG(1) << "Starting a new segment";
+    segment_started_ = true;
+    new_segment_cb_.Run();
+  }
+
+  if (!audio_config.Matches(last_audio_config_)) {
+    last_audio_config_ = audio_config;
+    config_cb_.Run(last_audio_config_, last_video_config_);
+  }
+
+  StreamParser::BufferQueue audio_queue;
+  StreamParser::BufferQueue video_queue;
+  while (!audio_buffer_queue_.empty() &&
+         audio_buffer_queue_.front().config.Matches(last_audio_config_)) {
+    audio_queue.push_back(audio_buffer_queue_.front().buffer);
+    audio_buffer_queue_.pop_front();
+  }
+  new_buffers_cb_.Run(audio_queue, video_queue);
+}
+
+void Mpeg2TsStreamParser::EmitVideoBuffers() {
+  DCHECK(!video_buffer_queue_.empty());
+
+  VideoDecoderConfig video_config = video_buffer_queue_.front().config;
+
+  if (!segment_started_) {
+    DVLOG(1) << "Starting a new segment";
+    segment_started_ = true;
+    new_segment_cb_.Run();
+  }
+
+  if (!video_config.Matches(last_video_config_)) {
+    last_video_config_ = video_config;
+    config_cb_.Run(last_audio_config_, last_video_config_);
+  }
+
+  StreamParser::BufferQueue audio_queue;
+  StreamParser::BufferQueue video_queue;
+  while (!video_buffer_queue_.empty() &&
+         video_buffer_queue_.front().config.Matches(last_video_config_)) {
+    video_queue.push_back(video_buffer_queue_.front().buffer);
+    video_buffer_queue_.pop_front();
+  }
+  new_buffers_cb_.Run(audio_queue, video_queue);
+}
+
+}  // namespace mpeg2ts
+}  // namespace media