From 30450df76569ae4a443f832884112845b46eaac0 Mon Sep 17 00:00:00 2001
From: Lee Fordyce <leefordyce@gmail.com>
Date: Tue, 23 Apr 2024 09:42:55 -0600
Subject: [PATCH] feat: dash event msg box handling (#23)

---
 include/packager/live_packager.h              |   4 ++
 include/packager/live_packager_export.h       |   4 ++
 include/packager/packager.h                   |   4 ++
 packager/app/muxer_factory.cc                 |   6 +-
 packager/app/muxer_factory.h                  |   7 ++-
 packager/live_packager.cc                     |   2 +
 packager/live_packager_export.cc              |   1 +
 packager/live_packager_test.cc                |  57 ++++++++++++++++--
 packager/media/base/fourccs.h                 |   1 +
 packager/media/demuxer/demuxer.cc             |  15 ++++-
 packager/media/demuxer/demuxer.h              |   7 ++-
 packager/media/formats/mp4/CMakeLists.txt     |   2 +
 packager/media/formats/mp4/box_definitions.cc |  31 ++++++++++
 packager/media/formats/mp4/box_definitions.h  |  15 +++++
 .../formats/mp4/dash_event_message_handler.cc |  22 +++++++
 .../formats/mp4/dash_event_message_handler.h  |  35 +++++++++++
 .../media/formats/mp4/mp4_media_parser.cc     |  33 +++++++---
 packager/media/formats/mp4/mp4_media_parser.h |  11 +++-
 packager/media/formats/mp4/mp4_muxer.cc       |   7 +++
 packager/media/formats/mp4/mp4_muxer.h        |   6 ++
 .../formats/mp4/multi_segment_segmenter.cc    |  10 +++
 .../formats/mp4/multi_segment_segmenter.h     |   5 ++
 .../encrypted/prd_data/decrypt/fmp4/00001.m4s | Bin 1288035 -> 1288167 bytes
 .../encrypted/prd_data/decrypt/fmp4/00002.m4s | Bin 1301124 -> 1301388 bytes
 .../encrypted/prd_data/decrypt/fmp4/00003.m4s | Bin 1073178 -> 1073442 bytes
 .../encrypted/prd_data/decrypt/fmp4/00004.m4s | Bin 987126 -> 987258 bytes
 .../encrypted/prd_data/decrypt/fmp4/00005.m4s | Bin 1136475 -> 1136739 bytes
 .../encrypted/prd_data/decrypt/fmp4/00006.m4s | Bin 978163 -> 978427 bytes
 .../encrypted/prd_data/decrypt/fmp4/00007.m4s | Bin 33578 -> 33710 bytes
 packager/packager.cc                          |  27 +++++++--
 30 files changed, 285 insertions(+), 27 deletions(-)
 create mode 100644 packager/media/formats/mp4/dash_event_message_handler.cc
 create mode 100644 packager/media/formats/mp4/dash_event_message_handler.h

diff --git a/include/packager/live_packager.h b/include/packager/live_packager.h
index b75bd93eaa1..eddbd18ba82 100644
--- a/include/packager/live_packager.h
+++ b/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 {
diff --git a/include/packager/live_packager_export.h b/include/packager/live_packager_export.h
index a2d978b8477..95f18dcab77 100644
--- a/include/packager/live_packager_export.h
+++ b/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];
diff --git a/include/packager/packager.h b/include/packager/packager.h
index 06ffbc28a12..71733735c12 100644
--- a/include/packager/packager.h
+++ b/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.
diff --git a/packager/app/muxer_factory.cc b/packager/app/muxer_factory.cc
index e57b434b813..d1aa3f27147 100644
--- a/packager/app/muxer_factory.cc
+++ b/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;
diff --git a/packager/app/muxer_factory.h b/packager/app/muxer_factory.h
index e4b6fa96df7..0098d141ffa 100644
--- a/packager/app/muxer_factory.h
+++ b/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.
diff --git a/packager/live_packager.cc b/packager/live_packager.cc
index 34cf1135f08..e06aceb9490 100644
--- a/packager/live_packager.cc
+++ b/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;
diff --git a/packager/live_packager_export.cc b/packager/live_packager_export.cc
index cd62af38152..834c0c4ef38 100644
--- a/packager/live_packager_export.cc
+++ b/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) {
diff --git a/packager/live_packager_test.cc b/packager/live_packager_test.cc
index 858248d152d..af037146ab1 100644
--- a/packager/live_packager_test.cc
+++ b/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;
diff --git a/packager/media/base/fourccs.h b/packager/media/base/fourccs.h
index d7a73fe47e7..dd6ec2de21c 100644
--- a/packager/media/base/fourccs.h
+++ b/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,
diff --git a/packager/media/demuxer/demuxer.cc b/packager/media/demuxer/demuxer.cc
index d975d10325d..40284d5eebe 100644
--- a/packager/media/demuxer/demuxer.cc
+++ b/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());
diff --git a/packager/media/demuxer/demuxer.h b/packager/media/demuxer/demuxer.h
index ba4b3469b7c..facd90333fa 100644
--- a/packager/media/demuxer/demuxer.h
+++ b/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
diff --git a/packager/media/formats/mp4/CMakeLists.txt b/packager/media/formats/mp4/CMakeLists.txt
index 0c3ec135a8c..67e8bdd7b94 100644
--- a/packager/media/formats/mp4/CMakeLists.txt
+++ b/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
diff --git a/packager/media/formats/mp4/box_definitions.cc b/packager/media/formats/mp4/box_definitions.cc
index 4972ca19156..e200593c659 100644
--- a/packager/media/formats/mp4/box_definitions.cc
+++ b/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
diff --git a/packager/media/formats/mp4/box_definitions.h b/packager/media/formats/mp4/box_definitions.h
index 184c8b221de..98dec4caeff 100644
--- a/packager/media/formats/mp4/box_definitions.h
+++ b/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);
 
