Current section
Files
Jump to
Current section
Files
lib/hls/packager.ex
defmodule HLS.Packager do
@moduledoc """
A fully functional HLS packager that performs pure state transformations.
Instead of performing I/O operations, this module returns `Action`s that the caller
must execute. This design provides:
- Pure functions (no side effects)
- Testability without mocking
- Explicit control flow
- Caller-controlled concurrency
- Ability to batch operations
## Usage Flow
```elixir
# 1. Initialize
{:ok, state} = Packager.new(
manifest_uri: URI.new!("stream.m3u8"),
max_segments: 10
)
# 2. Add tracks
{state, []} = Packager.add_track(state, "video",
stream: %VariantStream{...},
segment_extension: ".m4s",
target_segment_duration: 6
)
# 3. Add segment (returns upload action)
{state, [action]} = Packager.put_segment(state, "video", duration: 5.2, pts: 0)
# action = %Action.UploadSegment{
# id: "video_seg_1",
# uri: URI.parse("stream_video/00000/stream_video_00001.m4s"),
# track_id: "video"
# }
# 4. Caller uploads the segment with their payload
:ok = Storage.put(storage, action.uri, video_payload)
# 5. Confirm upload (may trigger playlist writes)
{state, actions} = Packager.confirm_upload(state, action.id)
# actions = [%Action.WritePlaylist{uri: ..., content: ...}]
# 6. Execute write actions
Enum.each(actions, fn
%Action.WritePlaylist{uri: uri, content: content} ->
Storage.put(storage, uri, content)
end)
# 7. Sync (triggers media playlist updates)
{state, actions} = Packager.sync(state, 3)
# actions = [
# %Action.WritePlaylist{type: :media, ...},
# %Action.WritePlaylist{type: :master, ...},
# %Action.DeleteSegment{uri: ...} # if sliding window
# ]
```
## Actions
All operations return `{new_state, [Action.t()]}` where actions must be executed
by the caller in order.
"""
alias HLS.{Segment, VariantStream, AlternativeRendition}
alias HLS.Playlist.{Media, Master}
# Action types returned by operations
defmodule Action do
@moduledoc "Actions that the caller must execute"
defmodule UploadSegment do
@moduledoc "Upload a media segment"
defstruct [:id, :uri, :track_id, :init_section_uri]
@type t :: %__MODULE__{
id: String.t(),
uri: URI.t(),
track_id: String.t(),
init_section_uri: URI.t() | nil
}
end
defmodule UploadInitSection do
@moduledoc "Upload an initialization section"
defstruct [:id, :uri, :track_id]
@type t :: %__MODULE__{id: String.t(), uri: URI.t(), track_id: String.t()}
end
defmodule WritePlaylist do
@moduledoc "Write a playlist"
defstruct [:type, :uri, :content]
@type playlist_type :: :master | :media | :pending
@type t :: %__MODULE__{
type: playlist_type(),
uri: URI.t(),
content: String.t()
}
end
defmodule DeleteSegment do
@moduledoc "Delete a segment"
defstruct [:uri, :track_id]
@type t :: %__MODULE__{uri: URI.t(), track_id: String.t()}
end
defmodule DeleteInitSection do
@moduledoc "Delete an init section"
defstruct [:uri, :track_id]
@type t :: %__MODULE__{uri: URI.t(), track_id: String.t()}
end
defmodule DeletePlaylist do
@moduledoc "Delete a playlist"
defstruct [:uri, :type]
@type t :: %__MODULE__{uri: URI.t(), type: :media | :pending | :master}
end
@type t ::
UploadSegment.t()
| UploadInitSection.t()
| WritePlaylist.t()
| DeleteSegment.t()
| DeleteInitSection.t()
| DeletePlaylist.t()
end
defmodule Error do
@moduledoc """
RFC 8216 compliance error or stall signal.
These errors are returned instead of producing non-compliant playlists.
"""
defexception [:code, :message, :details]
@type code ::
:segment_duration_over_target
| :timing_drift
| :discontinuity_point_missed
| :track_timing_mismatch_at_sync
| :mandatory_track_missing_segment_at_sync
| :sync_point_skipped
| :upload_id_not_found
| :track_conflict
| :resume_missing_playlist
| :resume_unexpected_playlist
| :resume_unknown_segment_extension
| :resume_inconsistent_playlist
| :resume_track_not_ready
@type t :: %__MODULE__{code: code(), message: String.t(), details: map()}
end
defmodule Track do
@type t :: %__MODULE__{
stream: VariantStream.t() | AlternativeRendition.t(),
duration: float(),
segment_count: non_neg_integer(),
segment_extension: String.t() | nil,
init_section: %{uri: URI.t()} | nil,
media_playlist: Media.t(),
pending_playlist: Media.t(),
pending_segments: [%{segment: Segment.t(), uploaded?: boolean(), id: String.t()}],
codecs: [String.t()],
mandatory?: boolean(),
last_timestamp_ns: non_neg_integer() | nil,
last_duration_ns: non_neg_integer() | nil,
base_pdt: DateTime.t() | nil,
base_timestamp_ns: non_neg_integer() | nil,
applied_discontinuities: MapSet.t(),
resume_incomplete?: boolean()
}
defstruct [
:stream,
:duration,
:segment_count,
:segment_extension,
:init_section,
:media_playlist,
:pending_playlist,
:base_pdt,
:base_timestamp_ns,
:mandatory?,
:last_timestamp_ns,
:last_duration_ns,
:applied_discontinuities,
resume_incomplete?: false,
pending_segments: [],
codecs: []
]
end
@type track_id :: String.t()
@type t :: %__MODULE__{
master_written?: boolean(),
manifest_uri: URI.t(),
tracks: %{track_id() => Track.t()},
max_segments: pos_integer() | nil,
timeline_reference: DateTime.t(),
timing_tolerance_ns: non_neg_integer(),
pending_discontinuities: list(),
skipped_sync_points: %{pos_integer() => MapSet.t(track_id())}
}
defstruct [
:manifest_uri,
:timeline_reference,
:timing_tolerance_ns,
master_written?: false,
tracks: %{},
max_segments: nil,
pending_discontinuities: [],
skipped_sync_points: %{}
]
@doc """
Creates a new packager state.
## Options
- `:manifest_uri` (required) - URI of the master playlist
- `:max_segments` - Maximum segments per media playlist (nil = unlimited)
- `:timing_tolerance_ms` - Allowed timing drift in milliseconds (default: 200)
## Examples
{:ok, state} = Packager.new(
manifest_uri: URI.new!("stream.m3u8"),
max_segments: 10
)
"""
@spec new(keyword()) :: {:ok, t()} | {:error, term()}
def new(opts) do
with {:ok, validated} <- validate_new_opts(opts) do
timeline_reference = DateTime.utc_now(:millisecond)
state = %__MODULE__{
manifest_uri: validated.manifest_uri,
max_segments: validated.max_segments,
timeline_reference: timeline_reference,
timing_tolerance_ns: validated.timing_tolerance_ns,
tracks: %{},
master_written?: false,
pending_discontinuities: [],
skipped_sync_points: %{}
}
{:ok, state}
end
end
@doc """
Resumes from existing playlists loaded by the caller.
The caller is responsible for loading the master playlist and all media playlists.
This function reconstructs the state from the loaded data.
## Examples
# Caller loads playlists
master = load_master_playlist(master_uri)
media_playlists =
Enum.map(master.streams, fn stream ->
load_media_playlist(stream.uri)
end)
# Resume
{:ok, state} = Packager.resume(
master_playlist: master,
media_playlists: media_playlists,
max_segments: 10
)
"""
@spec resume(keyword()) :: {:ok, t()} | {:error, term()}
def resume(opts) do
timeline_reference = DateTime.utc_now(:millisecond)
with {:ok, validated} <- validate_resume_opts(opts),
{:ok, {tracks, common_sync_point}} <-
build_tracks_from_playlists(Map.put(validated, :timeline_reference, timeline_reference)) do
master = validated.master_playlist
state = %__MODULE__{
manifest_uri: master.uri,
max_segments: validated.max_segments,
timeline_reference: timeline_reference,
timing_tolerance_ns: validated.timing_tolerance_ns,
master_written?: true,
tracks: tracks,
pending_discontinuities: [],
skipped_sync_points: %{}
}
case schedule_discontinuity(state, common_sync_point + 1, :resume, timeline_reference) do
{:ok, new_state} ->
{:ok, %{new_state | timeline_reference: timeline_reference}}
{:error, error, _state} ->
{:error, error}
end
end
end
@doc """
Adds a track to the packager. Returns updated state with no actions.
Tracks can only be added before the master playlist is written.
## Examples
{state, []} = Packager.add_track(state, "video_720p",
stream: %VariantStream{
bandwidth: 2_500_000,
resolution: {1280, 720},
codecs: ["avc1.64001f", "mp4a.40.2"]
},
segment_extension: ".m4s",
target_segment_duration: 6.0,
codecs: ["avc1.64001f"],
mandatory?: true
)
"""
@spec add_track(t(), track_id(), keyword()) :: {t(), []}
def add_track(state, track_id, opts) do
if Map.has_key?(state.tracks, track_id) do
opts = validate_add_track_opts!(opts)
{stream, media_playlist_uri} = build_track_stream(state, track_id, opts.stream)
mandatory? = resolve_mandatory(opts.mandatory?, stream)
track = state.tracks[track_id]
cond do
track.resume_incomplete? ->
case reconcile_incomplete_track(
track,
track_id,
stream,
media_playlist_uri,
opts,
mandatory?
) do
{:ok, updated_track} ->
new_state = put_in(state.tracks[track_id], updated_track)
{new_state, []}
{:error, error} ->
raise error
end
track_spec_matches?(track, stream, media_playlist_uri, opts, mandatory?) ->
{state, []}
true ->
raise Error,
code: :track_conflict,
message: "Track #{track_id} already exists with a different specification",
details: %{track_id: track_id}
end
else
if state.master_written? do
raise "Cannot add track after master playlist is written"
end
opts = validate_add_track_opts!(opts)
{stream, media_playlist_uri} = build_track_stream(state, track_id, opts.stream)
type = if state.max_segments, do: nil, else: :event
media_playlist = %Media{
uri: media_playlist_uri,
target_segment_duration: opts.target_segment_duration,
type: type
}
track = %Track{
stream: stream,
duration: 0.0,
segment_count: 0,
segment_extension: opts.segment_extension,
init_section: nil,
media_playlist: media_playlist,
pending_playlist: %{media_playlist | uri: append_to_path(media_playlist.uri, "_pending")},
codecs: opts.codecs,
mandatory?: resolve_mandatory(opts.mandatory?, stream),
last_timestamp_ns: nil,
last_duration_ns: nil,
base_pdt: state.timeline_reference,
base_timestamp_ns: nil,
pending_segments: [],
applied_discontinuities: MapSet.new(),
resume_incomplete?: false
}
new_state = put_in(state.tracks[track_id], track)
{new_state, []}
end
end
@doc """
Prepares a new init section upload.
Returns an action to upload the init section. The caller must upload the payload
and then call `confirm_init_upload/2`.
## Examples
{state, [action]} = Packager.put_init_section(state, "video")
# action = %Action.UploadInitSection{
# id: "video_init_1",
# uri: URI.parse("stream_video/00000/stream_video_00001_init.mp4"),
# track_id: "video"
# }
# Caller uploads
:ok = Storage.put(storage, action.uri, init_payload)
# Confirm
{state, []} = Packager.confirm_init_upload(state, action.id)
"""
@spec put_init_section(t(), track_id()) :: {t(), [Action.UploadInitSection.t()]}
def put_init_section(state, track_id) do
track = Map.fetch!(state.tracks, track_id)
next_index = track.segment_count + 1
extname =
case track.segment_extension do
".mp4" -> ".mp4"
".m4s" -> ".mp4"
other -> raise "Init section not supported for #{other}"
end
segment_uri = relative_segment_uri(track.media_playlist.uri, extname, next_index)
uri = append_to_path(segment_uri, "_init")
upload_id = "#{track_id}_init_#{next_index}"
action = %Action.UploadInitSection{
id: upload_id,
uri: uri,
track_id: track_id
}
# Store pending init section info
new_state =
update_in(state.tracks[track_id], fn t ->
%{t | init_section: %{uri: uri, pending_upload_id: upload_id}}
end)
{new_state, [action]}
end
@doc """
Confirms that an init section upload completed.
"""
@spec confirm_init_upload(t(), String.t()) :: {t(), []}
def confirm_init_upload(state, upload_id) do
# Find track with this upload_id
track_id =
Enum.find_value(state.tracks, fn {id, track} ->
if track.init_section && track.init_section[:pending_upload_id] == upload_id do
id
end
end)
if track_id do
new_state =
update_in(state.tracks[track_id], fn track ->
init_section = Map.drop(track.init_section, [:pending_upload_id])
%{track | init_section: init_section}
end)
{new_state, []}
else
{state, []}
end
end
@doc """
Adds a segment to a track and returns an upload action.
The caller must upload the segment payload and then call `confirm_upload/2`.
The caller must provide PTS in nanoseconds; DTS may be provided for video and
is used instead of PTS when present.
Returns an error when RFC 8216 compliance issues are detected.
## Examples
{state, [action]} = Packager.put_segment(state, "video", duration: 5.2, pts: 0)
# Caller uploads
:ok = Storage.put(storage, action.uri, segment_payload)
# Confirm (may trigger playlist writes)
{state, actions} = Packager.confirm_upload(state, action.id)
"""
@spec put_segment(t(), track_id(), keyword()) ::
{t(), [Action.UploadSegment.t()]}
| {:warning, Error.t(), t()}
| {:error, Error.t(), t()}
def put_segment(state, track_id, opts) do
track = Map.fetch!(state.tracks, track_id)
duration = Keyword.fetch!(opts, :duration)
pts = Keyword.fetch!(opts, :pts)
dts = Keyword.get(opts, :dts)
timestamp_ns = dts || pts
case resume_track_ready?(track, track_id) do
{:error, error} ->
{:error, error, state}
:ok ->
case skip_sync_point_for_track(state, track_id, track.segment_count + 1) do
{:skip, new_state} ->
warning = %Error{
code: :sync_point_skipped,
message: "Sync point #{track.segment_count + 1} was skipped for track '#{track_id}'",
details: %{
track_id: track_id,
sync_point: track.segment_count + 1
}
}
{:warning, warning, new_state}
:ok ->
# RFC 8216: EXTINF duration rounded to nearest integer MUST be <= target duration
if round(duration) > track.media_playlist.target_segment_duration do
error = %Error{
code: :segment_duration_over_target,
message:
"Segment duration #{duration}s exceeds target #{track.media_playlist.target_segment_duration}s (RFC 8216 violation)",
details: %{
track_id: track_id,
segment_index: track.segment_count + 1,
duration: duration,
target_duration: track.media_playlist.target_segment_duration
}
}
{:error, error, state}
else
duration_ns = duration_to_ns(duration)
if track.last_timestamp_ns && track.last_duration_ns do
expected = track.last_timestamp_ns + track.last_duration_ns
if abs(timestamp_ns - expected) > state.timing_tolerance_ns do
error = %Error{
code: :timing_drift,
message:
"Timing drift detected on track '#{track_id}' at segment #{track.segment_count + 1}",
details: %{
track_id: track_id,
segment_index: track.segment_count + 1,
timestamp_ns: timestamp_ns,
expected_ns: expected,
tolerance_ns: state.timing_tolerance_ns
}
}
{:error, error, state}
else
do_put_segment(
state,
track_id,
track,
duration,
duration_ns,
pts,
dts,
timestamp_ns
)
end
else
do_put_segment(state, track_id, track, duration, duration_ns, pts, dts, timestamp_ns)
end
end
end
end
end
defp do_put_segment(state, track_id, track, duration, duration_ns, pts, dts, timestamp_ns) do
next_index = track.segment_count + 1
segment_uri =
relative_segment_uri(track.media_playlist.uri, track.segment_extension, next_index)
{discontinuity, state, segment_pdt, base_pdt, base_timestamp_ns} =
apply_discontinuity_if_needed(state, track_id, next_index, timestamp_ns, track)
segment = %Segment{
uri: segment_uri,
duration: Float.round(duration, 5),
pts: pts,
dts: dts,
init_section: if(track.init_section, do: %{uri: track.init_section.uri}),
discontinuity: discontinuity,
program_date_time: segment_pdt
}
upload_id = "#{track_id}_seg_#{next_index}"
pending_segment = %{
segment: segment,
uploaded?: false,
id: upload_id
}
new_state =
update_in(state.tracks[track_id], fn t ->
t
|> Map.update!(:segment_count, &(&1 + 1))
|> Map.put(:base_pdt, base_pdt)
|> Map.put(:base_timestamp_ns, base_timestamp_ns)
|> Map.put(:last_timestamp_ns, timestamp_ns)
|> Map.put(:last_duration_ns, duration_ns)
|> Map.update!(:pending_segments, &(&1 ++ [pending_segment]))
end)
action = %Action.UploadSegment{
id: upload_id,
uri: segment_uri,
track_id: track_id,
init_section_uri: segment.init_section && segment.init_section.uri
}
{new_state, [action]}
end
@doc """
Confirms that a segment upload completed.
May return playlist write actions if enough segments are buffered.
## Examples
{state, actions} = Packager.confirm_upload(state, "video_seg_1")
# actions = [
# %Action.WritePlaylist{type: :pending, uri: ..., content: ...}
# ]
"""
@spec confirm_upload(t(), String.t()) ::
{t(), [Action.WritePlaylist.t()]} | {:warning, Error.t(), t()}
def confirm_upload(state, upload_id) do
# Find track with this segment
case Enum.find(state.tracks, fn {_id, track} ->
Enum.any?(track.pending_segments, &(&1.id == upload_id))
end) do
nil ->
warning = %Error{
code: :upload_id_not_found,
message: "Upload #{upload_id} not found",
details: %{upload_id: upload_id}
}
{:warning, warning, state}
{track_id, _track} ->
# Update the specific track
updated_track =
state.tracks[track_id]
|> Map.update!(:pending_segments, fn segments ->
Enum.map(segments, fn seg ->
if seg.id == upload_id, do: %{seg | uploaded?: true}, else: seg
end)
end)
# Check if we can move uploaded segments to pending playlist
{uploaded, still_pending} =
Enum.split_while(updated_track.pending_segments, & &1.uploaded?)
if Enum.empty?(uploaded) do
new_state = %{state | tracks: Map.put(state.tracks, track_id, updated_track)}
{new_state, []}
else
uploaded_segments = Enum.map(uploaded, & &1.segment)
new_pending_playlist = %{
updated_track.pending_playlist
| segments: updated_track.pending_playlist.segments ++ uploaded_segments
}
final_track =
updated_track
|> Map.put(:pending_segments, still_pending)
|> Map.put(:pending_playlist, new_pending_playlist)
new_state = %{state | tracks: Map.put(state.tracks, track_id, final_track)}
# Generate write action for pending playlist (if not in sliding window mode)
actions =
if state.max_segments do
[]
else
[
%Action.WritePlaylist{
type: :pending,
uri: new_pending_playlist.uri,
content: HLS.Playlist.marshal(new_pending_playlist)
}
]
end
{new_state, actions}
end
end
end
@doc """
Marks all tracks to add discontinuity to next segment.
"""
@spec discontinue(t()) :: {t(), []} | {:error, Error.t(), t()}
def discontinue(state) do
timeline_reference = DateTime.utc_now(:millisecond)
sync_point = next_discontinuity_sync_point(state)
case schedule_discontinuity(state, sync_point, :manual, timeline_reference) do
{:ok, new_state} ->
{%{new_state | timeline_reference: timeline_reference}, []}
{:error, error, _state} ->
{:error, error, state}
end
end
@doc """
Skips a sync point across all tracks and schedules a discontinuity.
Callers should invoke this when a segment is rejected for RFC compliance,
so all tracks drop the same segment index before continuing.
"""
@spec skip_sync_point(t(), pos_integer()) :: {t(), []} | {:error, Error.t(), t()}
def skip_sync_point(state, sync_point) when sync_point > 0 do
passed_tracks =
state.tracks
|> Enum.filter(fn {_id, track} -> length(track.media_playlist.segments) >= sync_point end)
|> Enum.map(&elem(&1, 0))
if passed_tracks != [] do
error = %Error{
code: :discontinuity_point_missed,
message:
"Discontinuity at segment #{sync_point} cannot be synchronized across all tracks",
details: %{
sync_point: sync_point,
reason: :skip,
tracks_ahead: passed_tracks
}
}
{:error, error, state}
else
skip_existing? =
Enum.all?(state.tracks, fn {_id, track} -> track.segment_count >= sync_point end)
state =
state
|> then(fn current_state ->
if skip_existing?, do: current_state, else: mark_skipped_sync_point(current_state, sync_point)
end)
|> drop_pending_segments_at_sync_point(sync_point)
discontinuity_sync_point = if skip_existing?, do: sync_point + 1, else: sync_point
apply_sync_point = if skip_existing?, do: sync_point, else: discontinuity_sync_point
timeline_reference = DateTime.utc_now(:millisecond)
case schedule_discontinuity(state, discontinuity_sync_point, :skip, timeline_reference) do
{:ok, new_state} ->
new_state =
new_state
|> Map.put(:timeline_reference, timeline_reference)
|> maybe_apply_pending_discontinuity(discontinuity_sync_point, apply_sync_point)
{new_state, []}
{:error, error, _state} ->
{:error, error, state}
end
end
end
defp mark_skipped_sync_point(state, sync_point) do
Map.update(
state,
:skipped_sync_points,
%{sync_point => MapSet.new(Map.keys(state.tracks))},
fn skips ->
Map.put_new(skips, sync_point, MapSet.new(Map.keys(state.tracks)))
end
)
end
defp drop_pending_segments_at_sync_point(state, sync_point) do
update_in(state.tracks, fn tracks ->
tracks
|> Enum.map(fn {id, track} ->
{id, drop_track_segment_at_sync_point(track, sync_point)}
end)
|> Map.new()
end)
end
defp drop_track_segment_at_sync_point(track, sync_point) do
media_count = length(track.media_playlist.segments)
pending_playlist_count = length(track.pending_playlist.segments)
relative_index = sync_point - media_count - 1
cond do
relative_index < 0 ->
track
relative_index < pending_playlist_count ->
{_removed, remaining} = List.pop_at(track.pending_playlist.segments, relative_index)
%{track | pending_playlist: %{track.pending_playlist | segments: remaining}}
relative_index < pending_playlist_count + length(track.pending_segments) ->
pending_idx = relative_index - pending_playlist_count
{_removed, remaining} = List.pop_at(track.pending_segments, pending_idx)
%{track | pending_segments: remaining}
true ->
track
end
end
defp maybe_apply_pending_discontinuity(state, discontinuity_sync_point, apply_sync_point) do
case find_discontinuity(state.pending_discontinuities, discontinuity_sync_point) do
nil ->
state
discontinuity ->
updated_tracks =
state.tracks
|> Enum.map(fn {id, track} ->
{id, apply_discontinuity_to_track_segment(track, discontinuity, apply_sync_point)}
end)
|> Map.new()
%{state | tracks: updated_tracks}
|> maybe_pop_discontinuities()
end
end
defp apply_discontinuity_to_track_segment(track, discontinuity, apply_sync_point) do
media_count = length(track.media_playlist.segments)
pending_playlist_count = length(track.pending_playlist.segments)
relative_index = apply_sync_point - media_count - 1
cond do
relative_index < 0 ->
track
relative_index < pending_playlist_count ->
{segment, remaining} = List.pop_at(track.pending_playlist.segments, relative_index)
timestamp_ns = segment_timestamp_ns(segment)
updated_segment = %{
segment
| discontinuity: true,
program_date_time: discontinuity.pdt_reference
}
updated_segments = List.insert_at(remaining, relative_index, updated_segment)
%{
track
| pending_playlist: %{track.pending_playlist | segments: updated_segments},
base_pdt: discontinuity.pdt_reference,
base_timestamp_ns: timestamp_ns,
applied_discontinuities: MapSet.put(track.applied_discontinuities, discontinuity.id)
}
relative_index < pending_playlist_count + length(track.pending_segments) ->
pending_idx = relative_index - pending_playlist_count
{pending_segment, remaining} = List.pop_at(track.pending_segments, pending_idx)
timestamp_ns = segment_timestamp_ns(pending_segment.segment)
updated_segment = %{
pending_segment.segment
| discontinuity: true,
program_date_time: discontinuity.pdt_reference
}
updated_pending_segment = %{pending_segment | segment: updated_segment}
updated_pending_segments = List.insert_at(remaining, pending_idx, updated_pending_segment)
%{
track
| pending_segments: updated_pending_segments,
base_pdt: discontinuity.pdt_reference,
base_timestamp_ns: timestamp_ns,
applied_discontinuities: MapSet.put(track.applied_discontinuities, discontinuity.id)
}
true ->
track
end
end
@doc """
Synchronizes all tracks to the specified segment index.
Moves segments from pending playlists to media playlists and returns actions
to write updated playlists. May also return delete actions if sliding window
is enabled.
## Examples
{state, actions} = Packager.sync(state, 5)
# actions = [
# %Action.WritePlaylist{type: :media, ...},
# %Action.WritePlaylist{type: :media, ...},
# %Action.WritePlaylist{type: :master, ...},
# %Action.DeleteSegment{...}, # if sliding window
# ]
"""
@spec sync(t(), pos_integer()) ::
{t(), [Action.t()]} | {:warning, [Error.t()], t()} | {:error, Error.t(), t()}
def sync(state, sync_point) do
case sync_if_ready(state, sync_point) do
{:warning, warnings} ->
{:warning, warnings, state}
{:error, error} ->
{:error, error, state}
:ready ->
do_sync(state, sync_point)
end
end
defp do_sync(state, sync_point) do
# Move segments to media playlists
{new_tracks, track_actions} =
state.tracks
|> Enum.map(fn {id, track} ->
sync_track(state, track, id, sync_point)
end)
|> Enum.unzip()
new_state = %{state | tracks: Map.new(new_tracks)}
# Flatten track actions
all_track_actions = List.flatten(track_actions)
# Maybe write master playlist
{new_state, master_actions} = maybe_write_master(new_state, sync_point)
{new_state, all_track_actions ++ master_actions}
end
@doc """
Returns the next synchronization point.
This is the maximum segment count across all tracks plus one.
"""
@spec next_sync_point(t()) :: pos_integer()
def next_sync_point(state) do
max_segments =
state.tracks
|> Enum.map(fn {_id, track} ->
length(track.media_playlist.segments)
end)
|> Enum.max(fn -> 0 end)
max_segments + 1
end
@doc """
Checks whether selected tracks have enough available segments to sync.
By default, only variant (video) tracks are checked.
Returns `{ready?, lagging_track_ids}`.
"""
@spec sync_ready?(t(), pos_integer(), (track_id(), Track.t() -> boolean())) ::
{boolean(), [track_id()]}
def sync_ready?(state, sync_point, track_filter \\ &variant_track?/2) do
lagging =
state.tracks
|> Enum.filter(fn {id, track} -> track_filter.(id, track) end)
|> Enum.reduce([], fn {id, track}, acc ->
available =
length(track.pending_playlist.segments) + length(track.media_playlist.segments)
if available < sync_point, do: [id | acc], else: acc
end)
|> Enum.reverse()
{lagging == [], lagging}
end
@doc """
Flushes the packager, finalizing all playlists.
When `max_segments` is nil (unlimited):
- Creates VOD playlists with EXT-X-ENDLIST
- Returns write actions for final playlists
When `max_segments` is set (sliding window):
- Returns delete actions for all segments and playlists
- Resets state to initial condition
## Examples
# VOD mode
{state, actions} = Packager.flush(state)
# actions = [
# %Action.WritePlaylist{type: :media, ...}, # VOD playlist
# %Action.WritePlaylist{type: :media, ...},
# %Action.DeletePlaylist{type: :pending, ...},
# ]
# Sliding window mode
{state, actions} = Packager.flush(sliding_state)
# actions = [
# %Action.DeleteSegment{...},
# %Action.DeleteSegment{...},
# %Action.DeletePlaylist{type: :media, ...},
# %Action.DeletePlaylist{type: :master, ...},
# ]
"""
@spec flush(t()) :: {t(), [Action.t()]}
def flush(state) do
if state.max_segments do
flush_with_cleanup(state)
else
flush_as_vod(state)
end
end
defp sync_track(state, track, track_id, sync_point) do
current_count = length(track.media_playlist.segments)
{moved_segments, remaining_segments, moved_duration} =
split_segments_at_sync_point(track.pending_playlist.segments, sync_point, current_count)
if Enum.empty?(moved_segments) do
# Nothing to sync
{{track_id, track}, []}
else
new_duration = track.duration + moved_duration
# Update media playlist
updated_media = %{
track.media_playlist
| segments: track.media_playlist.segments ++ moved_segments
}
# Apply sliding window
{windowed_media, removed_segments, updated_track} =
apply_sliding_window(
updated_media,
state.max_segments,
track,
state.pending_discontinuities
)
# Update pending playlist
updated_pending = %{track.pending_playlist | segments: remaining_segments}
# Generate actions
media_action = %Action.WritePlaylist{
type: :media,
uri: windowed_media.uri,
content: HLS.Playlist.marshal(windowed_media)
}
pending_action =
if state.max_segments do
[]
else
[
%Action.WritePlaylist{
type: :pending,
uri: updated_pending.uri,
content: HLS.Playlist.marshal(updated_pending)
}
]
end
delete_actions = build_delete_actions(removed_segments, windowed_media, track_id)
updated_track = %{
updated_track
| duration: new_duration,
media_playlist: windowed_media,
pending_playlist: updated_pending
}
{{track_id, updated_track}, [media_action] ++ pending_action ++ delete_actions}
end
end
defp split_segments_at_sync_point(pending_segments, sync_point, current_count) do
to_move = max(0, sync_point - current_count)
{moved, remaining} = Enum.split(pending_segments, to_move)
duration = Enum.reduce(moved, 0.0, &(&1.duration + &2))
{moved, remaining, duration}
end
defp apply_sliding_window(playlist, nil, track, _pending_discontinuities),
do: {playlist, [], track}
defp apply_sliding_window(playlist, max_segments, track, pending_discontinuities) do
segment_count = length(playlist.segments)
if segment_count > max_segments do
to_remove = segment_count - max_segments
{removed, remaining} = Enum.split(playlist.segments, to_remove)
# RFC 8216 §6.2.2: "The server SHOULD NOT remove a Media Segment …
# if that would produce a Playlist whose duration is less than three
# times the target duration."
min_duration = 3 * playlist.target_segment_duration
{removed, remaining} =
give_back_for_min_duration(removed, remaining, min_duration)
actually_removed = length(removed)
removed_discontinuities = Enum.count(removed, & &1.discontinuity)
updated = %{
playlist
| segments: remaining,
media_sequence_number: playlist.media_sequence_number + actually_removed,
discontinuity_sequence: playlist.discontinuity_sequence + removed_discontinuities
}
{updated_track, extra_discontinuities} =
advance_discontinuities_past_window(
track,
pending_discontinuities,
updated.media_sequence_number
)
updated = %{
updated
| discontinuity_sequence: updated.discontinuity_sequence + extra_discontinuities
}
{updated, removed, updated_track}
else
{playlist, [], track}
end
end
# Give segments back from `removed` to `remaining` (one at a time, from the
# tail of removed to the head of remaining) until remaining's total duration
# ≥ min_duration, or there are no more segments to give back.
defp give_back_for_min_duration([], remaining, _min_duration), do: {[], remaining}
defp give_back_for_min_duration(removed, remaining, min_duration) do
remaining_duration = Enum.reduce(remaining, 0.0, &(&1.duration + &2))
if remaining_duration >= min_duration do
{removed, remaining}
else
# Move the last removed segment (closest to the window) back
{still_removed, [given_back]} = Enum.split(removed, -1)
give_back_for_min_duration(
still_removed,
[given_back | remaining],
min_duration
)
end
end
defp build_delete_actions(removed_segments, media_playlist, track_id) do
if Enum.empty?(removed_segments) do
[]
else
# Get init sections still in use
preserved_init_uris =
media_playlist.segments
|> Enum.filter(& &1.init_section)
|> Enum.map(& &1.init_section.uri)
|> MapSet.new()
# Collect init sections from removed segments
removed_init_uris =
removed_segments
|> Enum.filter(& &1.init_section)
|> Enum.map(& &1.init_section.uri)
|> MapSet.new()
# Delete orphaned init sections
orphaned_init_uris = MapSet.difference(removed_init_uris, preserved_init_uris)
segment_deletes =
Enum.map(removed_segments, fn seg ->
%Action.DeleteSegment{uri: seg.uri, track_id: track_id}
end)
init_deletes =
Enum.map(orphaned_init_uris, fn uri ->
%Action.DeleteInitSection{uri: uri, track_id: track_id}
end)
segment_deletes ++ init_deletes
end
end
defp maybe_write_master(state, _sync_point) when state.master_written? do
{state, []}
end
defp maybe_write_master(state, sync_point) do
if sync_point >= 3 do
master = build_master(state)
action = %Action.WritePlaylist{
type: :master,
uri: state.manifest_uri,
content: HLS.Playlist.marshal(master)
}
{%{state | master_written?: true}, [action]}
else
{state, []}
end
end
defp build_master(state) do
alternative_tracks =
state.tracks
|> Map.values()
|> Enum.filter(&is_struct(&1.stream, AlternativeRendition))
variant_tracks =
state.tracks
|> Map.values()
|> Enum.filter(&is_struct(&1.stream, VariantStream))
|> Enum.map(fn track ->
group_ids = VariantStream.associated_group_ids(track.stream)
alternative_codecs =
alternative_tracks
|> Enum.filter(&Enum.member?(group_ids, &1.stream.group_id))
|> Enum.flat_map(& &1.codecs)
all_codecs =
(track.stream.codecs ++ track.codecs ++ alternative_codecs)
|> Enum.uniq()
%{track | stream: %{track.stream | codecs: all_codecs}}
end)
# RFC 8216 §4.3.1.2: Use the highest version required by any playlist.
max_version =
state.tracks
|> Map.values()
|> Enum.map(& &1.media_playlist.version)
|> Enum.max(fn -> 7 end)
%Master{
version: max_version,
uri: state.manifest_uri,
independent_segments: false,
streams: Enum.map(variant_tracks, & &1.stream),
alternative_renditions: Enum.map(alternative_tracks, & &1.stream)
}
end
defp flush_as_vod(state) do
# Move all pending segments to media playlists
{new_tracks, actions} =
state.tracks
|> Enum.map(fn {id, track} ->
# All pending segments become part of media playlist
all_segments =
track.pending_playlist.segments ++ Enum.map(track.pending_segments, & &1.segment)
pending_duration = Enum.reduce(all_segments, 0.0, &(&1.duration + &2))
vod_media = %{
track.media_playlist
| segments: track.media_playlist.segments ++ all_segments,
finished: true,
type: :vod
}
updated_track = %{
track
| duration: track.duration + pending_duration,
media_playlist: vod_media,
pending_segments: []
}
media_action = %Action.WritePlaylist{
type: :media,
uri: vod_media.uri,
content: HLS.Playlist.marshal(vod_media)
}
delete_pending_action = %Action.DeletePlaylist{
type: :pending,
uri: track.pending_playlist.uri
}
{{id, updated_track}, [media_action, delete_pending_action]}
end)
|> Enum.unzip()
master = build_master(%{state | tracks: Map.new(new_tracks)})
master_action = %Action.WritePlaylist{
type: :master,
uri: state.manifest_uri,
content: HLS.Playlist.marshal(master)
}
new_state = %{state | tracks: Map.new(new_tracks), master_written?: true}
{new_state, List.flatten(actions) ++ [master_action]}
end
defp flush_with_cleanup(state) do
# Collect all segments
delete_actions =
state.tracks
|> Enum.flat_map(fn {id, track} ->
all_segments =
track.media_playlist.segments ++
track.pending_playlist.segments ++
Enum.map(track.pending_segments, & &1.segment)
segment_deletes =
Enum.map(all_segments, fn seg ->
%Action.DeleteSegment{uri: seg.uri, track_id: id}
end)
# Delete all init sections
init_uris =
all_segments
|> Enum.filter(& &1.init_section)
|> Enum.map(& &1.init_section.uri)
|> Enum.uniq()
init_deletes =
Enum.map(init_uris, fn uri ->
%Action.DeleteInitSection{uri: uri, track_id: id}
end)
media_delete = %Action.DeletePlaylist{
type: :media,
uri: track.media_playlist.uri
}
pending_delete = %Action.DeletePlaylist{
type: :pending,
uri: track.pending_playlist.uri
}
segment_deletes ++ init_deletes ++ [media_delete, pending_delete]
end)
master_delete = %Action.DeletePlaylist{
type: :master,
uri: state.manifest_uri
}
# Reset state
new_state = %{
state
| tracks: %{},
master_written?: false
}
{new_state, delete_actions ++ [master_delete]}
end
# URI helpers (copied from original)
defp relative_segment_uri(playlist_uri, extname, segment_index) do
root_path =
playlist_uri
|> to_string()
|> Path.basename()
|> String.trim_trailing(".m3u8")
{dir, suffix} =
segment_index
|> to_string()
|> String.pad_leading(10, "0")
|> String.split_at(5)
[root_path, dir, "#{Path.basename(root_path)}_#{suffix}#{extname}"]
|> Path.join()
|> URI.new!()
end
defp append_to_path(uri, append) do
field = if is_nil(uri.path), do: :host, else: :path
Map.update!(uri, field, fn path ->
extname = Path.extname(path)
without_ext = String.trim_trailing(path, extname)
"#{without_ext}#{append}#{extname}"
end)
end
defp build_track_variant_uri(manifest_uri, track_id) do
manifest_uri
|> append_to_path("_#{track_id}")
|> to_string()
|> Path.basename()
|> URI.new!()
end
defp sync_if_ready(state, sync_point) do
mandatory_tracks =
state.tracks
|> Enum.filter(fn {_id, track} -> track.mandatory? end)
warnings =
mandatory_tracks
|> Enum.flat_map(fn {id, track} ->
available =
length(track.pending_playlist.segments) + length(track.media_playlist.segments)
if available < sync_point do
[
%Error{
code: :mandatory_track_missing_segment_at_sync,
message:
"Mandatory track '#{id}' only has #{available} segments but sync point is #{sync_point}",
details: %{
track_id: id,
available_segments: available,
sync_point: sync_point,
missing_segments: sync_point - available
}
}
]
else
[]
end
end)
if warnings != [] do
{:warning, warnings}
else
case check_track_alignment(state.tracks, mandatory_tracks, sync_point, state) do
:ok -> :ready
{:error, error} -> {:error, error}
end
end
end
defp check_track_alignment(tracks, mandatory_tracks, sync_point, state) do
mandatory_ids = MapSet.new(Enum.map(mandatory_tracks, &elem(&1, 0)))
{timestamps, missing} =
Enum.reduce(tracks, {[], []}, fn {id, track}, {acc, missing_acc} ->
segment = segment_at_sync_point(track, sync_point)
cond do
is_nil(segment) ->
{acc, missing_acc}
segment_timestamp_ns(segment) == nil ->
{acc, [id | missing_acc]}
true ->
ts = segment_timestamp_ns(segment)
{[{id, ts} | acc], missing_acc}
end
end)
case timestamps do
[] ->
{:error,
%Error{
code: :track_timing_mismatch_at_sync,
message: "No mandatory tracks have timing information at sync point #{sync_point}",
details: %{sync_point: sync_point, missing_tracks: missing}
}}
[{reference_id, reference_ts} | rest] ->
mismatches =
Enum.reduce(rest, [], fn {id, ts}, acc ->
if abs(ts - reference_ts) > state.timing_tolerance_ns do
[%{track_id: id, timestamp_ns: ts} | acc]
else
acc
end
end)
missing_mandatory =
missing
|> Enum.filter(&MapSet.member?(mandatory_ids, &1))
if mismatches == [] and missing == [] do
:ok
else
{:error,
%Error{
code: :track_timing_mismatch_at_sync,
message: "Tracks are misaligned at sync point #{sync_point}",
details: %{
sync_point: sync_point,
reference_track_id: reference_id,
reference_timestamp_ns: reference_ts,
mismatched_tracks: mismatches,
missing_tracks: missing,
missing_mandatory_tracks: missing_mandatory,
tolerance_ns: state.timing_tolerance_ns
}
}}
end
end
end
defp segment_at_sync_point(track, sync_point) do
segments = track.media_playlist.segments ++ track.pending_playlist.segments
Enum.at(segments, sync_point - 1)
end
defp skip_sync_point_for_track(state, track_id, sync_point) do
case Map.get(state.skipped_sync_points, sync_point) do
nil ->
:ok
track_ids ->
if MapSet.member?(track_ids, track_id) do
remaining = MapSet.delete(track_ids, track_id)
updated =
if MapSet.size(remaining) == 0 do
Map.delete(state.skipped_sync_points, sync_point)
else
Map.put(state.skipped_sync_points, sync_point, remaining)
end
{:skip, %{state | skipped_sync_points: updated}}
else
:ok
end
end
end
defp apply_discontinuity_if_needed(state, track_id, segment_index, timestamp_ns, track) do
case find_discontinuity(state.pending_discontinuities, segment_index) do
nil ->
{segment_pdt, base_pdt, base_timestamp_ns} =
compute_segment_pdt(track.base_pdt, track.base_timestamp_ns, timestamp_ns)
{false, state, segment_pdt, base_pdt, base_timestamp_ns}
discontinuity ->
segment_pdt = discontinuity.pdt_reference
base_pdt = discontinuity.pdt_reference
base_timestamp_ns = timestamp_ns
updated_state =
update_in(state.tracks[track_id].applied_discontinuities, fn applied ->
MapSet.put(applied, discontinuity.id)
end)
|> maybe_pop_discontinuities()
{true, updated_state, segment_pdt, base_pdt, base_timestamp_ns}
end
end
defp schedule_discontinuity(
state,
sync_point,
reason,
pdt_reference,
timestamp_reference_ns \\ nil
) do
if discontinuity_exists?(state.pending_discontinuities, sync_point) do
{:ok, state}
else
passed_tracks =
state.tracks
|> Enum.filter(fn {_id, track} -> length(track.media_playlist.segments) >= sync_point end)
|> Enum.map(&elem(&1, 0))
if passed_tracks != [] do
error = %Error{
code: :discontinuity_point_missed,
message:
"Discontinuity at segment #{sync_point} cannot be synchronized across all tracks",
details: %{
sync_point: sync_point,
reason: reason,
tracks_ahead: passed_tracks
}
}
{:error, error, state}
else
entry = %{
id: make_ref(),
sync_point: sync_point,
pdt_reference: pdt_reference,
timestamp_reference_ns: timestamp_reference_ns,
reason: reason
}
{:ok, %{state | pending_discontinuities: state.pending_discontinuities ++ [entry]}}
end
end
end
defp discontinuity_exists?(discontinuities, sync_point) do
Enum.any?(discontinuities, fn disc -> disc.sync_point == sync_point end)
end
defp find_discontinuity(discontinuities, sync_point) do
Enum.find(discontinuities, fn disc -> disc.sync_point == sync_point end)
end
defp maybe_pop_discontinuities(state) do
case state.pending_discontinuities do
[] ->
state
[head | rest] ->
all_applied? =
Enum.all?(state.tracks, fn {_id, track} ->
MapSet.member?(track.applied_discontinuities, head.id)
end)
if all_applied? do
maybe_pop_discontinuities(%{state | pending_discontinuities: rest})
else
state
end
end
end
defp advance_discontinuities_past_window(track, pending_discontinuities, media_sequence_number) do
{applied_ids, count, updated_track} =
pending_discontinuities
|> Enum.reduce({track.applied_discontinuities, 0, track}, fn disc,
{applied, acc, acc_track} ->
if MapSet.member?(applied, disc.id) do
{applied, acc, acc_track}
else
if disc.sync_point <= media_sequence_number do
updated_track =
acc_track
|> Map.put(:base_pdt, disc.pdt_reference)
|> Map.put(:base_timestamp_ns, disc.timestamp_reference_ns)
{MapSet.put(applied, disc.id), acc + 1, updated_track}
else
{applied, acc, acc_track}
end
end
end)
{%{updated_track | applied_discontinuities: applied_ids}, count}
end
defp next_discontinuity_sync_point(state) do
state.tracks
|> Enum.map(fn {_id, track} -> track.segment_count end)
|> Enum.max(fn -> 0 end)
|> Kernel.+(1)
end
defp segment_timestamp_ns(%Segment{dts: dts, pts: pts}) do
dts || pts
end
defp compute_segment_pdt(base_pdt, base_timestamp_ns, timestamp_ns) when is_nil(base_pdt) do
{nil, nil, base_timestamp_ns || timestamp_ns}
end
defp compute_segment_pdt(base_pdt, base_timestamp_ns, timestamp_ns)
when is_nil(base_timestamp_ns) do
{base_pdt, base_pdt, timestamp_ns}
end
defp compute_segment_pdt(base_pdt, base_timestamp_ns, timestamp_ns) do
offset_ns = timestamp_ns - base_timestamp_ns
{DateTime.add(base_pdt, offset_ns, :nanosecond), base_pdt, base_timestamp_ns}
end
defp duration_to_ns(duration) do
round(duration * 1_000_000_000)
end
defp timing_tolerance_to_ns(tolerance_ms) when is_integer(tolerance_ms) do
tolerance_ms * 1_000_000
end
defp resolve_mandatory(nil, stream), do: is_struct(stream, VariantStream)
defp resolve_mandatory(mandatory?, _stream), do: mandatory?
defp variant_track?(_track_id, track) do
is_struct(track.stream, VariantStream)
end
# Validation helpers
defp validate_new_opts(opts) do
with {:ok, manifest_uri} <- Keyword.fetch(opts, :manifest_uri) do
{:ok,
%{
manifest_uri: manifest_uri,
max_segments: Keyword.get(opts, :max_segments),
timing_tolerance_ns:
opts
|> Keyword.get(:timing_tolerance_ms, 200)
|> timing_tolerance_to_ns()
}}
else
:error -> {:error, "manifest_uri is required"}
end
end
defp validate_resume_opts(opts) do
with {:ok, master} <- Keyword.fetch(opts, :master_playlist),
{:ok, medias} <- Keyword.fetch(opts, :media_playlists) do
{:ok,
%{
master_playlist: master,
media_playlists: medias,
max_segments: Keyword.get(opts, :max_segments),
timing_tolerance_ns:
opts
|> Keyword.get(:timing_tolerance_ms, 200)
|> timing_tolerance_to_ns()
}}
else
:error -> {:error, "master_playlist and media_playlists required"}
end
end
defp validate_add_track_opts!(opts) do
%{
stream: Keyword.fetch!(opts, :stream),
segment_extension: Keyword.fetch!(opts, :segment_extension),
target_segment_duration: Keyword.fetch!(opts, :target_segment_duration),
codecs: Keyword.get(opts, :codecs, []),
mandatory?: Keyword.get(opts, :mandatory?, nil)
}
end
defp build_tracks_from_playlists(%{
master_playlist: master,
media_playlists: medias,
max_segments: max_segments,
timeline_reference: timeline_reference
}) do
media_by_uri =
Enum.reduce(medias, %{}, fn %Media{uri: uri} = media, acc ->
Map.put(acc, to_string(uri), media)
end)
streams = master.streams ++ master.alternative_renditions
with {:ok, entries, referenced_uris} <-
build_resume_entries(streams, media_by_uri, master.uri),
:ok <- validate_no_extra_playlists(media_by_uri, referenced_uris),
{:ok, common_sync_point} <- compute_common_sync_point(entries),
{:ok, trimmed_entries} <- trim_entries_to_sync_point(entries, common_sync_point),
{:ok, tracks} <-
build_tracks_map(trimmed_entries, max_segments, timeline_reference) do
{:ok, {tracks, common_sync_point}}
end
end
defp build_resume_entries(streams, media_by_uri, manifest_uri) do
Enum.reduce_while(
streams,
{:ok, [], MapSet.new(), MapSet.new()},
fn stream, {:ok, acc, referenced_uris, track_ids} ->
case Map.fetch(stream, :uri) do
:error ->
error = %Error{
code: :resume_missing_playlist,
message: "Stream is missing a media playlist URI",
details: %{stream: stream}
}
{:halt, {:error, error}}
{:ok, nil} ->
error = %Error{
code: :resume_missing_playlist,
message: "Stream is missing a media playlist URI",
details: %{stream: stream}
}
{:halt, {:error, error}}
{:ok, uri} ->
track_id = derive_track_id_from_uri(manifest_uri, uri)
if MapSet.member?(track_ids, track_id) do
error = %Error{
code: :resume_inconsistent_playlist,
message: "Duplicate track id derived from media playlist URI",
details: %{track_id: track_id, uri: uri}
}
{:halt, {:error, error}}
else
{media, missing?} =
case Map.get(media_by_uri, to_string(uri)) do
nil ->
{%Media{uri: uri, media_sequence_number: 0, segments: []}, true}
%Media{} = media ->
{media, false}
end
entry = %{track_id: track_id, stream: stream, media: media, missing_media?: missing?}
{:cont,
{:ok, [entry | acc], MapSet.put(referenced_uris, to_string(uri)),
MapSet.put(track_ids, track_id)}}
end
end
end
)
|> case do
{:ok, entries, referenced_uris, _track_ids} ->
{:ok, Enum.reverse(entries), referenced_uris}
{:error, _} = error ->
error
end
end
defp validate_no_extra_playlists(media_by_uri, referenced_uris) do
extra =
media_by_uri
|> Map.keys()
|> Enum.reject(&MapSet.member?(referenced_uris, &1))
if extra == [] do
:ok
else
{:error,
%Error{
code: :resume_unexpected_playlist,
message: "Media playlists not referenced by the master playlist",
details: %{uris: extra}
}}
end
end
defp compute_common_sync_point([]) do
{:error,
%Error{
code: :resume_inconsistent_playlist,
message: "No tracks available to resume",
details: %{}
}}
end
defp compute_common_sync_point(entries) do
counts =
Enum.map(entries, fn entry ->
entry.media.media_sequence_number + length(entry.media.segments)
end)
{:ok, Enum.min(counts)}
end
defp trim_entries_to_sync_point(entries, common_sync_point) do
Enum.reduce_while(entries, {:ok, []}, fn entry, {:ok, acc} ->
media = entry.media
media_sequence = media.media_sequence_number
segment_count = media_sequence + length(media.segments)
cond do
media_sequence > common_sync_point ->
error = %Error{
code: :resume_inconsistent_playlist,
message: "Media playlist starts after the common sync point",
details: %{track_id: entry.track_id, sync_point: common_sync_point}
}
{:halt, {:error, error}}
segment_count <= common_sync_point ->
{:cont, {:ok, [entry | acc]}}
true ->
excess = segment_count - common_sync_point
trimmed_segments = Enum.drop(media.segments, -excess)
trimmed_media = %{media | segments: trimmed_segments}
trimmed_entry = %{entry | media: trimmed_media}
{:cont, {:ok, [trimmed_entry | acc]}}
end
end)
|> case do
{:ok, trimmed} -> {:ok, Enum.reverse(trimmed)}
{:error, _} = error -> error
end
end
defp build_tracks_map(entries, max_segments, timeline_reference) do
tracks =
Enum.reduce(entries, %{}, fn entry, acc ->
{:ok, track} = build_track_from_entry(entry, max_segments, timeline_reference)
Map.put(acc, entry.track_id, track)
end)
{:ok, tracks}
end
defp build_track_from_entry(
%{stream: stream, media: media, missing_media?: missing_media?},
max_segments,
timeline_reference
) do
segment_extension =
case infer_segment_extension(media.segments) do
{:ok, ext} -> ext
{:error, _error} -> nil
end
init_section = last_init_section(media.segments)
type = if max_segments, do: nil, else: media.type || :event
duration = Media.compute_playlist_duration(media)
segment_count = media.media_sequence_number + length(media.segments)
resume_incomplete? =
missing_media? or is_nil(segment_extension) or is_nil(media.target_segment_duration)
{:ok,
%Track{
stream: stream,
duration: duration,
segment_count: segment_count,
segment_extension: segment_extension,
init_section: init_section,
media_playlist: %{media | type: type},
pending_playlist: %{media | uri: append_to_path(media.uri, "_pending"), segments: []},
codecs: [],
mandatory?: resolve_mandatory(nil, stream),
last_timestamp_ns: nil,
last_duration_ns: nil,
base_pdt: timeline_reference,
base_timestamp_ns: nil,
pending_segments: [],
applied_discontinuities: MapSet.new(),
resume_incomplete?: resume_incomplete?
}}
end
defp infer_segment_extension([]) do
{:error,
%Error{
code: :resume_unknown_segment_extension,
message: "Unable to infer segment extension from an empty media playlist",
details: %{}
}}
end
defp infer_segment_extension(segments) do
ext =
segments
|> List.last()
|> Map.get(:uri)
|> to_string()
|> Path.extname()
if ext == "" do
{:error,
%Error{
code: :resume_unknown_segment_extension,
message: "Unable to infer segment extension from media playlist",
details: %{}
}}
else
{:ok, ext}
end
end
defp last_init_section(segments) do
Enum.reduce(segments, nil, fn seg, acc ->
if seg.init_section, do: seg.init_section, else: acc
end)
end
defp derive_track_id_from_uri(manifest_uri, media_uri) do
manifest_base =
manifest_uri
|> to_string()
|> Path.basename()
|> Path.rootname()
media_base =
media_uri
|> to_string()
|> Path.basename()
|> Path.rootname()
prefix = manifest_base <> "_"
if String.starts_with?(media_base, prefix) do
String.replace_prefix(media_base, prefix, "")
else
media_base
end
end
defp build_track_stream(state, track_id, stream) do
case stream do
%AlternativeRendition{type: :closed_captions} = alt ->
{%{alt | uri: nil}, build_track_variant_uri(state.manifest_uri, track_id)}
stream ->
stream_uri = build_track_variant_uri(state.manifest_uri, track_id)
{%{stream | uri: stream_uri}, stream_uri}
end
end
defp resume_track_ready?(track, track_id) do
missing =
[]
|> maybe_add_missing(:segment_extension, is_nil(track.segment_extension))
|> maybe_add_missing(
:target_segment_duration,
is_nil(track.media_playlist.target_segment_duration)
)
if missing == [] do
:ok
else
{:error,
%Error{
code: :resume_track_not_ready,
message: "Track '#{track_id}' is not ready after resume",
details: %{track_id: track_id, missing: missing}
}}
end
end
defp maybe_add_missing(list, _field, false), do: list
defp maybe_add_missing(list, field, true), do: [field | list]
defp reconcile_incomplete_track(track, track_id, stream, media_playlist_uri, opts, mandatory?) do
cond do
track.stream != stream ->
{:error,
%Error{
code: :track_conflict,
message: "Track #{track_id} already exists with a different stream",
details: %{track_id: track_id}
}}
track.media_playlist.uri != media_playlist_uri ->
{:error,
%Error{
code: :track_conflict,
message: "Track #{track_id} already exists with a different playlist URI",
details: %{track_id: track_id}
}}
not is_nil(track.media_playlist.target_segment_duration) and
track.media_playlist.target_segment_duration != opts.target_segment_duration ->
{:error,
%Error{
code: :track_conflict,
message: "Track #{track_id} already exists with a different target duration",
details: %{track_id: track_id}
}}
not is_nil(track.segment_extension) and track.segment_extension != opts.segment_extension ->
{:error,
%Error{
code: :track_conflict,
message: "Track #{track_id} already exists with a different extension",
details: %{track_id: track_id}
}}
track.codecs != [] and track.codecs != opts.codecs ->
{:error,
%Error{
code: :track_conflict,
message: "Track #{track_id} already exists with different codecs",
details: %{track_id: track_id}
}}
track.mandatory? != mandatory? ->
{:error,
%Error{
code: :track_conflict,
message: "Track #{track_id} already exists with a different mandatory flag",
details: %{track_id: track_id}
}}
true ->
target_duration =
track.media_playlist.target_segment_duration || opts.target_segment_duration
updated_media = %{track.media_playlist | target_segment_duration: target_duration}
updated_pending = %{track.pending_playlist | target_segment_duration: target_duration}
updated_track = %{
track
| segment_extension: track.segment_extension || opts.segment_extension,
media_playlist: updated_media,
pending_playlist: updated_pending,
codecs: if(track.codecs == [], do: opts.codecs, else: track.codecs),
mandatory?: mandatory?,
resume_incomplete?: false
}
{:ok, updated_track}
end
end
defp track_spec_matches?(track, stream, media_playlist_uri, opts, mandatory?) do
codecs_match =
cond do
track.codecs == opts.codecs ->
true
track.codecs == [] and is_struct(stream, VariantStream) ->
Enum.all?(opts.codecs, &(&1 in stream.codecs))
track.codecs == [] and is_struct(stream, AlternativeRendition) ->
true
true ->
false
end
track.stream == stream and
track.segment_extension == opts.segment_extension and
track.media_playlist.target_segment_duration == opts.target_segment_duration and
track.media_playlist.uri == media_playlist_uri and
codecs_match and
track.mandatory? == mandatory?
end
end