This article provides an example of how to use the Wowza Streaming Engine™ Java API to access and insert MPEG-TS SCTE-35 tags when streaming an HLS live stream. We recommend using the ModuleAdMarkers class over the legacy stream target API method.
Overview
With the ModuleAdMarkers class, you can prepare Apple HLS streams for ad insertion based on SCTE-35 events in live MPEG-TS source streams. The resulting Apple HLS live streams are written to disk with the chunks broken at ad-insertion points. The media playlist contains EXT-X-CUE-IN, EXT-X-CUE-OUT-CONT, and EXT-X-CUE-OUT custom headers at those ad insertion markers.
This module is beneficial because it allows the splitting of chunks at exact locations so you can switch to an ad halfway through an HLS segment, a task previously impossible with Wowza Streaming Engine versions before 4.8.26. Additionally, you can stream directly from Wowza Streaming Engine and insert the SCTE-35 ad markers without creating a stream target. Bypassing stream target creation helps to achieve typical HLS latency.
Prerequisites
- Wowza Streaming Engine 4.8.26 or later is required.
- All source streams must be MPEG-TS-based and contain SCTE-35 ad markers.
Installation
Copy the following ad markers module and compile it with the Wowza IDE.
package com.wowza.example.cue;
import com.wowza.wms.amf.*;
import com.wowza.wms.application.IApplicationInstance;
import com.wowza.wms.httpstreamer.cupertinostreaming.livestreampacketizer.*;
import com.wowza.wms.media.mp3.model.idtags.ID3Frames;
import com.wowza.wms.module.ModuleBase;
import com.wowza.wms.stream.IMediaStream;
import com.wowza.wms.stream.livepacketizer.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import static com.wowza.wms.amf.AMFData.*;
import static com.wowza.wms.rtp.depacketizer.RTPDePacketizerMPEGTSMonitorCUE.AMF_SCTE_HANDLER_NAME;
public class ModuleAdMarkers extends ModuleBase
{
public static final Class<ModuleAdMarkers> CLASS = ModuleAdMarkers.class;
public static final String CLASS_NAME = CLASS.getSimpleName();
private IApplicationInstance appInstance;
// Call this method when the application starts to set up a listener for live stream packetizer events
public void onAppStart(IApplicationInstance appInstance)
{
getLogger(CLASS, appInstance).info(String.format("%s.onAppStart [%s]", CLASS_NAME, appInstance.getContextStr()));
this.appInstance = appInstance;
appInstance.addLiveStreamPacketizerListener(new LiveStreamPacketizerActionNotifyBase()
{
@Override
public void onLiveStreamPacketizerInit(ILiveStreamPacketizer packetizer, String streamName)
{
// When the live stream packetizer is initialized, check if it's of type LiveStreamPacketizerCupertino
if (packetizer instanceof LiveStreamPacketizerCupertino)
{
IMediaStream stream = appInstance.getStreams().getStream(streamName);
((LiveStreamPacketizerCupertino)packetizer).setDataHandler(new LiveStreamPacketizerCupertinoDataHandlerCue(
(LiveStreamPacketizerCupertino) packetizer, stream));
}
}
});
}
// Define a class to process ad cue-related data during the packetization of the live stream
class LiveStreamPacketizerCupertinoDataHandlerCue implements IHTTPStreamerCupertinoLivePacketizerDataHandler2
{
private final Map<Long, OnCueEvent> events = new ConcurrentHashMap<>();
private final LiveStreamPacketizerCupertino liveStreamPacketizer;
private final IMediaStream stream;
public LiveStreamPacketizerCupertinoDataHandlerCue(LiveStreamPacketizerCupertino liveStreamPacketizer, IMediaStream stream)
{
this.liveStreamPacketizer = liveStreamPacketizer;
this.stream = stream;
}
// Handle the splice event from the source stream
@Override
public void onFillChunkDataPacket(LiveStreamPacketizerCupertinoChunk chunk, CupertinoPacketHolder holder, AMFPacket packet, ID3Frames id3Frames)
{
// Extract the actual SCTE data from the stream packets
extractSCTEData(packet).ifPresent(data -> {
long ptsAdjustment = (long) Math.ceil( data.getDouble("ptsAdjustment") / 90) ;
AMFDataObj commandObj = data.getObject("command");
String command = commandObj.getString("SpliceCommand");
if (command.equalsIgnoreCase("insert"))
{
AMFDataObj eventObj = commandObj.getObject("event");
long eventId = eventObj.getLong("eventID");
// Create new event and add it to the events map
events.computeIfAbsent(eventId, id -> {
boolean spliceOut = eventObj.getBoolean("outOfNetwork");
// Only processing splice_out events, splice_in calculated by the event duration
if (!spliceOut)
return null;
AMFDataObj spliceTimeObj = eventObj.getObject("spliceTime");
long spliceTimecode = spliceTimeObj.getLong("spliceTimeMS");
boolean durationFlag = eventObj.getBoolean("durationFlag");
long breakDuration = durationFlag ? (long) (eventObj.getDouble("breakDuration") / 90) : 0L;
// Don't create an event if we don't have a duration
if (breakDuration <= 0)
return null;
OnCueEvent newEvent = new OnCueEvent(spliceTimecode + ptsAdjustment, breakDuration);
// Check if this event is in the current chunk and split if needed
checkAdjustChunkEnd(newEvent.startTime);
return newEvent;
});
}
});
}
private Optional<AMFDataObj> extractSCTEData(AMFPacket packet)
{
byte[] buffer = packet.getData();
if (buffer == null || packet.getSize() <= 2)
return Optional.empty();
int offset = buffer[0] != 0 ? 0 : 1;
AMFDataList amfList = new AMFDataList(buffer, offset, buffer.length - offset);
if (amfList.size() <= 1)
return Optional.empty();
if (amfList.get(0).getType() != DATA_TYPE_STRING && amfList.get(1).getType() != DATA_TYPE_OBJECT)
return Optional.empty();
String handlerName = amfList.getString(0);
AMFDataObj data = amfList.getObject(1);
if (!handlerName.equalsIgnoreCase(AMF_SCTE_HANDLER_NAME))
return Optional.empty();
getLogger(CLASS, appInstance).info(String.format("%s.extractSCTEData [%s] timecode: %d, data: %s", CLASS_NAME, stream.getContextStr(), packet.getAbsTimecode(), data));
return Optional.ofNullable(data);
}
// Modify the segment stop timecode for the current segment so the new segment starts at the correct point
private void checkAdjustChunkEnd(long timecode)
{
long chunkStart = liveStreamPacketizer.getSegmentStartKeyTimecode();
long chunkEnd = liveStreamPacketizer.getSegmentStopKeyTimecode();
long minDuration = liveStreamPacketizer.getMinChunkDuration();
long targetDuration = liveStreamPacketizer.getChunkDurationTarget();
if (timecode < chunkStart || timecode == chunkEnd)
return;
if (timecode == chunkStart)
{
if ((chunkEnd - chunkStart < minDuration))
liveStreamPacketizer.setSegmentStopKeyTimecode(chunkStart + minDuration);
return;
}
// If split is in the next chunk, that chunk would be too short (split too close to the start)
if ((timecode > chunkEnd && timecode < chunkEnd + minDuration))
{
// If currentDuration is shorter than the target, extend to timecode
if (timecode - chunkStart <= targetDuration)
{
getLogger(CLASS, appInstance).info(String.format("%s.checkAdjustChunkEnd [%s] Timecode within the target duration. extending current chunk: timecode: %d, chunkStart: %d, chunkEnd: %d", CLASS_NAME, stream.getContextStr(), timecode, chunkStart, chunkEnd));
liveStreamPacketizer.setSegmentStopKeyTimecode(timecode);
}
// If the current chunk is larger than the minDuration, then we need to split the current chunk early so the next chunk won't be too small
else if (chunkEnd - chunkStart >= minDuration)
{
getLogger(CLASS, appInstance).info(String.format("%s.checkAdjustChunkEnd [%s] next chunk will be too small. Splitting current chunk: minDuration: %d, chunkStart: %d, chunkEnd: %d", CLASS_NAME, stream.getContextStr(), minDuration, chunkStart, chunkEnd));
liveStreamPacketizer.setSegmentStopKeyTimecode(chunkEnd - minDuration);
}
}
else if (timecode < chunkEnd)
{
getLogger(CLASS, appInstance).info(String.format("%s.checkAdjustChunkEnd [%s] timecode: %d, chunkStart: %d, chunkEnd: %d", CLASS_NAME, stream.getContextStr(), timecode, chunkStart, chunkEnd));
liveStreamPacketizer.setSegmentStopKeyTimecode(timecode);
}
// If a previous split left, this chunk too short (split too close to the end of the chunk)
else if (chunkEnd - chunkStart < minDuration)
{
getLogger(CLASS, appInstance).info(String.format("%s.checkAdjustChunkEnd [%s] this chunk is too small. Extending current chunk: minDuration: %d, chunkStart: %d, chunkEnd: %d", CLASS_NAME, stream.getContextStr(), minDuration, chunkStart, chunkEnd));
liveStreamPacketizer.setSegmentStopKeyTimecode(chunkStart + minDuration);
}
}
// Detect an event in the stream and store it in a map with an event ID
@Override
public void onFillChunkStart(LiveStreamPacketizerCupertinoChunk chunk)
{
// Loop through events and see if we need to split the current chunk for the event start or end
events.forEach((id, event) -> {
checkAdjustChunkEnd(event.startTime);
checkAdjustChunkEnd(event.startTime + event.duration);
});
// Clear any events that expired in the previous chunk
events.entrySet().removeIf(e -> e.getValue().expired);
}
// Determine if we've got to add a tag for a specific segment or chunk
// Depending on where we're in the event, we create a EXT-X-CUE-OUT, EXT-X-OUT-CONT, or EXT-X-CUE-IN tag
@Override
public void onFillChunkEnd(LiveStreamPacketizerCupertinoChunk chunk, long timecode)
{
CupertinoUserManifestHeaders chunkHeaders = chunk.getUserManifestHeaders();
events.forEach((id, event) -> {
long elapsed = chunk.getStartTimecode() - event.startTime;
// Define first chunk for event
// Add EXT-X-CUE-OUT tag the very first time we encounter the event to signal we should go to an ad break
if (event.startTime >= chunk.getStartTimecode() && event.startTime < timecode)
{
String tag = String.format("EXT-X-CUE-OUT:%.3f", event.duration / 1000d);
chunkHeaders.addHeader(tag);
}
// Define continuation chunk
// As the ad progresses, add the EXT-X-CUE-OUT-CONT tag for each chunk to signal elapsed time/duration and inform the player we're still in the ad break
else if (elapsed > 0 && elapsed < event.duration)
{
String tag = String.format("EXT-X-CUE-OUT-CONT: %.3f/%.3f", elapsed / 1000d, event.duration / 1000d);
chunkHeaders.addHeader(tag);
}
// Define first chunk after event
// At the end of the ad break, add the EXT-X-CUE-IN tag to say we've hit the ad duration and are back in the program
else if (elapsed >= event.duration)
{
chunkHeaders.addHeader("EXT-X-CUE-IN");
event.expired = true;
}
});
}
// Method included but not used in the module
@Override
public void onFillChunkMediaPacket(LiveStreamPacketizerCupertinoChunk chunk, CupertinoPacketHolder holder, AMFPacket packet)
{
// no-op
}
}
// Create class to ad marker events and contain information about start time, duration, and expiration status of the event
static class OnCueEvent
{
final long startTime;
final long duration;
boolean expired = false;
public OnCueEvent(long startTime, long duration)
{
this.startTime = startTime;
this.duration = duration;
}
@Override
public String toString()
{
return "OnCueEvent{" +
"startTime=" + startTime +
", duration=" + duration +
", expired=" + expired +
'}';
}
}
}
Configuration
Add the following definition to your application configuration to enable this module in Wowza Streaming Engine Manager. See Configure modules for details.
|
Name
|
ModuleAdMarkers
|
| Description | Inserts SCTE-35 ad markers for HLS live streams. |
| Fully Qualified Class Name | com.wowza.example.cue.ModuleAdMarkers |
Properties
After enabling the module, add these custom properties to your application in Wowza Streaming Engine Manager. See Add a custom property for details.
|
Path
|
/Root/Application/RTP
|
| Name | rtpDePacketizerMPEGTSListenerClass |
| Type | String |
| Value | com.wowza.wms.rtp.depacketizer.RTPDePacketizerMPEGTSMonitorCUE |
| Notes | Default MPEG-TS monitor implementation for ad marker (SCTE-35) ingestion. |
| Path | /Root/Application/RTP |
| Name | rtpDePacketizerMPEGTSMonitorCUEDebugLog |
| Type | Boolean |
| Value | true |
| Notes | Enables debugging and logging for the RTPDePacketizerMPEGTSMonitorCUE class. |
Usage
With a successful setup, you can start a stream on your application and query a Cupertino-packetized HLS stream to view the SCTE tags.
See About Apple HLS headers for steps to request the playlist and then the chunklist in the initial server response. You can run a cURL command against any Cupertino-packetized HLS playback URL or media playlist to get information about the stream's contents or playlist.
The Wowza Streaming Engine chunklist response to the client includes the new SCTE-35 headers. From a server-side ad insertion point of view, you can take these markers and replace each corresponding segment by swapping it with an actual ad.
For instance, this example HLS stream starts out not in an ad break. It then goes to a 110-second ad break when the EXT-X-CUE-OUT tag is inserted:
#EXTM3U
#EXT-X-VERSION: 3
#EXT-X-TARGETDURATION: 7
#EXT-X-MEDIA-SEQUENCE: 39
#EXT-X-DISCONTINUITY-SEQUENCE: 0
#EXTINF:6.0,
media_w1775459898_39.ts
#EXTINF:5.667,
media_w1775459898_40.ts
#EXTINF: 3.0,
#EXT-X-CUE-OUT: 110.000
media_w1775459898_41.ts
In the ad break, the EXT-X-CUE-OUT-CONT tag then indicates the elapsed time and duration for the advertisement:
#EXTINF:3.133,
#EXT-X-CUE-OUT-CONT: 0.001/110.000
media_w1775459898_42.ts
#EXTINF:5.967,
#EXT-X-CUE-OUT-CONT: 3.134/110.000
media_w1775459898_43.ts
#EXTINF:6.0,
#EXT-X-CUE-OUT-CONT: 9.101/110.000
media_w1775459898_44.ts
When the ad duration is up, the EXT-X-CUE-IN marker gets us back into the program, and the ad break is finished:
#EXTINF:5.833,
#EXT-X-CUE-OUT-CONT: 99.101/110.000
media_w1775459898_59.ts
#EXTINF:5.067,
#EXT-X-CUE-OUT-CONT: 104.934/110.000
media_w1775459898_60.ts
#EXTINF:1.0,
#EXT-X-CUE-IN
media_w1775459898_61.ts
#EXTINF:6.0,
media_w1775459898_62.ts
#EXTINF:6.0,
media_w1775459898_63.ts