diff --git a/packager/media/formats/mp4/dash_event_message_handler.cc b/packager/media/formats/mp4/dash_event_message_handler.cc
new file mode 100644
index 00000000000..4588547fcc4
--- /dev/null
+++ b/packager/media/formats/mp4/dash_event_message_handler.cc
@@ -0,0 +1,22 @@
+#include <packager/media/formats/mp4/dash_event_message_handler.h>
+
+namespace shaka {
+namespace media {
+namespace mp4 {
+
+DashEventMessageHandler::DashEventMessageHandler() = default;
+DashEventMessageHandler::~DashEventMessageHandler() = default;
+void DashEventMessageHandler::FlushEventMessages(BufferWriter* writer) {
+  for (const auto& event : dash_event_message_queue_) {
+    event->Write(writer);
+  }
+  dash_event_message_queue_.clear();
+}
+void DashEventMessageHandler::OnDashEvent(
+    std::shared_ptr<DASHEventMessageBox> emsg_box_info) {
+  dash_event_message_queue_.push_back(std::move(emsg_box_info));
+}
+
+}  // namespace mp4
+}  // namespace media
+}  // namespace shaka
diff --git a/packager/media/formats/mp4/dash_event_message_handler.h b/packager/media/formats/mp4/dash_event_message_handler.h
new file mode 100644
index 00000000000..4c9893734e9
--- /dev/null
+++ b/packager/media/formats/mp4/dash_event_message_handler.h
@@ -0,0 +1,35 @@
+#ifndef PACKAGER_MEDIA_FORMATS_MP4_EVENT_MESSAGE_HANDLER_H_
+#define PACKAGER_MEDIA_FORMATS_MP4_EVENT_MESSAGE_HANDLER_H_
+
+#include <packager/status.h>
+#include <deque>
+#include <functional>
+#include <memory>
+
+#include "box_definitions.h"
+
+namespace shaka {
+namespace media {
+namespace mp4 {
+
+class DashEventMessageHandler {
+ public:
+  DashEventMessageHandler();
+  ~DashEventMessageHandler();
+
+  void OnDashEvent(std::shared_ptr<DASHEventMessageBox> emsg_box_info);
+
+  void FlushEventMessages(BufferWriter* writer);
+
+  DashEventMessageHandler(const DashEventMessageHandler&) = delete;
+
+  DashEventMessageHandler& operator=(const DashEventMessageHandler&) = delete;
+
+ private:
+  std::deque<std::shared_ptr<DASHEventMessageBox>> dash_event_message_queue_;
+};
+}  // namespace mp4
+}  // namespace media
+}  // namespace shaka
+
+#endif  // PACKAGER_MEDIA_FORMATS_MP4_EVENT_MESSAGE_HANDLER_H_
diff --git a/packager/media/formats/mp4/mp4_media_parser.cc b/packager/media/formats/mp4/mp4_media_parser.cc
index cc2aed57411..999be7883dc 100644
--- a/packager/media/formats/mp4/mp4_media_parser.cc
+++ b/packager/media/formats/mp4/mp4_media_parser.cc
@@ -186,6 +186,11 @@ MP4MediaParser::MP4MediaParser(bool cts_offset_adjustment)
 
 MP4MediaParser::~MP4MediaParser() {}
 
