feat: dash event msg box handling (#23)

include/packager/live_packager.h

@@ -122,6 +122,10 @@ struct LiveConfig {
   /// Decryption parameters
   std::vector<uint8_t> decryption_key;
   std::vector<uint8_t> decryption_key_id;
+
+  /// Flag used to enable parsing of EMSG (Event Message) boxes during fmp4
+  /// parsing, and writing EMSG box data to output segments.
+  bool emsg_processing = false;
 };
 
 class LivePackager {

include/packager/live_packager_export.h

@@ -64,6 +64,10 @@ typedef struct LivePackagerConfig {
   /// output format is either VTT in MP4 or TTML in MP4.
   int64_t timed_text_decode_time;
 
+  /// Flag used to enable parsing of EMSG (Event Message) boxes during fmp4
+  /// parsing, and writing EMSG box data to output segments.
+  bool emsg_processing;
+
   /// Decryption parameters
   bool enable_decryption;
   uint8_t decryption_key[KEY_SIZE];

include/packager/packager.h

@@ -85,6 +85,10 @@ struct PackagingParams {
   /// Flag used as a workaround in the case of header only input WEBVTT and the
   /// need to produce an output segment
   bool webvtt_header_only_output_segment = false;
+
+  /// Flag used to enable parsing of EMSG (Event Message) boxes during fmp4
+  /// parsing, and writing EMSG box data to output segments.
+  bool emsg_processing = false;
 };
 
 /// Defines a single input/output stream.

packager/app/muxer_factory.cc

@@ -15,6 +15,7 @@
 #include <packager/media/formats/webm/webm_muxer.h>
 #include <packager/media/formats/webvtt/webvtt_muxer.h>
 #include <packager/packager.h>
+#include "packager/media/formats/mp4/dash_event_message_handler.h"
 
 namespace shaka {
 namespace media {
@@ -30,7 +31,8 @@ MuxerFactory::MuxerFactory(const PackagingParams& packaging_params)
 
 std::shared_ptr<Muxer> MuxerFactory::CreateMuxer(
     MediaContainerName output_format,
-    const StreamDescriptor& stream) {
+    const StreamDescriptor& stream,
+    std::shared_ptr<mp4::DashEventMessageHandler> dash_handler) {
   MuxerOptions options;
   options.mp4_params = mp4_params_;
   options.transport_stream_timestamp_offset_ms =
@@ -68,6 +70,8 @@ std::shared_ptr<Muxer> MuxerFactory::CreateMuxer(
         break;
       }
       muxer = std::make_shared<mp4::MP4Muxer>(options);
+      dynamic_cast<mp4::MP4Muxer*>(muxer.get())
+          ->SetDashEventMessageHandler(dash_handler);
       break;
     default:
       LOG(ERROR) << "Cannot support muxing to " << output_format;

packager/app/muxer_factory.h

@@ -13,6 +13,7 @@
 #include <packager/media/base/container_names.h>
 #include <packager/mp4_output_params.h>
 #include <packager/mpd/base/mpd_builder.h>
+#include "packager/media/formats/mp4/dash_event_message_handler.h"
 
 namespace shaka {
 struct PackagingParams;
@@ -32,8 +33,10 @@ class MuxerFactory {
 
   /// Create a new muxer using the factory's settings for the given
   /// stream.
-  std::shared_ptr<Muxer> CreateMuxer(MediaContainerName output_format,
-                                     const StreamDescriptor& stream);
+  std::shared_ptr<Muxer> CreateMuxer(
+      MediaContainerName output_format,
+      const StreamDescriptor& stream,
+      std::shared_ptr<mp4::DashEventMessageHandler> dash_handler);
 
   /// For testing, if you need to replace the clock that muxers work with
   /// this will replace the clock for all muxers created after this call.

packager/live_packager.cc

@@ -13,6 +13,7 @@
 
 #include <absl/log/globals.h>
 #include <absl/log/log.h>
+#include <absl/strings/escaping.h>
 #include <packager/chunking_params.h>
 #include <packager/file.h>
 
@@ -398,6 +399,7 @@ Status LivePackager::Package(const Segment& init_segment,
   packaging_params.enable_null_ts_packet_stuffing = true;
   packaging_params.cts_offset_adjustment =
       config_.format == LiveConfig::OutputFormat::TS;
+  packaging_params.emsg_processing = config_.emsg_processing;
 
   if (!config_.decryption_key.empty() && !config_.decryption_key_id.empty()) {
     DecryptionParams& decryption_params = packaging_params.decryption_params;

packager/live_packager_export.cc

@@ -21,6 +21,7 @@ LivePackager_t livepackager_new(LivePackagerConfig_t cfg) {
       .timed_text_decode_time = cfg.timed_text_decode_time,
       .decryption_key = {},
       .decryption_key_id = {},
+      .emsg_processing = cfg.emsg_processing,
   };
 
   if (cfg.protection_scheme != ENCRYPTION_SCHEME_NONE) {

packager/live_packager_test.cc

@@ -220,6 +220,11 @@ class MP4MediaParserTest {
     return samples_;
   }
 
+  const std::vector<std::shared_ptr<media::mp4::DASHEventMessageBox>>&
+  GetEmsgSamples() {
+    return emsg_samples_;
+  }
+
   bool Parse(const uint8_t* buf, size_t len) {
     // Use a memoryfile so we can read inputs directly without going to disk
     const std::string input_fname = "memory://file1";
@@ -275,11 +280,18 @@ class MP4MediaParserTest {
         std::bind(&MP4MediaParserTest::NewTextSampleF, this,
                   std::placeholders::_1, std::placeholders::_2),
         decryption_key_source);
+
+    parser_->SetEventMessageBoxCB(
+        [this](std::shared_ptr<media::mp4::DASHEventMessageBox> info) {
+          emsg_samples_.push_back(std::move(info));
+          return true;
+        });
   }
 
   std::unique_ptr<media::mp4::MP4MediaParser> parser_ =
       std::make_unique<media::mp4::MP4MediaParser>();
   std::vector<std::shared_ptr<media::MediaSample>> samples_;
+  std::vector<std::shared_ptr<media::mp4::DASHEventMessageBox>> emsg_samples_;
 };
 
 void CheckVideoInitSegment(const SegmentBuffer& buffer, media::FourCC format) {
@@ -1070,6 +1082,7 @@ struct LivePackagerReEncryptCase {
   LiveConfig::OutputFormat output_format;
   LiveConfig::TrackType track_type;
   const char* media_segment_format;
+  bool emsg_processing;
 };
 
 class LivePackagerTestReEncrypt
@@ -1117,6 +1130,16 @@ class LivePackagerTestReEncrypt
       std::make_unique<MP4MediaParserTest>(key_source_.get());
 };
 
+inline bool operator==(const media::mp4::DASHEventMessageBox& lhs,
+                       const media::mp4::DASHEventMessageBox& rhs) {
+  return std::tie(lhs.scheme_id_uri, lhs.value, lhs.timescale,
+                  lhs.presentation_time_delta, lhs.event_duration, lhs.id,
+                  lhs.message_data) ==
+         std::tie(rhs.scheme_id_uri, rhs.value, rhs.timescale,
+                  rhs.presentation_time_delta, rhs.event_duration, rhs.id,
+                  rhs.message_data);
+}
+
 TEST_P(LivePackagerTestReEncrypt, VerifyReEncryption) {
   std::vector<uint8_t> init_segment_buffer =
       ReadTestDataFile(GetParam().init_segment_name);
@@ -1137,12 +1160,13 @@ TEST_P(LivePackagerTestReEncrypt, VerifyReEncryption) {
 
     SegmentBuffer out;
     LiveConfig live_config;
-    live_config.segment_number = i;
+    live_config.segment_number = i + 1;
     live_config.format = GetParam().output_format;
     live_config.track_type = GetParam().track_type;
     live_config.protection_scheme = GetParam().encryption_scheme;
     live_config.decryption_key = HexStringToVector(kKeyHex);
     live_config.decryption_key_id = HexStringToVector(kKeyIdHex);
+    live_config.emsg_processing = GetParam().emsg_processing;
 
     SetupLivePackagerConfig(live_config);
 
@@ -1155,33 +1179,56 @@ TEST_P(LivePackagerTestReEncrypt, VerifyReEncryption) {
   auto expected_buf = ReadExpectedData();
   CHECK(parser_noenc_->Parse(expected_buf.data(), expected_buf.size()));
   auto& expected_samples = parser_noenc_->GetSamples();
+  auto& expected_emsg_samples = parser_noenc_->GetEmsgSamples();
 
   CHECK(parser_enc_->Parse(actual_buf.Data(), actual_buf.Size()));
   auto& actual_samples = parser_enc_->GetSamples();
+  auto& actual_emsg_samples = parser_enc_->GetEmsgSamples();
 
   CHECK_EQ(expected_samples.size(), actual_samples.size());
+  ASSERT_GT(expected_samples.size(), 0);
   CHECK(std::equal(
       expected_samples.begin(), expected_samples.end(), actual_samples.begin(),
       actual_samples.end(), [](const auto& s1, const auto& s2) {
         return s1->data_size() == s2->data_size() &&
                0 == memcmp(s1->data(), s2->data(), s1->data_size());
       }));
+
+  if (GetParam().emsg_processing) {
+    ASSERT_GT(expected_emsg_samples.size(), 0);
+    CHECK_EQ(expected_emsg_samples.size(), actual_emsg_samples.size());
+    CHECK(
+        std::equal(expected_emsg_samples.begin(), expected_emsg_samples.end(),
+                   actual_emsg_samples.begin(), actual_emsg_samples.end(),
+                   [](const auto& s1, const auto& s2) { return *s1 == *s2; }));
+  } else {
+    ASSERT_EQ(actual_emsg_samples.size(), 0);
+  }
 }
 
 INSTANTIATE_TEST_CASE_P(
     LivePackagerReEncryptTypes,
     LivePackagerTestReEncrypt,
     ::testing::Values(
-        // Verify decrypt FMP4 and re-encrypt to FMP4 with CENC encryption.
+        // Verify decrypt FMP4 and re-encrypt to FMP4 with CENC encryption,
+        // ENABLE processing EMSG.
         LivePackagerReEncryptCase{
             7, "encrypted/prd_data/init.mp4",
             LiveConfig::EncryptionScheme::CENC, LiveConfig::OutputFormat::FMP4,
-            LiveConfig::TrackType::VIDEO, "encrypted/prd_data/%05d.m4s"},
-        // Verify decrypt FMP4 and re-encrypt to FMP4 with CBCS encryption.
+            LiveConfig::TrackType::VIDEO, "encrypted/prd_data/%05d.m4s", true},
+        // Verify decrypt FMP4 and re-encrypt to FMP4 with CBCS encryption,
+        // ENABLE processing EMSG.
         LivePackagerReEncryptCase{
             7, "encrypted/prd_data/init.mp4",
             LiveConfig::EncryptionScheme::CBCS, LiveConfig::OutputFormat::FMP4,
-            LiveConfig::TrackType::VIDEO, "encrypted/prd_data/%05d.m4s"}));
+            LiveConfig::TrackType::VIDEO, "encrypted/prd_data/%05d.m4s", true},
+        // Verify decrypt FMP4 and re-encrypt to FMP4 with CBCS encryption,
+        // DISABLE processing EMSG
+        LivePackagerReEncryptCase{7, "encrypted/prd_data/init.mp4",
+                                  LiveConfig::EncryptionScheme::CBCS,
+                                  LiveConfig::OutputFormat::FMP4,
+                                  LiveConfig::TrackType::VIDEO,
+                                  "encrypted/prd_data/%05d.m4s", false}));
 
 struct TimedTextTestCase {
   const char* media_segment_format;

packager/media/base/fourccs.h

@@ -14,6 +14,7 @@ namespace media {
 enum FourCC : uint32_t {
   FOURCC_NULL = 0,
 
+  FOURCC_emsg = 0x656d7367,  // 'emsg'
   FOURCC_ID32 = 0x49443332,
   FOURCC_Head = 0x48656164,
   FOURCC_Opus = 0x4f707573,

packager/media/demuxer/demuxer.cc

@@ -73,7 +73,7 @@ bool GetStreamIndex(const std::string& stream_label, size_t* stream_index) {
   return true;
 }
 
-}
+}  // namespace
 
 namespace shaka {
 namespace media {
@@ -145,6 +145,11 @@ Status Demuxer::SetHandler(const std::string& stream_label,
   return MediaHandler::SetHandler(stream_index, std::move(handler));
 }
 
+void Demuxer::SetDashEventMessageHandler(
+    const std::shared_ptr<mp4::DashEventMessageHandler>& handler) {
+  dash_event_handler_ = handler;
+}
+
 void Demuxer::SetLanguageOverride(const std::string& stream_label,
                                   const std::string& language_override) {
   size_t stream_index = kInvalidStreamIndex;
@@ -185,6 +190,14 @@ Status Demuxer::InitializeParser() {
   switch (container_name_) {
     case CONTAINER_MOV:
       parser_.reset(new mp4::MP4MediaParser(cts_offset_adjustment_));
+      dynamic_cast<mp4::MP4MediaParser*>(parser_.get())
+          ->SetEventMessageBoxCB(
+              [this](std::shared_ptr<mp4::DASHEventMessageBox> emsg_box_info) {
+                if (dash_event_handler_) {
+                  dash_event_handler_->OnDashEvent(std::move(emsg_box_info));
+                }
+                return true;
+              });
       break;
     case CONTAINER_MPEG2TS:
       parser_.reset(new mp2t::Mp2tMediaParser());

packager/media/demuxer/demuxer.h

@@ -13,6 +13,7 @@
 
 #include <packager/macros/classes.h>
 #include <packager/media/base/container_names.h>
+#include <packager/media/formats/mp4/dash_event_message_handler.h>
 #include <packager/media/origin/origin_handler.h>
 #include <packager/status.h>
 
@@ -63,6 +64,9 @@ class Demuxer : public OriginHandler {
   Status SetHandler(const std::string& stream_label,
                     std::shared_ptr<MediaHandler> handler);
 
+  void SetDashEventMessageHandler(
+      const std::shared_ptr<mp4::DashEventMessageHandler>& handler);
+
   /// Override the language in the specified stream. If the specified stream is
   /// a video stream or invalid, this function is a no-op.
   /// @param stream_label can be 'audio', 'video', or stream number (zero
@@ -128,7 +132,6 @@ class Demuxer : public OriginHandler {
                            std::shared_ptr<MediaSample> sample);
   bool NewTextSampleEvent(uint32_t track_id,
                           std::shared_ptr<TextSample> sample);
-  // Helper function to push the sample to corresponding stream.
   bool PushMediaSample(uint32_t track_id, std::shared_ptr<MediaSample> sample);
   bool PushTextSample(uint32_t track_id, std::shared_ptr<TextSample> sample);
 
@@ -162,6 +165,8 @@ class Demuxer : public OriginHandler {
   // need to produce an output segment
   bool webvtt_header_only_output_segment_ = false;
   Status init_event_status_;
+
+  std::shared_ptr<mp4::DashEventMessageHandler> dash_event_handler_;
 };
 
 }  // namespace media

packager/media/formats/mp4/CMakeLists.txt

@@ -39,6 +39,8 @@ add_library(mp4 STATIC
   track_run_iterator.h
   mp4_init_muxer.h
   mp4_init_muxer.cc
+  dash_event_message_handler.cc
+  dash_event_message_handler.h
 )
 target_link_libraries(mp4
   media_base

packager/media/formats/mp4/box_definitions.cc

@@ -6,6 +6,7 @@
 
 #include <algorithm>
 #include <limits>
+#include <utility>
 #include <vector>
 
 #include <absl/flags/flag.h>
@@ -3088,6 +3089,36 @@ size_t VTTCueBox::ComputeSizeInternal() {
          cue_payload.ComputeSize();
 }
 
+DASHEventMessageBox::DASHEventMessageBox() = default;
+DASHEventMessageBox::~DASHEventMessageBox() = default;
+
+FourCC DASHEventMessageBox::BoxType() const {
+  return FOURCC_emsg;
+}
+
+size_t DASHEventMessageBox::ComputeSizeInternal() {
+  const static uint8_t kNull_sz = 1;
+  const static uint8_t kNumUint32s_v0 = 4;
+  size_t size = HeaderSize() + scheme_id_uri.size() + kNull_sz + value.size() +
+                kNull_sz + (kNumUint32s_v0 * sizeof(uint32_t)) +
+                message_data.size();
+  return size;
+}
+
+bool DASHEventMessageBox::ReadWriteInternal(BoxBuffer* buffer) {
+  uint8_t num_bytes = (version == 1) ? sizeof(uint64_t) : sizeof(uint32_t);
+  RCHECK(
+      ReadWriteHeaderInternal(buffer) &&
+      buffer->ReadWriteCString(&scheme_id_uri) &&
+      buffer->ReadWriteCString(&value) && buffer->ReadWriteUInt32(&timescale) &&
+      buffer->ReadWriteUInt64NBytes(&presentation_time_delta, num_bytes) &&
+      buffer->ReadWriteUInt32(&event_duration) && buffer->ReadWriteUInt32(&id));
+
+  size_t size = buffer->Reading() ? buffer->BytesLeft() : message_data.size();
+  RCHECK(buffer->ReadWriteVector(&message_data, size));
+  return true;
+}
+
 }  // namespace mp4
 }  // namespace media
 }  // namespace shaka

packager/media/formats/mp4/box_definitions.h

@@ -246,6 +246,21 @@ struct ID3v2 : FullBox {
   std::vector<uint8_t> id3v2_data;
 };
 
+// DASHEventMessageBox v0 and v1
+struct DASHEventMessageBox : FullBox {
+  DECLARE_BOX_METHODS(DASHEventMessageBox);
+
+  inline uint32_t GetID() const { return id; }
+
+  std::string scheme_id_uri;
+  std::string value;
+  uint32_t timescale = 0u;
+  uint64_t presentation_time_delta = 0u;
+  uint32_t event_duration = 0u;
+  uint32_t id = 0u;
+  std::vector<uint8_t> message_data;
+};
+
 struct Metadata : FullBox {
   DECLARE_BOX_METHODS(Metadata);