Handle DASH `emsg' events targeting player.

For live streaming, there are several types of DASH `emsg' events that directly
target the player. These events can signal whether the manifest is expired, or
the live streaming has ended, and should be handle directly within the player.

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=182034591

RELEASENOTES.md

@@ -32,7 +32,13 @@
   seeking to the closest sync points before, either side or after specified seek
   positions.
   * Note: `SeekParameters` are not currently supported when playing HLS streams.
-* DASH: Support DASH manifest EventStream elements.
+* DRM: Optimistically attempt playback of DRM protected content that does not
+  declare scheme specific init data
+  ([#3630](https://github.com/google/ExoPlayer/issues/3630)).
+* DASH:
+  * Support in-band Emsg events targeting player with scheme id
+    "urn:mpeg:dash:event:2012" and scheme value of either "1", "2" or "3".
+  * Support DASH manifest EventStream elements.
 * HLS: Add opt-in support for chunkless preparation in HLS. This allows an
   HLS source to finish preparation without downloading any chunks, which can
   significantly reduce initial buffering time

library/core/src/main/java/com/google/android/exoplayer2/source/chunk/ChunkSampleStream.java

@@ -15,6 +15,7 @@
  */
 package com.google.android.exoplayer2.source.chunk;
 
+import android.support.annotation.Nullable;
 import android.util.Log;
 import com.google.android.exoplayer2.C;
 import com.google.android.exoplayer2.Format;
@@ -41,6 +42,17 @@ import java.util.List;
 public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, SequenceableLoader,
     Loader.Callback<Chunk>, Loader.ReleaseCallback {
 
+  /** A callback to be notified when a sample stream has finished being released. */
+  public interface ReleaseCallback<T extends ChunkSource> {
+
+    /**
+     * Called when the {@link ChunkSampleStream} has finished being released.
+     *
+     * @param chunkSampleStream The released sample stream.
+     */
+    void onSampleStreamReleased(ChunkSampleStream<T> chunkSampleStream);
+  }
+
   private static final String TAG = "ChunkSampleStream";
 
   public final int primaryTrackType;
@@ -61,6 +73,7 @@ public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, S
   private final BaseMediaChunkOutput mediaChunkOutput;
 
   private Format primaryDownstreamTrackFormat;
+  private ReleaseCallback<T> releaseCallback;
   private long pendingResetPositionUs;
   /* package */ long lastSeekPositionUs;
   /* package */ boolean loadingFinished;
@@ -247,10 +260,26 @@ public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, S
 
   /**
    * Releases the stream.
-   * <p>
-   * This method should be called when the stream is no longer required.
+   *
+   * <p>This method should be called when the stream is no longer required. Either this method or
+   * {@link #release(ReleaseCallback)} can be used to release this stream.
    */
   public void release() {
+    release(null);
+  }
+
+  /**
+   * Releases the stream.
+   *
+   * <p>This method should be called when the stream is no longer required. Either this method or
+   * {@link #release()} can be used to release this stream.
+   *
+   * @param callback A callback to be called when the release ends. Will be called synchronously
+   *     from this method if no load is in progress, or asynchronously once the load has been
+   *     canceled otherwise.
+   */
+  public void release(@Nullable ReleaseCallback<T> callback) {
+    this.releaseCallback = callback;
     boolean releasedSynchronously = loader.release(this);
     if (!releasedSynchronously) {
       // Discard as much as we can synchronously.
@@ -267,6 +296,9 @@ public class ChunkSampleStream<T extends ChunkSource> implements SampleStream, S
     for (SampleQueue embeddedSampleQueue : embeddedSampleQueues) {
       embeddedSampleQueue.reset();
     }
+    if (releaseCallback != null) {
+      releaseCallback.onSampleStreamReleased(this);
+    }
   }
 
   // SampleStream implementation.

library/dash/src/main/java/com/google/android/exoplayer2/source/dash/DashChunkSource.java

@@ -16,7 +16,9 @@
 package com.google.android.exoplayer2.source.dash;
 
 import android.os.SystemClock;
+import android.support.annotation.Nullable;
 import com.google.android.exoplayer2.source.chunk.ChunkSource;
+import com.google.android.exoplayer2.source.dash.PlayerEmsgHandler.PlayerTrackEmsgHandler;
 import com.google.android.exoplayer2.source.dash.manifest.DashManifest;
 import com.google.android.exoplayer2.trackselection.TrackSelection;
 import com.google.android.exoplayer2.upstream.LoaderErrorThrower;
@@ -53,7 +55,8 @@ public interface DashChunkSource extends ChunkSource {
         int type,
         long elapsedRealtimeOffsetMs,
         boolean enableEventMessageTrack,
-        boolean enableCea608Track);
+        boolean enableCea608Track,
+        @Nullable PlayerTrackEmsgHandler playerEmsgHandler);
   }
 
   /**

library/dash/src/main/java/com/google/android/exoplayer2/source/dash/DashManifestExpiredException.java

+/*
+ * Copyright (C) 2018 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.android.exoplayer2.source.dash;
+
+import java.io.IOException;
+
+/** Thrown when a live playback's manifest is expired and a new manifest could not be loaded. */
+public final class DashManifestExpiredException extends IOException {}

library/dash/src/main/java/com/google/android/exoplayer2/source/dash/DashMediaPeriod.java

@@ -31,6 +31,8 @@ import com.google.android.exoplayer2.source.TrackGroup;
 import com.google.android.exoplayer2.source.TrackGroupArray;
 import com.google.android.exoplayer2.source.chunk.ChunkSampleStream;
 import com.google.android.exoplayer2.source.chunk.ChunkSampleStream.EmbeddedSampleStream;
+import com.google.android.exoplayer2.source.dash.PlayerEmsgHandler.PlayerEmsgCallback;
+import com.google.android.exoplayer2.source.dash.PlayerEmsgHandler.PlayerTrackEmsgHandler;
 import com.google.android.exoplayer2.source.dash.manifest.AdaptationSet;
 import com.google.android.exoplayer2.source.dash.manifest.DashManifest;
 import com.google.android.exoplayer2.source.dash.manifest.Descriptor;
@@ -47,14 +49,15 @@ import java.lang.annotation.RetentionPolicy;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
+import java.util.IdentityHashMap;
 import java.util.List;
 import java.util.Map;
 
-/**
- * A DASH {@link MediaPeriod}.
- */
-/* package */ final class DashMediaPeriod implements MediaPeriod,
-    SequenceableLoader.Callback<ChunkSampleStream<DashChunkSource>> {
+/** A DASH {@link MediaPeriod}. */
+/* package */ final class DashMediaPeriod
+    implements MediaPeriod,
+        SequenceableLoader.Callback<ChunkSampleStream<DashChunkSource>>,
+        ChunkSampleStream.ReleaseCallback<DashChunkSource> {
 
   /* package */ final int id;
   private final DashChunkSource.Factory chunkSourceFactory;
@@ -66,6 +69,9 @@ import java.util.Map;
   private final TrackGroupArray trackGroups;
   private final TrackGroupInfo[] trackGroupInfos;
   private final CompositeSequenceableLoaderFactory compositeSequenceableLoaderFactory;
+  private final PlayerEmsgHandler playerEmsgHandler;
+  private final IdentityHashMap<ChunkSampleStream<DashChunkSource>, PlayerTrackEmsgHandler>
+      trackEmsgHandlerBySampleStream;
 
   private Callback callback;
   private ChunkSampleStream<DashChunkSource>[] sampleStreams;
@@ -75,11 +81,18 @@ import java.util.Map;
   private int periodIndex;
   private List<EventStream> eventStreams;
 
-  public DashMediaPeriod(int id, DashManifest manifest, int periodIndex,
-      DashChunkSource.Factory chunkSourceFactory, int minLoadableRetryCount,
-      EventDispatcher eventDispatcher, long elapsedRealtimeOffset,
-      LoaderErrorThrower manifestLoaderErrorThrower, Allocator allocator,
-      CompositeSequenceableLoaderFactory compositeSequenceableLoaderFactory) {
+  public DashMediaPeriod(
+      int id,
+      DashManifest manifest,
+      int periodIndex,
+      DashChunkSource.Factory chunkSourceFactory,
+      int minLoadableRetryCount,
+      EventDispatcher eventDispatcher,
+      long elapsedRealtimeOffset,
+      LoaderErrorThrower manifestLoaderErrorThrower,
+      Allocator allocator,
+      CompositeSequenceableLoaderFactory compositeSequenceableLoaderFactory,
+      PlayerEmsgCallback playerEmsgCallback) {
     this.id = id;
     this.manifest = manifest;
     this.periodIndex = periodIndex;
@@ -90,8 +103,10 @@ import java.util.Map;
     this.manifestLoaderErrorThrower = manifestLoaderErrorThrower;
     this.allocator = allocator;
     this.compositeSequenceableLoaderFactory = compositeSequenceableLoaderFactory;
+    playerEmsgHandler = new PlayerEmsgHandler(manifest, playerEmsgCallback, allocator);
     sampleStreams = newSampleStreamArray(0);
     eventSampleStreams = new EventSampleStream[0];
+    trackEmsgHandlerBySampleStream = new IdentityHashMap<>();
     compositeSequenceableLoader =
         compositeSequenceableLoaderFactory.createCompositeSequenceableLoader(sampleStreams);
     Period period = manifest.getPeriod(periodIndex);
@@ -111,14 +126,14 @@ import java.util.Map;
   public void updateManifest(DashManifest manifest, int periodIndex) {
     this.manifest = manifest;
     this.periodIndex = periodIndex;
-    Period period = manifest.getPeriod(periodIndex);
+    playerEmsgHandler.updateManifest(manifest);
     if (sampleStreams != null) {
       for (ChunkSampleStream<DashChunkSource> sampleStream : sampleStreams) {
         sampleStream.getChunkSource().updateManifest(manifest, periodIndex);
       }
       callback.onContinueLoadingRequested(this);
     }
-    eventStreams = period.eventStreams;
+    eventStreams = manifest.getPeriod(periodIndex).eventStreams;
     for (EventSampleStream eventSampleStream : eventSampleStreams) {
       for (EventStream eventStream : eventStreams) {
         if (eventStream.id().equals(eventSampleStream.eventStreamId())) {
@@ -130,11 +145,24 @@ import java.util.Map;
   }
 
   public void release() {
+    playerEmsgHandler.release();
     for (ChunkSampleStream<DashChunkSource> sampleStream : sampleStreams) {
-      sampleStream.release();
+      sampleStream.release(this);
+    }
+  }
+
+  // ChunkSampleStream.ReleaseCallback implementation.
+
+  @Override
+  public void onSampleStreamReleased(ChunkSampleStream<DashChunkSource> stream) {
+    PlayerTrackEmsgHandler trackEmsgHandler = trackEmsgHandlerBySampleStream.remove(stream);
+    if (trackEmsgHandler != null) {
+      trackEmsgHandler.release();
     }
   }
 
+  // MediaPeriod implementation.
+
   @Override
   public void prepare(Callback callback, long positionUs) {
     this.callback = callback;
@@ -181,7 +209,7 @@ import java.util.Map;
         @SuppressWarnings("unchecked")
         ChunkSampleStream<DashChunkSource> stream = (ChunkSampleStream<DashChunkSource>) streams[i];
         if (selections[i] == null || !mayRetainStreamFlags[i]) {
-          stream.release();
+          stream.release(this);
           streams[i] = null;
         } else {
           int trackGroupIndex = trackGroups.indexOf(selections[i].getTrackGroup());
@@ -501,10 +529,22 @@ import java.util.Map;
       embeddedTrackFormats = Arrays.copyOf(embeddedTrackFormats, embeddedTrackCount);
       embeddedTrackTypes = Arrays.copyOf(embeddedTrackTypes, embeddedTrackCount);
     }
-    DashChunkSource chunkSource = chunkSourceFactory.createDashChunkSource(
-        manifestLoaderErrorThrower, manifest, periodIndex, trackGroupInfo.adaptationSetIndices,
-        selection, trackGroupInfo.trackType, elapsedRealtimeOffset, enableEventMessageTrack,
-        enableCea608Track);
+    PlayerTrackEmsgHandler trackPlayerEmsgHandler =
+        manifest.dynamic && enableEventMessageTrack
+            ? playerEmsgHandler.newPlayerTrackEmsgHandler()
+            : null;
+    DashChunkSource chunkSource =
+        chunkSourceFactory.createDashChunkSource(
+            manifestLoaderErrorThrower,
+            manifest,
+            periodIndex,
+            trackGroupInfo.adaptationSetIndices,
+            selection,
+            trackGroupInfo.trackType,
+            elapsedRealtimeOffset,
+            enableEventMessageTrack,
+            enableCea608Track,
+            trackPlayerEmsgHandler);
     ChunkSampleStream<DashChunkSource> stream =
         new ChunkSampleStream<>(
             trackGroupInfo.trackType,
@@ -516,6 +556,7 @@ import java.util.Map;
             positionUs,
             minLoadableRetryCount,
             eventDispatcher);
+    trackEmsgHandlerBySampleStream.put(stream, trackPlayerEmsgHandler);
     return stream;
   }
 
@@ -581,9 +622,8 @@ import java.util.Map;
     private static final int CATEGORY_PRIMARY = 0;
 
     /**
-     * A track group whose samples are embedded within one of the primary streams.
-     * For example: an EMSG track has its sample embedded in `emsg' atoms in one of the primary
-     * streams.
+     * A track group whose samples are embedded within one of the primary streams. For example: an
+     * EMSG track has its sample embedded in emsg atoms in one of the primary streams.
      */
     private static final int CATEGORY_EMBEDDED = 1;
 

library/dash/src/main/java/com/google/android/exoplayer2/source/dash/DashMediaSource.java

@@ -35,6 +35,7 @@ import com.google.android.exoplayer2.source.MediaSourceEventListener;
 import com.google.android.exoplayer2.source.MediaSourceEventListener.EventDispatcher;
 import com.google.android.exoplayer2.source.SequenceableLoader;
 import com.google.android.exoplayer2.source.ads.AdsMediaSource;
+import com.google.android.exoplayer2.source.dash.PlayerEmsgHandler.PlayerEmsgCallback;
 import com.google.android.exoplayer2.source.dash.manifest.DashManifest;
 import com.google.android.exoplayer2.source.dash.manifest.DashManifestParser;
 import com.google.android.exoplayer2.source.dash.manifest.UtcTimingElement;
@@ -56,9 +57,7 @@ import java.util.TimeZone;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
-/**
- * A DASH {@link MediaSource}.
- */
+/** A DASH {@link MediaSource}. */
 public final class DashMediaSource implements MediaSource {
 
   static {
@@ -280,6 +279,7 @@ public final class DashMediaSource implements MediaSource {
   private final SparseArray<DashMediaPeriod> periodsById;
   private final Runnable refreshManifestRunnable;
   private final Runnable simulateManifestRefreshRunnable;
+  private final PlayerEmsgCallback playerEmsgCallback;
 
   private Listener sourceListener;
   private DataSource dataSource;
@@ -291,7 +291,11 @@ public final class DashMediaSource implements MediaSource {
   private long manifestLoadEndTimestamp;
   private DashManifest manifest;
   private Handler handler;
+  private boolean pendingManifestLoading;
   private long elapsedRealtimeOffsetMs;
+  private long expiredManifestPublishTimeUs;
+  private boolean dynamicMediaPresentationEnded;
+  private int staleManifestReloadAttempt;
 
   private int firstPeriodId;
 
@@ -446,6 +450,8 @@ public final class DashMediaSource implements MediaSource {
     eventDispatcher = new EventDispatcher(eventHandler, eventListener);
     manifestUriLock = new Object();
     periodsById = new SparseArray<>();
+    playerEmsgCallback = new DefaultPlayerEmsgCallback();
+    expiredManifestPublishTimeUs = C.TIME_UNSET;
     if (sideloadedManifest) {
       Assertions.checkState(!manifest.dynamic);
       manifestCallback = null;
@@ -507,9 +513,19 @@ public final class DashMediaSource implements MediaSource {
     int periodIndex = periodId.periodIndex;
     EventDispatcher periodEventDispatcher = eventDispatcher.copyWithMediaTimeOffsetMs(
         manifest.getPeriod(periodIndex).startMs);
-    DashMediaPeriod mediaPeriod = new DashMediaPeriod(firstPeriodId + periodIndex, manifest,
-        periodIndex, chunkSourceFactory, minLoadableRetryCount, periodEventDispatcher,
-        elapsedRealtimeOffsetMs, loaderErrorThrower, allocator, compositeSequenceableLoaderFactory);
+    DashMediaPeriod mediaPeriod =
+        new DashMediaPeriod(
+            firstPeriodId + periodIndex,
+            manifest,
+            periodIndex,
+            chunkSourceFactory,
+            minLoadableRetryCount,
+            periodEventDispatcher,
+            elapsedRealtimeOffsetMs,
+            loaderErrorThrower,
+            allocator,
+            compositeSequenceableLoaderFactory,
+            playerEmsgCallback);
     periodsById.put(mediaPeriod.id, mediaPeriod);
     return mediaPeriod;
   }
@@ -523,6 +539,7 @@ public final class DashMediaSource implements MediaSource {
 
   @Override
   public void releaseSource() {
+    pendingManifestLoading = false;
     dataSource = null;
     loaderErrorThrower = null;
     if (loader != null) {
@@ -540,6 +557,24 @@ public final class DashMediaSource implements MediaSource {
     periodsById.clear();
   }
 
+  // PlayerEmsgCallback callbacks.
+
+  /* package */ void onDashManifestRefreshRequested() {
+    handler.removeCallbacks(simulateManifestRefreshRunnable);
+    startLoadingManifest();
+  }
+
+  /* package */ void onDashLiveMediaPresentationEndSignalEncountered() {
+    this.dynamicMediaPresentationEnded = true;
+  }
+
+  /* package */ void onDashManifestPublishTimeExpired(long expiredManifestPublishTimeUs) {
+    if (this.expiredManifestPublishTimeUs == C.TIME_UNSET
+        || this.expiredManifestPublishTimeUs < expiredManifestPublishTimeUs) {
+      this.expiredManifestPublishTimeUs = expiredManifestPublishTimeUs;
+    }
+  }
+
   // Loadable callbacks.
 
   /* package */ void onManifestLoadCompleted(ParsingLoadable<DashManifest> loadable,
@@ -566,9 +601,16 @@ public final class DashMediaSource implements MediaSource {
       return;
     }
 
+    if (maybeReloadStaleDynamicManifest(newManifest)) {
+      return;
+    }
     manifest = newManifest;
     manifestLoadStartTimestamp = elapsedRealtimeMs - loadDurationMs;
     manifestLoadEndTimestamp = elapsedRealtimeMs;
+    staleManifestReloadAttempt = 0;
+    if (!manifest.dynamic) {
+      pendingManifestLoading = false;
+    }
     if (manifest.location != null) {
       synchronized (manifestUriLock) {
         // This condition checks that replaceManifestUri wasn't called between the start and end of
@@ -622,11 +664,41 @@ public final class DashMediaSource implements MediaSource {
 
   // Internal methods.
 
+  /**
+   * Reloads a stale dynamic manifest to get a more recent version if possible.
+   *
+   * @return True if the reload is scheduled. False if we have already retried too many times.
+   */
+  private boolean maybeReloadStaleDynamicManifest(DashManifest manifest) {
+    if (!isManifestStale(manifest)) {
+      return false;
+    }
+    String warning =
+        "Loaded a stale dynamic manifest "
+            + manifest.publishTimeMs
+            + " "
+            + dynamicMediaPresentationEnded
+            + " "
+            + expiredManifestPublishTimeUs;
+    Log.w(TAG, warning);
+    if (staleManifestReloadAttempt++ < minLoadableRetryCount) {
+      startLoadingManifest();
+      return true;
+    }
+    return false;
+  }
+
   private void startLoadingManifest() {
+    handler.removeCallbacks(refreshManifestRunnable);
+    if (loader.isLoading()) {
+      pendingManifestLoading = true;
+      return;
+    }
     Uri manifestUri;
     synchronized (manifestUriLock) {
       manifestUri = this.manifestUri;
     }
+    pendingManifestLoading = false;
     startLoading(new ParsingLoadable<>(dataSource, manifestUri, C.DATA_TYPE_MANIFEST,
         manifestParser), manifestCallback, minLoadableRetryCount);
   }
@@ -753,13 +825,21 @@ public final class DashMediaSource implements MediaSource {
       if (windowChangingImplicitly) {
         handler.postDelayed(simulateManifestRefreshRunnable, NOTIFY_MANIFEST_INTERVAL_MS);
       }
+      if (pendingManifestLoading) {
+        startLoadingManifest();
+      } else if (scheduleRefresh) {
         // Schedule an explicit refresh if needed.
-      if (scheduleRefresh) {
         scheduleManifestRefresh();
       }
     }
   }
 
+  private boolean isManifestStale(DashManifest manifest) {
+    return manifest.dynamic
+        && (dynamicMediaPresentationEnded
+            || manifest.publishTimeMs <= expiredManifestPublishTimeUs);
+  }
+
   private void scheduleManifestRefresh() {
     if (!manifest.dynamic) {
       return;
@@ -948,6 +1028,24 @@ public final class DashMediaSource implements MediaSource {
 
   }
 
+  private final class DefaultPlayerEmsgCallback implements PlayerEmsgCallback {
+
+    @Override
+    public void onDashManifestRefreshRequested() {
+      DashMediaSource.this.onDashManifestRefreshRequested();
+    }
+
+    @Override
+    public void onDashManifestPublishTimeExpired(long expiredManifestPublishTimeUs) {
+      DashMediaSource.this.onDashManifestPublishTimeExpired(expiredManifestPublishTimeUs);
+    }
+
+    @Override
+    public void onDashLiveMediaPresentationEndSignalEncountered() {
+      DashMediaSource.this.onDashLiveMediaPresentationEndSignalEncountered();
+    }
+  }
+
   private final class ManifestCallback implements Loader.Callback<ParsingLoadable<DashManifest>> {
 
     @Override
@@ -1039,5 +1137,4 @@ public final class DashMediaSource implements MediaSource {
     }
 
   }
-
 }

library/dash/src/main/java/com/google/android/exoplayer2/source/dash/DefaultDashChunkSource.java

@@ -17,12 +17,14 @@ package com.google.android.exoplayer2.source.dash;
 
 import android.net.Uri;
 import android.os.SystemClock;
+import android.support.annotation.Nullable;
 import com.google.android.exoplayer2.C;
 import com.google.android.exoplayer2.Format;
 import com.google.android.exoplayer2.SeekParameters;
 import com.google.android.exoplayer2.extractor.ChunkIndex;
 import com.google.android.exoplayer2.extractor.Extractor;
 import com.google.android.exoplayer2.extractor.SeekMap;
+import com.google.android.exoplayer2.extractor.TrackOutput;
 import com.google.android.exoplayer2.extractor.mkv.MatroskaExtractor;
 import com.google.android.exoplayer2.extractor.mp4.FragmentedMp4Extractor;
 import com.google.android.exoplayer2.extractor.rawcc.RawCcExtractor;
@@ -35,6 +37,7 @@ import com.google.android.exoplayer2.source.chunk.ContainerMediaChunk;
 import com.google.android.exoplayer2.source.chunk.InitializationChunk;
 import com.google.android.exoplayer2.source.chunk.MediaChunk;
 import com.google.android.exoplayer2.source.chunk.SingleSampleMediaChunk;
+import com.google.android.exoplayer2.source.dash.PlayerEmsgHandler.PlayerTrackEmsgHandler;
 import com.google.android.exoplayer2.source.dash.manifest.AdaptationSet;
 import com.google.android.exoplayer2.source.dash.manifest.DashManifest;
 import com.google.android.exoplayer2.source.dash.manifest.RangedUri;
@@ -71,14 +74,31 @@ public class DefaultDashChunkSource implements DashChunkSource {
     }
 
     @Override
-    public DashChunkSource createDashChunkSource(LoaderErrorThrower manifestLoaderErrorThrower,
-        DashManifest manifest, int periodIndex, int[] adaptationSetIndices,
-        TrackSelection trackSelection, int trackType, long elapsedRealtimeOffsetMs,
-        boolean enableEventMessageTrack, boolean enableCea608Track) {
+    public DashChunkSource createDashChunkSource(
+        LoaderErrorThrower manifestLoaderErrorThrower,
+        DashManifest manifest,
+        int periodIndex,
+        int[] adaptationSetIndices,
+        TrackSelection trackSelection,
+        int trackType,
+        long elapsedRealtimeOffsetMs,
+        boolean enableEventMessageTrack,
+        boolean enableCea608Track,
+        @Nullable PlayerTrackEmsgHandler playerEmsgHandler) {
       DataSource dataSource = dataSourceFactory.createDataSource();
-      return new DefaultDashChunkSource(manifestLoaderErrorThrower, manifest, periodIndex,
-          adaptationSetIndices, trackSelection, trackType, dataSource, elapsedRealtimeOffsetMs,
-          maxSegmentsPerLoad, enableEventMessageTrack, enableCea608Track);
+      return new DefaultDashChunkSource(
+          manifestLoaderErrorThrower,
+          manifest,
+          periodIndex,
+          adaptationSetIndices,
+          trackSelection,
+          trackType,
+          dataSource,
+          elapsedRealtimeOffsetMs,
+          maxSegmentsPerLoad,
+          enableEventMessageTrack,
+          enableCea608Track,
+          playerEmsgHandler);
     }
 
   }
@@ -90,6 +110,7 @@ public class DefaultDashChunkSource implements DashChunkSource {
   private final DataSource dataSource;
   private final long elapsedRealtimeOffsetMs;
   private final int maxSegmentsPerLoad;
+  @Nullable private final PlayerTrackEmsgHandler playerTrackEmsgHandler;
 
   protected final RepresentationHolder[] representationHolders;
 
@@ -110,18 +131,28 @@ public class DefaultDashChunkSource implements DashChunkSource {
    * @param elapsedRealtimeOffsetMs If known, an estimate of the instantaneous difference between
    *     server-side unix time and {@link SystemClock#elapsedRealtime()} in milliseconds, specified
    *     as the server's unix time minus the local elapsed time. If unknown, set to 0.
-   * @param maxSegmentsPerLoad The maximum number of segments to combine into a single request.
-   *     Note that segments will only be combined if their {@link Uri}s are the same and if their
-   *     data ranges are adjacent.
+   * @param maxSegmentsPerLoad The maximum number of segments to combine into a single request. Note
+   *     that segments will only be combined if their {@link Uri}s are the same and if their data
+   *     ranges are adjacent.
    * @param enableEventMessageTrack Whether the chunks generated by the source may output an event
    *     message track.
    * @param enableCea608Track Whether the chunks generated by the source may output a CEA-608 track.
+   * @param playerTrackEmsgHandler The {@link PlayerTrackEmsgHandler} instance to handle emsg
+   *     messages targeting the player. Maybe null if this is not necessary.
    */
-  public DefaultDashChunkSource(LoaderErrorThrower manifestLoaderErrorThrower,
-      DashManifest manifest, int periodIndex, int[] adaptationSetIndices,
-      TrackSelection trackSelection, int trackType, DataSource dataSource,
-      long elapsedRealtimeOffsetMs, int maxSegmentsPerLoad, boolean enableEventMessageTrack,
-      boolean enableCea608Track) {
+  public DefaultDashChunkSource(
+      LoaderErrorThrower manifestLoaderErrorThrower,
+      DashManifest manifest,
+      int periodIndex,
+      int[] adaptationSetIndices,
+      TrackSelection trackSelection,
+      int trackType,
+      DataSource dataSource,
+      long elapsedRealtimeOffsetMs,
+      int maxSegmentsPerLoad,
+      boolean enableEventMessageTrack,
+      boolean enableCea608Track,
+      @Nullable PlayerTrackEmsgHandler playerTrackEmsgHandler) {
     this.manifestLoaderErrorThrower = manifestLoaderErrorThrower;
     this.manifest = manifest;
     this.adaptationSetIndices = adaptationSetIndices;
@@ -131,15 +162,23 @@ public class DefaultDashChunkSource implements DashChunkSource {
     this.periodIndex = periodIndex;
     this.elapsedRealtimeOffsetMs = elapsedRealtimeOffsetMs;
     this.maxSegmentsPerLoad = maxSegmentsPerLoad;
+    this.playerTrackEmsgHandler = playerTrackEmsgHandler;
 
     long periodDurationUs = manifest.getPeriodDurationUs(periodIndex);
     liveEdgeTimeUs = C.TIME_UNSET;
+
     List<Representation> representations = getRepresentations();
     representationHolders = new RepresentationHolder[trackSelection.length()];
     for (int i = 0; i < representationHolders.length; i++) {
       Representation representation = representations.get(trackSelection.getIndexInTrackGroup(i));
-      representationHolders[i] = new RepresentationHolder(periodDurationUs, trackType,
-          representation, enableEventMessageTrack, enableCea608Track);
+      representationHolders[i] =
+          new RepresentationHolder(
+              periodDurationUs,
+              trackType,
+              representation,
+              enableEventMessageTrack,
+              enableCea608Track,
+              playerTrackEmsgHandler);
     }
   }
 
@@ -203,6 +242,20 @@ public class DefaultDashChunkSource implements DashChunkSource {
 
     long bufferedDurationUs = loadPositionUs - playbackPositionUs;
     long timeToLiveEdgeUs = resolveTimeToLiveEdgeUs(playbackPositionUs);
+    long presentationPositionUs =
+        C.msToUs(manifest.availabilityStartTimeMs)
+            + C.msToUs(manifest.getPeriod(periodIndex).startMs)
+            + loadPositionUs;
+    try {
+      if (playerTrackEmsgHandler != null
+          && playerTrackEmsgHandler.maybeRefreshManifestBeforeLoadingNextChunk(
+              presentationPositionUs)) {
+        return;
+      }
+    } catch (DashManifestExpiredException e) {
+      fatalError = e;
+      return;
+    }
     trackSelection.updateSelectedTrack(playbackPositionUs, bufferedDurationUs, timeToLiveEdgeUs);
 
     RepresentationHolder representationHolder =
@@ -298,6 +351,9 @@ public class DefaultDashChunkSource implements DashChunkSource {
         }
       }
     }
+    if (playerTrackEmsgHandler != null) {
+      playerTrackEmsgHandler.onChunkLoadCompleted(chunk);
+    }
   }
 
   @Override
@@ -305,6 +361,10 @@ public class DefaultDashChunkSource implements DashChunkSource {
     if (!cancelable) {
       return false;
     }
+    if (playerTrackEmsgHandler != null
+        && playerTrackEmsgHandler.maybeRefreshManifestOnLoadingError(chunk)) {
+      return true;
+    }
     // Workaround for missing segment at the end of the period
     if (!manifest.dynamic && chunk instanceof MediaChunk
         && e instanceof InvalidResponseCodeException
@@ -426,8 +486,13 @@ public class DefaultDashChunkSource implements DashChunkSource {
     private long periodDurationUs;
     private int segmentNumShift;
 
-    /* package */ RepresentationHolder(long periodDurationUs, int trackType,
-        Representation representation, boolean enableEventMessageTrack, boolean enableCea608Track) {
+    /* package */ RepresentationHolder(
+        long periodDurationUs,
+        int trackType,
+        Representation representation,
+        boolean enableEventMessageTrack,
+        boolean enableCea608Track,
+        TrackOutput playerEmsgTrackOutput) {
       this.periodDurationUs = periodDurationUs;
       this.representation = representation;
       String containerMimeType = representation.format.containerMimeType;
@@ -449,7 +514,10 @@ public class DefaultDashChunkSource implements DashChunkSource {
               ? Collections.singletonList(
                   Format.createTextSampleFormat(null, MimeTypes.APPLICATION_CEA608, 0, null))
               : Collections.<Format>emptyList();
-          extractor = new FragmentedMp4Extractor(flags, null, null, null, closedCaptionFormats);
+
+          extractor =
+              new FragmentedMp4Extractor(
+                  flags, null, null, null, closedCaptionFormats, playerEmsgTrackOutput);
         }
         // Prefer drmInitData obtained from the manifest over drmInitData obtained from the stream,
         // as per DASH IF Interoperability Recommendations V3.0, 7.5.3.
@@ -534,7 +602,5 @@ public class DefaultDashChunkSource implements DashChunkSource {
     private static boolean mimeTypeIsRawText(String mimeType) {
       return MimeTypes.isText(mimeType) || MimeTypes.APPLICATION_TTML.equals(mimeType);
     }
-
   }
-
 }

library/dash/src/main/java/com/google/android/exoplayer2/source/dash/PlayerEmsgHandler.java

+/*
+ * Copyright (C) 2017 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.android.exoplayer2.source.dash;
+
+import static com.google.android.exoplayer2.util.Util.parseXsDateTime;
+
+import android.os.Handler;
+import android.os.Message;
+import android.support.annotation.Nullable;
+import com.google.android.exoplayer2.C;
+import com.google.android.exoplayer2.Format;
+import com.google.android.exoplayer2.FormatHolder;
+import com.google.android.exoplayer2.ParserException;
+import com.google.android.exoplayer2.extractor.ExtractorInput;
+import com.google.android.exoplayer2.extractor.TrackOutput;
+import com.google.android.exoplayer2.metadata.Metadata;
+import com.google.android.exoplayer2.metadata.MetadataInputBuffer;
+import com.google.android.exoplayer2.metadata.emsg.EventMessage;
+import com.google.android.exoplayer2.metadata.emsg.EventMessageDecoder;
+import com.google.android.exoplayer2.source.SampleQueue;
+import com.google.android.exoplayer2.source.chunk.Chunk;
+import com.google.android.exoplayer2.source.dash.manifest.DashManifest;
+import com.google.android.exoplayer2.upstream.Allocator;
+import com.google.android.exoplayer2.util.ParsableByteArray;
+import java.io.IOException;
+import java.util.Map;
+import java.util.TreeMap;
+
+/**
+ * Handles all emsg messages from all media tracks for the player.
+ *
+ * <p>This class will only respond to emsg messages which have schemeIdUri
+ * "urn:mpeg:dash:event:2012", and value "1"/"2"/"3". When it encounters one of these messages, it
+ * will handle the message according to Section 4.5.2.1 DASH -IF IOP Version 4.1:
+ *
+ * <ul>
+ *   <li>If both presentation time delta and event duration are zero, it means the media
+ *       presentation has ended.
+ *   <li>Else, it will parse the message data from the emsg message to find the publishTime of the
+ *       expired manifest, and mark manifest with publishTime smaller than that values to be
+ *       expired.
+ * </ul>
+ *
+ * In both cases, the DASH media source will be notified, and a manifest reload should be triggered.
+ */
+public final class PlayerEmsgHandler implements Handler.Callback {
+
+  private static final int EMSG_MEDIA_PRESENTATION_ENDED = 1;
+  private static final int EMSG_MANIFEST_EXPIRED = 2;
+
+  /** Callbacks for player emsg events encountered during DASH live stream. */
+  public interface PlayerEmsgCallback {
+
+    /** Called when the current manifest should be refreshed. */
+    void onDashManifestRefreshRequested();
+
+    /**
+     * Called when the manifest with the publish time has been expired.
+     *
+     * @param expiredManifestPublishTimeUs The manifest publish time that has been expired.
+     */
+    void onDashManifestPublishTimeExpired(long expiredManifestPublishTimeUs);
+
+    /** Called when a media presentation end signal is encountered during live stream. * */
+    void onDashLiveMediaPresentationEndSignalEncountered();
+  }
+
+  private final Allocator allocator;
+  private final PlayerEmsgCallback playerEmsgCallback;
+  private final EventMessageDecoder decoder;
+  private final Handler handler;
+  private final TreeMap<Long, Long> manifestPublishTimeToExpiryTimeUs;
+
+  private DashManifest manifest;
+
+  private boolean dynamicMediaPresentationEnded;
+  private long expiredManifestPublishTimeUs;
+  private long lastLoadedChunkEndTimeUs;
+  private long lastLoadedChunkEndTimeBeforeRefreshUs;
+  private boolean isWaitingForManifestRefresh;
+  private boolean released;
+  private DashManifestExpiredException fatalError;
+
+  /**
+   * @param manifest The initial manifest.
+   * @param playerEmsgCallback The callback that this event handler can invoke when handling emsg
+   *     messages that generate DASH media source events.
+   * @param allocator An {@link Allocator} from which allocations can be obtained.
+   */
+  public PlayerEmsgHandler(
+      DashManifest manifest, PlayerEmsgCallback playerEmsgCallback, Allocator allocator) {
+    this.manifest = manifest;
+    this.playerEmsgCallback = playerEmsgCallback;
+    this.allocator = allocator;
+
+    manifestPublishTimeToExpiryTimeUs = new TreeMap<>();
+    handler = new Handler(this);
+    decoder = new EventMessageDecoder();
+    lastLoadedChunkEndTimeUs = C.TIME_UNSET;
+    lastLoadedChunkEndTimeBeforeRefreshUs = C.TIME_UNSET;
+  }
+
+  /**
+   * Updates the {@link DashManifest} that this handler works on.
+   *
+   * @param newManifest The updated manifest.
+   */
+  public void updateManifest(DashManifest newManifest) {
+    if (isManifestStale(newManifest)) {
+      fatalError = new DashManifestExpiredException();
+    }
+
+    isWaitingForManifestRefresh = false;
+    expiredManifestPublishTimeUs = C.TIME_UNSET;
+    this.manifest = newManifest;
+  }
+
+  private boolean isManifestStale(DashManifest manifest) {
+    return manifest.dynamic
+        && (dynamicMediaPresentationEnded
+            || manifest.publishTimeMs <= expiredManifestPublishTimeUs);
+  }
+
+  /* package*/ boolean maybeRefreshManifestBeforeLoadingNextChunk(long presentationPositionUs)
+      throws DashManifestExpiredException {
+    if (fatalError != null) {
+      throw fatalError;
+    }
+    if (!manifest.dynamic) {
+      return false;
+    }
+    if (isWaitingForManifestRefresh) {
+      return true;
+    }
+    boolean manifestRefreshNeeded = false;
+    if (dynamicMediaPresentationEnded) {
+      // The manifest we have is dynamic, but we know a non-dynamic one representing the final state
+      // should be available.
+      manifestRefreshNeeded = true;
+    } else {
+      // Find the smallest publishTime (greater than or equal to the current manifest's publish
+      // time) that has a corresponding expiry time.
+      Map.Entry<Long, Long> expiredEntry = ceilingExpiryEntryForPublishTime(manifest.publishTimeMs);
+      if (expiredEntry != null) {
+        long expiredPointUs = expiredEntry.getValue();
+        if (expiredPointUs < presentationPositionUs) {
+          expiredManifestPublishTimeUs = expiredEntry.getKey();
+          notifyManifestPublishTimeExpired();
+          manifestRefreshNeeded = true;
+        }
+      }
+    }
+    if (manifestRefreshNeeded) {
+      maybeNotifyDashManifestRefreshNeeded();
+    }
+    return manifestRefreshNeeded;
+  }
+
+  /**
+   * For live streaming with emsg event stream, forward seeking can seek pass the emsg messages that
+   * signals end-of-stream or Manifest expiry, which results in load error. In this case, we should
+   * notify the Dash media source to refresh its manifest.
+   *
+   * @param chunk The chunk whose load encountered the error.
+   * @return True if manifest refresh has been requested, false otherwise.
+   */
+  /* package */ boolean maybeRefreshManifestOnLoadingError(Chunk chunk) {
+    if (!manifest.dynamic) {
+      return false;
+    }
+    if (isWaitingForManifestRefresh) {
+      return true;
+    }
+    boolean isAfterForwardSeek =
+        lastLoadedChunkEndTimeUs != C.TIME_UNSET && lastLoadedChunkEndTimeUs < chunk.startTimeUs;
+    if (isAfterForwardSeek) {
+      // if we are after a forward seek, and the playback is dynamic with embedded emsg stream,
+      // there's a chance that we have seek over the emsg messages, in which case we should ask
+      // media source for a refresh.
+      maybeNotifyDashManifestRefreshNeeded();
+      return true;
+    }
+    return false;
+  }
+
+  /**
+   * Called when the a new chunk in the current media stream has been loaded.
+   *
+   * @param chunk The chunk whose load has been completed.
+   */
+  /* package */ void onChunkLoadCompleted(Chunk chunk) {
+    if (lastLoadedChunkEndTimeUs != C.TIME_UNSET || chunk.endTimeUs > lastLoadedChunkEndTimeUs) {
+      lastLoadedChunkEndTimeUs = chunk.endTimeUs;
+    }
+  }
+
+  /**
+   * Returns whether an event with given schemeIdUri and value is a DASH emsg event targeting the
+   * player.
+   */
+  public static boolean isPlayerEmsgEvent(String schemeIdUri, String value) {
+    return "urn:mpeg:dash:event:2012".equals(schemeIdUri)
+        && ("1".equals(value) || "2".equals(value) || "3".equals(value));
+  }
+
+  /** Returns a {@link TrackOutput} that emsg messages could be written to. */
+  public PlayerTrackEmsgHandler newPlayerTrackEmsgHandler() {
+    return new PlayerTrackEmsgHandler(new SampleQueue(allocator));
+  }
+
+  /** Release this emsg handler. It should not be reused after this call. */
+  public void release() {
+    released = true;
+    handler.removeCallbacksAndMessages(null);
+  }
+
+  @Override
+  public boolean handleMessage(Message message) {
+    if (released) {
+      return true;
+    }
+    switch (message.what) {
+      case (EMSG_MEDIA_PRESENTATION_ENDED):
+        handleMediaPresentationEndedMessageEncountered();
+        return true;
+      case (EMSG_MANIFEST_EXPIRED):
+        ManifestExpiryEventInfo messageObj = (ManifestExpiryEventInfo) message.obj;
+        handleManifestExpiredMessage(
+            messageObj.eventTimeUs, messageObj.manifestPublishTimeMsInEmsg);
+        return true;
+      default:
+        // Do nothing.
+    }
+    return false;
+  }
+
+  // Internal methods.
+
+  private void handleManifestExpiredMessage(long eventTimeUs, long manifestPublishTimeMsInEmsg) {
+    if (!manifestPublishTimeToExpiryTimeUs.containsKey(manifestPublishTimeMsInEmsg)) {
+      manifestPublishTimeToExpiryTimeUs.put(manifestPublishTimeMsInEmsg, eventTimeUs);
+    } else {
+      long previousExpiryTimeUs =
+          manifestPublishTimeToExpiryTimeUs.get(manifestPublishTimeMsInEmsg);
+      if (previousExpiryTimeUs > eventTimeUs) {
+        manifestPublishTimeToExpiryTimeUs.put(manifestPublishTimeMsInEmsg, eventTimeUs);
+      }
+    }
+  }
+
+  private void handleMediaPresentationEndedMessageEncountered() {
+    dynamicMediaPresentationEnded = true;
+    notifySourceMediaPresentationEnded();
+  }
+
+  private Map.Entry<Long, Long> ceilingExpiryEntryForPublishTime(long publishTimeMs) {
+    if (manifestPublishTimeToExpiryTimeUs.isEmpty()) {
+      return null;
+    }
+    return manifestPublishTimeToExpiryTimeUs.ceilingEntry(publishTimeMs);
+  }
+
+  private void notifyManifestPublishTimeExpired() {
+    playerEmsgCallback.onDashManifestPublishTimeExpired(expiredManifestPublishTimeUs);
+  }
+
+  private void notifySourceMediaPresentationEnded() {
+    playerEmsgCallback.onDashLiveMediaPresentationEndSignalEncountered();
+  }
+
+  /** Requests DASH media manifest to be refreshed if necessary. */
+  private void maybeNotifyDashManifestRefreshNeeded() {
+    if (lastLoadedChunkEndTimeBeforeRefreshUs != C.TIME_UNSET
+        && lastLoadedChunkEndTimeBeforeRefreshUs == lastLoadedChunkEndTimeUs) {
+      // Already requested manifest refresh.
+      return;
+    }
+    isWaitingForManifestRefresh = true;
+    lastLoadedChunkEndTimeBeforeRefreshUs = lastLoadedChunkEndTimeUs;
+    playerEmsgCallback.onDashManifestRefreshRequested();
+  }
+
+  private static long getManifestPublishTimeMsInEmsg(EventMessage eventMessage) {
+    try {
+      return parseXsDateTime(new String(eventMessage.messageData));
+    } catch (ParserException ignored) {
+      // if we can't parse this event, ignore
+      return C.TIME_UNSET;
+    }
+  }
+
+  private static boolean isMessageSignalingMediaPresentationEnded(EventMessage eventMessage) {
+    // According to section 4.5.2.1 DASH-IF IOP, if both presentation time delta and event duration
+    // are zero, the media presentation is ended.
+    return eventMessage.presentationTimeUs == 0 && eventMessage.durationMs == 0;
+  }
+
+  /** Handles emsg messages for a specific track for the player. */
+  public final class PlayerTrackEmsgHandler implements TrackOutput {
+
+    private final SampleQueue sampleQueue;
+    private final FormatHolder formatHolder;
+    private final MetadataInputBuffer buffer;
+
+    /* package */ PlayerTrackEmsgHandler(SampleQueue sampleQueue) {
+      this.sampleQueue = sampleQueue;
+
+      formatHolder = new FormatHolder();
+      buffer = new MetadataInputBuffer();
+    }
+
+    @Override
+    public void format(Format format) {
+      sampleQueue.format(format);
+    }
+
+    @Override
+    public int sampleData(ExtractorInput input, int length, boolean allowEndOfInput)
+        throws IOException, InterruptedException {
+      return sampleQueue.sampleData(input, length, allowEndOfInput);
+    }
+
+    @Override
+    public void sampleData(ParsableByteArray data, int length) {
+      sampleQueue.sampleData(data, length);
+    }
+
+    @Override
+    public void sampleMetadata(
+        long timeUs, int flags, int size, int offset, CryptoData encryptionData) {
+      sampleQueue.sampleMetadata(timeUs, flags, size, offset, encryptionData);
+      parseAndDiscardSamples();
+    }
+
+    /**
+     * For live streaming, check if the DASH manifest is expired before the next segment start time.
+     * If it is, the DASH media source will be notified to refresh the manifest.
+     *
+     * @param presentationPositionUs The next load position in presentation time.
+     * @return True if manifest refresh has been requested, false otherwise.
+     * @throws DashManifestExpiredException If the current DASH manifest is expired, but a new
+     *     manifest could not be loaded.
+     */
+    public boolean maybeRefreshManifestBeforeLoadingNextChunk(long presentationPositionUs)
+        throws DashManifestExpiredException {
+      return PlayerEmsgHandler.this.maybeRefreshManifestBeforeLoadingNextChunk(
+          presentationPositionUs);
+    }
+
+    /**
+     * Called when the a new chunk in the current media stream has been loaded.
+     *
+     * @param chunk The chunk whose load has been completed.
+     */
+    public void onChunkLoadCompleted(Chunk chunk) {
+      PlayerEmsgHandler.this.onChunkLoadCompleted(chunk);
+    }
+
+    /**
+     * For live streaming with emsg event stream, forward seeking can seek pass the emsg messages
+     * that signals end-of-stream or Manifest expiry, which results in load error. In this case, we
+     * should notify the Dash media source to refresh its manifest.
+     *
+     * @param chunk The chunk whose load encountered the error.
+     * @return True if manifest refresh has been requested, false otherwise.
+     */
+    public boolean maybeRefreshManifestOnLoadingError(Chunk chunk) {
+      return PlayerEmsgHandler.this.maybeRefreshManifestOnLoadingError(chunk);
+    }
+
+    /** Release this track emsg handler. It should not be reused after this call. */
+    public void release() {
+      sampleQueue.reset();
+    }
+
+    // Internal methods.
+
+    private void parseAndDiscardSamples() {
+      while (sampleQueue.hasNextSample()) {
+        MetadataInputBuffer inputBuffer = dequeueSample();
+        if (inputBuffer == null) {
+          continue;
+        }
+        long eventTimeUs = inputBuffer.timeUs;
+        Metadata metadata = decoder.decode(inputBuffer);
+        EventMessage eventMessage = (EventMessage) metadata.get(0);
+        if (isPlayerEmsgEvent(eventMessage.schemeIdUri, eventMessage.value)) {
+          parsePlayerEmsgEvent(eventTimeUs, eventMessage);
+        }
+      }
+      sampleQueue.discardToRead();
+    }
+
+    @Nullable
+    private MetadataInputBuffer dequeueSample() {
+      buffer.clear();
+      int result = sampleQueue.read(formatHolder, buffer, false, false, 0);
+      if (result == C.RESULT_BUFFER_READ) {
+        buffer.flip();
+        return buffer;
+      }
+      return null;
+    }
+
+    private void parsePlayerEmsgEvent(long eventTimeUs, EventMessage eventMessage) {
+      long manifestPublishTimeMsInEmsg = getManifestPublishTimeMsInEmsg(eventMessage);
+      if (manifestPublishTimeMsInEmsg == C.TIME_UNSET) {
+        return;
+      }
+
+      if (isMessageSignalingMediaPresentationEnded(eventMessage)) {
+        onMediaPresentationEndedMessageEncountered();
+      } else {
+        onManifestExpiredMessageEncountered(eventTimeUs, manifestPublishTimeMsInEmsg);
+      }
+    }
+
+    private void onMediaPresentationEndedMessageEncountered() {
+      handler.sendMessage(handler.obtainMessage(EMSG_MEDIA_PRESENTATION_ENDED));
+    }
+
+    private void onManifestExpiredMessageEncountered(
+        long eventTimeUs, long manifestPublishTimeMsInEmsg) {
+      ManifestExpiryEventInfo manifestExpiryEventInfo =
+          new ManifestExpiryEventInfo(eventTimeUs, manifestPublishTimeMsInEmsg);
+      handler.sendMessage(handler.obtainMessage(EMSG_MANIFEST_EXPIRED, manifestExpiryEventInfo));
+    }
+  }
+
+  /** Holds information related to a manifest expiry event. */
+  private static final class ManifestExpiryEventInfo {
+
+    public final long eventTimeUs;
+    public final long manifestPublishTimeMsInEmsg;
+
+    public ManifestExpiryEventInfo(long eventTimeUs, long manifestPublishTimeMsInEmsg) {
+      this.eventTimeUs = eventTimeUs;
+      this.manifestPublishTimeMsInEmsg = manifestPublishTimeMsInEmsg;
+    }
+  }
+}