+void MP4MediaParser::SetEventMessageBoxCB(
+    const DASHEventMessageBoxCB& event_message_cb) {
+  event_message_cb_ = event_message_cb;
+}
+
 void MP4MediaParser::Init(const InitCB& init_cb,
                           const NewMediaSampleCB& new_media_sample_cb,
                           const NewTextSampleCB& new_text_sample_cb,
@@ -260,7 +265,7 @@ bool MP4MediaParser::LoadMoov(const std::string& file_path) {
   }
   if (!file->Seek(0)) {
     LOG(WARNING) << "Filesystem does not support seeking on file '" << file_path
-               << "'";
+                 << "'";
     return false;
   }
 
@@ -315,7 +320,7 @@ bool MP4MediaParser::LoadMoov(const std::string& file_path) {
       }
       queue_.Reset();  // So that we don't need to adjust data offsets.
       mdat_tail_ = 0;  // So it will skip boxes until mdat.
-      break;  // Done.
+      break;           // Done.
     }
     file_position += box_size;
     if (!file->Seek(file_position)) {
@@ -361,6 +366,8 @@ bool MP4MediaParser::ParseBox(bool* err) {
 
   if (reader->type() == FOURCC_moov) {
     *err = !ParseMoov(reader.get());
+  } else if (reader->type() == FOURCC_emsg) {
+    *err = !ParseEmsg(reader.get());
   } else if (reader->type() == FOURCC_moof) {
     moof_head_ = queue_.head();
     *err = !ParseMoof(reader.get());
@@ -378,6 +385,15 @@ bool MP4MediaParser::ParseBox(bool* err) {
   return !(*err);
 }
 
+bool MP4MediaParser::ParseEmsg(BoxReader* reader) {
+  if (event_message_cb_) {
+    auto emsg_box = std::make_shared<DASHEventMessageBox>();
+    RCHECK(emsg_box->Parse(reader));
+    event_message_cb_(emsg_box);
+  }
+  return true;
+}
+
 bool MP4MediaParser::ParseMoov(BoxReader* reader) {
   if (moov_)
     return true;  // Already parsed the 'moov' box.
@@ -399,8 +415,7 @@ bool MP4MediaParser::ParseMoov(BoxReader* reader) {
     } else if (moov_->extends.header.fragment_duration > 0) {
       DCHECK(moov_->header.timescale != 0);
       duration = Rescale(moov_->extends.header.fragment_duration,
-                         moov_->header.timescale,
-                         timescale);
+                         moov_->header.timescale, timescale);
     } else if (moov_->header.duration > 0 &&
                moov_->header.duration != std::numeric_limits<uint64_t>::max()) {
       DCHECK(moov_->header.timescale != 0);
@@ -840,8 +855,8 @@ bool MP4MediaParser::EnqueueSample(bool* err) {
   queue_.PeekAt(sample_offset, &buf, &buf_size);
   if (buf_size < runs_->sample_size()) {
     if (sample_offset < queue_.head()) {
-      LOG(ERROR) << "Incorrect sample offset " << sample_offset
-                 << " < " << queue_.head();
+      LOG(ERROR) << "Incorrect sample offset " << sample_offset << " < "
+                 << queue_.head();
       *err = true;
     }
     return false;
@@ -891,10 +906,8 @@ bool MP4MediaParser::EnqueueSample(bool* err) {
   stream_sample->set_duration(runs_->duration());
 
   DVLOG(3) << "Pushing frame: "
-           << ", key=" << runs_->is_keyframe()
-           << ", dur=" << runs_->duration()
-           << ", dts=" << runs_->dts()
-           << ", cts=" << runs_->cts()
+           << ", key=" << runs_->is_keyframe() << ", dur=" << runs_->duration()
+           << ", dts=" << runs_->dts() << ", cts=" << runs_->cts()
            << ", size=" << runs_->sample_size();
 
   if (!new_sample_cb_(runs_->track_id(), stream_sample)) {
diff --git a/packager/media/formats/mp4/mp4_media_parser.h b/packager/media/formats/mp4/mp4_media_parser.h
index 619557273d9..817ef208505 100644
--- a/packager/media/formats/mp4/mp4_media_parser.h
+++ b/packager/media/formats/mp4/mp4_media_parser.h
@@ -16,6 +16,7 @@
 #include <packager/media/base/decryptor_source.h>
 #include <packager/media/base/media_parser.h>
 #include <packager/media/base/offset_byte_queue.h>
+#include <packager/media/formats/mp4/box_definitions.h>
 
 namespace shaka {
 namespace media {
@@ -28,7 +29,7 @@ struct ProtectionSystemSpecificHeader;
 
 class MP4MediaParser : public MediaParser {
  public:
-  MP4MediaParser(bool cts_offset_adjustment = false);
+  explicit MP4MediaParser(bool cts_offset_adjustment = false);
   ~MP4MediaParser() override;
 
   /// @name MediaParser implementation overrides.
@@ -49,6 +50,12 @@ class MP4MediaParser : public MediaParser {
   /// @return true if successful, false otherwise.
   bool LoadMoov(const std::string& file_path);
 
+  typedef std::function<bool(
+      std::shared_ptr<mp4::DASHEventMessageBox> emsg_box_info)>
+      DASHEventMessageBoxCB;
+
+  void SetEventMessageBoxCB(const DASHEventMessageBoxCB& event_message_cb);
+
  private:
   enum State {
     kWaitingForInit,
@@ -60,6 +67,7 @@ class MP4MediaParser : public MediaParser {
   bool ParseBox(bool* err);
   bool ParseMoov(mp4::BoxReader* reader);
   bool ParseMoof(mp4::BoxReader* reader);
+  bool ParseEmsg(mp4::BoxReader* reader);
 
   bool FetchKeysIfNecessary(
       const std::vector<ProtectionSystemSpecificHeader>& headers);
@@ -83,6 +91,7 @@ class MP4MediaParser : public MediaParser {
   State state_;
   InitCB init_cb_;
   NewMediaSampleCB new_sample_cb_;
+  DASHEventMessageBoxCB event_message_cb_;
   KeySource* decryption_key_source_;
   std::unique_ptr<DecryptorSource> decryptor_source_;
 
diff --git a/packager/media/formats/mp4/mp4_muxer.cc b/packager/media/formats/mp4/mp4_muxer.cc
index 2a6ea6189e8..013e58f8fc4 100644
--- a/packager/media/formats/mp4/mp4_muxer.cc
+++ b/packager/media/formats/mp4/mp4_muxer.cc
@@ -169,6 +169,11 @@ Status MP4Muxer::InitializeMuxer() {
   return Status::OK;
 }
 
+void MP4Muxer::SetDashEventMessageHandler(
+    const std::shared_ptr<mp4::DashEventMessageHandler>& emsg_handler) {
+  emsg_handler_ = emsg_handler;
+}
+
 Status MP4Muxer::Finalize() {
   // This happens on streams that are not initialized, i.e. not going through
   // DelayInitializeMuxer, which can only happen if there are no samples from
@@ -312,6 +317,8 @@ Status MP4Muxer::DelayInitializeMuxer() {
   } else {
     segmenter_.reset(
         new MultiSegmentSegmenter(options(), std::move(ftyp), std::move(moov)));
+    dynamic_cast<MultiSegmentSegmenter*>(segmenter_.get())
+        ->SetDashEventMessageHandler(emsg_handler_);
   }
 
   const Status segmenter_initialized =
diff --git a/packager/media/formats/mp4/mp4_muxer.h b/packager/media/formats/mp4/mp4_muxer.h
index ab0ea114c34..e0b03101784 100644
--- a/packager/media/formats/mp4/mp4_muxer.h
+++ b/packager/media/formats/mp4/mp4_muxer.h
@@ -12,6 +12,7 @@
 
 #include <packager/macros/classes.h>
 #include <packager/media/base/muxer.h>
+#include "dash_event_message_handler.h"
 
 namespace shaka {
 namespace media {
@@ -36,6 +37,9 @@ class MP4Muxer : public Muxer {
   explicit MP4Muxer(const MuxerOptions& options);
   ~MP4Muxer() override;
 
+  void SetDashEventMessageHandler(
+      const std::shared_ptr<mp4::DashEventMessageHandler>& emsg_handler);
+
  protected:
   Status DelayInitializeMuxer();
 
@@ -76,6 +80,8 @@ class MP4Muxer : public Muxer {
 
   std::unique_ptr<Segmenter> segmenter_;
 
+  std::shared_ptr<mp4::DashEventMessageHandler> emsg_handler_;
+
   DISALLOW_COPY_AND_ASSIGN(MP4Muxer);
 };
 
diff --git a/packager/media/formats/mp4/multi_segment_segmenter.cc b/packager/media/formats/mp4/multi_segment_segmenter.cc
index 5381a308e75..2211ad50810 100644
--- a/packager/media/formats/mp4/multi_segment_segmenter.cc
+++ b/packager/media/formats/mp4/multi_segment_segmenter.cc
@@ -22,6 +22,7 @@
 #include <packager/media/event/muxer_listener.h>
 #include <packager/media/formats/mp4/box_definitions.h>
 #include <packager/media/formats/mp4/key_frame_info.h>
+#include "dash_event_message_handler.h"
 
 namespace shaka {
 namespace media {
@@ -127,6 +128,10 @@ Status MultiSegmentSegmenter::WriteSegment() {
   if (options().mp4_params.generate_sidx_in_media_segments)
     sidx()->Write(buffer.get());
 
+  if (emsg_handler_) {
+    emsg_handler_->FlushEventMessages(buffer.get());
+  }
+
   const size_t segment_header_size = buffer->Size();
   const size_t segment_size = segment_header_size + fragment_buffer()->Size();
   DCHECK_NE(segment_size, 0u);
@@ -168,6 +173,11 @@ Status MultiSegmentSegmenter::WriteSegment() {
   return Status::OK;
 }
 
+void MultiSegmentSegmenter::SetDashEventMessageHandler(
+    const std::shared_ptr<mp4::DashEventMessageHandler>& handler) {
+  emsg_handler_ = handler;
+}
+
 }  // namespace mp4
 }  // namespace media
 }  // namespace shaka
diff --git a/packager/media/formats/mp4/multi_segment_segmenter.h b/packager/media/formats/mp4/multi_segment_segmenter.h
index ab19465d7d8..c82ac5ce9d7 100644
--- a/packager/media/formats/mp4/multi_segment_segmenter.h
+++ b/packager/media/formats/mp4/multi_segment_segmenter.h
@@ -9,6 +9,7 @@
 
 #include <packager/macros/classes.h>
 #include <packager/media/formats/mp4/segmenter.h>
+#include "dash_event_message_handler.h"
 
 namespace shaka {
 namespace media {
@@ -35,6 +36,9 @@ class MultiSegmentSegmenter : public Segmenter {
   std::vector<Range> GetSegmentRanges() override;
   /// @}
 
+  void SetDashEventMessageHandler(
+      const std::shared_ptr<mp4::DashEventMessageHandler>& handler);
+
  private:
   // Segmenter implementation overrides.
   Status DoInitialize() override;
@@ -47,6 +51,7 @@ class MultiSegmentSegmenter : public Segmenter {
 
   std::unique_ptr<SegmentType> styp_;
   uint32_t num_segments_;
+  std::shared_ptr<mp4::DashEventMessageHandler> emsg_handler_;
 
   DISALLOW_COPY_AND_ASSIGN(MultiSegmentSegmenter);
 };
diff --git a/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00001.m4s b/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00001.m4s
index 5c190292c1c79c026904cc06da249433dd68ecbf..9879bb337ab1d084a9c771d4ef50b23e0b1aa589 100644
GIT binary patch
delta 201
zcmaES&hPnozX@UW3@xd-#pw({P+nfHSCCU$lCM`%#$ah_2^6vciG#rZ{`oc;o-W2L
z3<^M=Z$OY|7?5^G)nt{NlbLN5Y3x-H<yeuU?-`mN=<XlonCcf1nP!=l7#?KenC|G9
z6rPiv>geg{k#1|tz`zVdjUlZejIAL|ts%^<AuO#StgRtzts(5KAsnqCoUI{Tts&g4
eAv~=iysaU8ts(raAp)%-f~_G!TSJ8TmjeKBYC5g}

delta 69
zcmV~$rxAca006-u$0zpnl70~9>4FMq1N>P8ZLs@2@x9}*?P7u{W|(7vC01BtgDrO0
PW55ANoN&hFat-GX*-#<$

diff --git a/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00002.m4s b/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00002.m4s
index b2e4ee647e0446c304c9b4681c1e83ed6638131e..38c7409f4d35687774a0a26c4ea64a236236bbe2 100644
GIT binary patch
delta 334
zcmZqK>ff{7e?nM2LrZFIaXJGKl$V$5737qb<m;7`F<4q!0)=cq;vn$9f4)tor;9NQ
zg94D}8xZ6f2Be))HCZL+WM*4M8haH)IacK8dxoY5y8A~trus!hrdehsh6kBAraQVN
zh390aI(j;Kq}$puFw`?JFak{?<cKK^Kt&)&WKr1>9AH;4FbFh;w1zOYhA_2;Ft>)V
zw1%*@hOo7Uu(yVAw1#lDhH$lpaJPo=w1)7uhVZq9@VABtw1x<_h6rs95jMC505Er6
AcK`qY

delta 89
zcmeC#?%%T2e?nMu9AkSNBM37AF*6Xe05K~NvjH(X5OV-AClGT1F*gwN05LBR^8qnG
S5DNgYAP@^}k7E?py9EGx@fYg=

diff --git a/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00003.m4s b/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00003.m4s
index 90ac8711ea3647b4dc2acf6f2af9d31f7b5dc0f3..7a461879ee96f60f5036017edccc2ef0be9783ca 100644
GIT binary patch
delta 325
zcmbPrz;V$b#|dHe3@xd-#pw({P+nfHSCCU$lCM`%#$ah_2^6vciG#rZ{`of9o-W2L
z3<^M=Z$OY|7?5^G)nt{NlbLN5Y3x-H<yeuU?-`mN=<XlonCcf1nP!=l7#?KenC|GE
z6rPiv>geg{k#1|tz);V?zz8&nfFl?+foi~x$f2?$IKZx8U{GibX$@g)4Pj~xVQvj!
rX$@g*4Pk2yVQ&rLXbs_P4dH4H;cgA#X$|3R4dH7I;olk}a4a4Gp>JC&

delta 77
zcmZ2<$Z^&I#|dH0ag6P8j3CSe#LPg<0>rF9%m&2lK+FNeoIuP4#N0s41H`;Q%m>8$
L+v6Aoj>H21ejpV@

diff --git a/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00004.m4s b/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00004.m4s
index 633290d8342dae4ca0b56243ac9fcb5fc3e25fe6..464b248f542094b70eae07cd6884481b99d64767 100644
GIT binary patch
delta 189
zcmex1-{#i>n+aj{3@xd-#pw({P+nfHSCCU$lCM`%#$ah_2^6vciG#rZ{`oe!o-W2L
z3<^M=Z$OY|7?5^G)nt{NlbLN5Y3x-H<yeuU?-`mN=<XlonCcf1nP!=l7#?KenC=*m
z6rPiv>geg{k#1|tz`)GFz}Ohl8p7Bb!qghV+#15t8p7Hd!qytX-WtNu8p7Ec!qpnW
T-5SEv8p7Ke!nZYqe-S?biVZfk

delta 57
zcmV~$OBH}1006-!ia$^ak;6V7_HF|yVVMo$!ILvztM6T};}km#*yDf^N1SlR1y@X%
Kal`$wwDJQ?D;v=O

diff --git a/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00005.m4s b/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00005.m4s
index 309c9d7a4d1b9072f3a51c106463b5c1916e96cd..852cdaa712b0f877042ff16557fb1d64f19ad51d 100644
GIT binary patch
delta 328
zcmcb;&h_yI*9l?u3@xd-#pw({P+nfHSCCU$lCM`%#$ah_2^6vciG#rZ{`od}o-W2L
z3<^M=Z$OY|7?5^G)nt{NlbLN5Y3x-H<yeuU?-`mN=<XlonCcf1nP!=l7#?KenC=*w
z6rPiv>geg{k#1|tz);V?zz8&nkRv8G02P58kxykuaDZLGz~Imr(i+0p8p6~X!rU6d
u(i+0r8p75Z!rmIf(Hg?p8p72Y!rdCe(;C9t8p78a!rvMqur)-GJsbekM_kna

delta 81
zcmaF7!S(h!*9l?Gag6P8j3CSe#LPg<0>rF9%m&2lK+FNeoIuP4#N0s41H`;Q%m>8$
NKrFC5j!}>`902iF71{s*

diff --git a/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00006.m4s b/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00006.m4s
index b49c4e33814c01fa47fc1f4e72920f4b2cdf25bb..130d7c4120494af2f36b96efea1283adb72657a7 100644
GIT binary patch
delta 319
zcmex-)B5*q>j`1?3@xd-#pw({P+nfHSCCU$lCM`%#$ah_2^6vciG#rZ{`ocqo-W2L
z3<^M=Z$OY|7?5^G)nt{NlbLN5Y3x-H<yeuU?-`mN=<XlonCcf1nP!=l7#?KenC=*r
z6rPiv>geg{k#1|tz);V?zz8&nfFl^yfNH>wD5SC@IKZx8U}$IzX$@g)4Pj~xVQvj!
lX$@g*4Pk2yVQ&rLXbs_P4dH4H;cgA#X$|4s8p1dK4FI{ETO9xZ

delta 69
zcmex;+xqiO>j`1aag6P8j3CSe#LPg<0>rF9%m&2lK+FNeoIuP4#N0s41H`=B;~4qo
Gz5xIj&=i*d

diff --git a/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00007.m4s b/packager/media/test/data/live_packager_tests/encrypted/prd_data/decrypt/fmp4/00007.m4s
index 1d9215bc1286ae96a362faab215fe4d00be19aad..1685cacf918ea0ff51287534385288a8dbb376f8 100644
GIT binary patch
delta 143
zcmZ40#<Z@TX+n5?OKNU$Is*`tmzV1m<dl}=>y?x-SXx>Fg=|3LAn?C`zD<#*i!lp>
z0+8n$5abyKq@7VUStaLWW?MxXdlf`ER^;e=hNcI)`$svZ`b9*hS!N}M2bnmgJEkUu
Y=VYfkdOCWf+u8!nU}9k07;>!%0B8s)2><{9

delta 12
TcmZ42&a|qHX+rqM+n1XFB;5uw

diff --git a/packager/packager.cc b/packager/packager.cc
index 25467f9132f..39bf518f710 100644
--- a/packager/packager.cc
+++ b/packager/packager.cc
@@ -35,6 +35,7 @@
 #include <packager/media/demuxer/demuxer.h>
 #include <packager/media/event/muxer_listener_factory.h>
 #include <packager/media/event/vod_media_info_dump_muxer_listener.h>
+#include <packager/media/formats/mp4/dash_event_message_handler.h>
 #include <packager/media/formats/ttml/ttml_to_mp4_handler.h>
 #include <packager/media/formats/webvtt/text_padder.h>
 #include <packager/media/formats/webvtt/webvtt_to_mp4_handler.h>
@@ -458,9 +459,17 @@ bool StreamInfoToTextMediaInfo(const StreamDescriptor& stream_descriptor,
 /// Create a new demuxer handler for the given stream. If a demuxer cannot be
 /// created, an error will be returned. If a demuxer can be created, this
 /// |new_demuxer| will be set and Status::OK will be returned.
-Status CreateDemuxer(const StreamDescriptor& stream,
-                     const PackagingParams& packaging_params,
-                     std::shared_ptr<Demuxer>* new_demuxer) {
+Status CreateDemuxer(
+    const StreamDescriptor& stream,
+    const PackagingParams& packaging_params,
+    std::shared_ptr<Demuxer>* new_demuxer,
+    std::shared_ptr<mp4::DashEventMessageHandler>* new_emsg_handler) {
+  if (packaging_params.emsg_processing) {
+    std::shared_ptr<mp4::DashEventMessageHandler> emsg_handler =
+        std::make_shared<mp4::DashEventMessageHandler>();
+    *new_emsg_handler = std::move(emsg_handler);
+  }
+
   std::shared_ptr<Demuxer> demuxer = std::make_shared<Demuxer>(stream.input);
   demuxer->set_dump_stream_info(packaging_params.test_params.dump_stream_info);
   demuxer->set_cts_offset_adjustment(packaging_params.cts_offset_adjustment);
@@ -614,6 +623,8 @@ Status CreateAudioVideoJobs(
   // order.
   std::map<std::string, std::shared_ptr<Demuxer>> sources;
   std::map<std::string, std::shared_ptr<MediaHandler>> cue_aligners;
+  std::map<std::string, std::shared_ptr<mp4::DashEventMessageHandler>>
+      emsg_handlers;
 
   for (const StreamDescriptor& stream : streams) {
     bool seen_input_before = sources.find(stream.input) != sources.end();
@@ -621,8 +632,9 @@ Status CreateAudioVideoJobs(
       continue;
     }
 
-    RETURN_IF_ERROR(
-        CreateDemuxer(stream, packaging_params, &sources[stream.input]));
+    RETURN_IF_ERROR(CreateDemuxer(stream, packaging_params,
+                                  &sources[stream.input],
+                                  &emsg_handlers[stream.input]));
     cue_aligners[stream.input] =
         sync_points ? std::make_shared<CueAlignmentHandler>(sync_points)
                     : nullptr;
@@ -643,6 +655,7 @@ Status CreateAudioVideoJobs(
     // Get the demuxer for this stream.
     auto& demuxer = sources[stream.input];
     auto& cue_aligner = cue_aligners[stream.input];
+    auto& emsg_handler = emsg_handlers[stream.input];
 
     const bool new_input_file = stream.input != previous_input;
     const bool new_stream =
@@ -687,10 +700,12 @@ Status CreateAudioVideoJobs(
       RETURN_IF_ERROR(demuxer->SetHandler(stream.stream_selector, handlers[0]));
     }
 
+    demuxer->SetDashEventMessageHandler(emsg_handler);
+
     // Create the muxer (output) for this track.
     const auto output_format = GetOutputFormat(stream);
     std::shared_ptr<Muxer> muxer =
-        muxer_factory->CreateMuxer(output_format, stream);
+        muxer_factory->CreateMuxer(output_format, stream, emsg_handler);
     if (!muxer) {
       return Status(error::INVALID_ARGUMENT, "Failed to create muxer for " +
                                                  stream.input + ":" +