Track recordings #180
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,193 @@ | ||
| defmodule ExWebRTC.Recorder do | ||
| @moduledoc """ | ||
| Saves received RTP packets to a file for later processing/analysis. | ||
|
|
||
| Dumps raw RTP packets fed to it in a custom format. Use `Recorder.Converter` to process them. | ||
| """ | ||
|
|
||
| use GenServer | ||
|
|
||
| alias ExWebRTC.MediaStreamTrack | ||
|
|
||
| require Logger | ||
|
|
||
| @default_base_dir "./recordings" | ||
|
|
||
| @typedoc """ | ||
| Options that can be passed to `start_link/1`. | ||
|
|
||
| * `base_dir` - Base directory where Recorder will save its artifacts. `#{@default_base_dir}` by default. | ||
| * `on_start` - Callback that will be executed just after the Recorder is (re)started. | ||
| It should return the initial list of tracks to be added. | ||
| """ | ||
| @type option :: | ||
| {:base_dir, String.t()} | ||
| | {:on_start, (-> [MediaStreamTrack.t()])} | ||
|
|
||
| @type options :: [option()] | ||
|
|
||
| # Necessary to start Recorder under a supervisor using `{Recorder, [recorder_opts, gen_server_opts]}` | ||
| @doc false | ||
| @spec child_spec(list()) :: Supervisor.child_spec() | ||
| def child_spec(args) do | ||
| %{ | ||
| id: __MODULE__, | ||
| start: {__MODULE__, :start_link, args} | ||
| } | ||
| end | ||
|
|
||
| @doc """ | ||
| Starts a new `ExWebRTC.Recorder` process. | ||
|
|
||
| `ExWebRTC.Recorder` is a `GenServer` under the hood, thus this function allows for | ||
| passing the generic `t:GenServer.options/0` as an argument. | ||
| """ | ||
| @spec start(options(), GenServer.options()) :: GenServer.on_start() | ||
| def start(recorder_opts \\ [], gen_server_opts \\ []) do | ||
| GenServer.start(__MODULE__, recorder_opts, gen_server_opts) | ||
| end | ||
|
|
||
| @doc """ | ||
| Starts a new `ExWebRTC.Recorder` process. | ||
|
|
||
| Works identically to `start/2`, but links to the calling process. | ||
| """ | ||
| @spec start_link(options(), GenServer.options()) :: GenServer.on_start() | ||
| def start_link(recorder_opts \\ [], gen_server_opts \\ []) do | ||
| GenServer.start_link(__MODULE__, recorder_opts, gen_server_opts) | ||
| end | ||
|
|
||
| @doc """ | ||
| Adds new tracks to the recording. | ||
| """ | ||
| @spec add_tracks(GenServer.server(), [MediaStreamTrack.t()]) :: :ok | :error | ||
| def add_tracks(recorder, tracks) do | ||
| # XXX need? | ||
| try do | ||
| GenServer.call(recorder, {:add_tracks, tracks}) | ||
| catch | ||
| _exit_or_error, _e -> | ||
| Logger.error("Recorder is down, not adding tracks") | ||
| :error | ||
| end | ||
| end | ||
|
|
||
| @doc """ | ||
| Records a received packet on the given track. | ||
| """ | ||
| @spec record( | ||
| GenServer.server(), | ||
| MediaStreamTrack.id(), | ||
| MediaStreamTrack.rid() | nil, | ||
| ExRTP.Packet.t() | ||
| ) :: :ok | ||
| def record(recorder, track_id, rid, %ExRTP.Packet{} = packet) do | ||
| recv_time = System.monotonic_time(:millisecond) | ||
| GenServer.cast(recorder, {:record, track_id, rid, recv_time, packet}) | ||
| end | ||
|
|
||
| @impl true | ||
| def init(config) do | ||
| base_dir = | ||
| (config[:base_dir] || @default_base_dir) | ||
| |> Path.join(UUID.uuid4()) | ||
mickel8 marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| |> Path.expand() | ||
|
|
||
| :ok = File.mkdir_p!(base_dir) | ||
| Logger.info("Starting recorder. Recordings will be saved under: #{base_dir}") | ||
|
|
||
| state = %{ | ||
| base_dir: base_dir, | ||
| tracks: %{} | ||
| } | ||
|
|
||
| if config[:on_start] == nil do | ||
| {:ok, state} | ||
| else | ||
| {:ok, state, {:continue, {:on_start, config[:on_start]}}} | ||
| end | ||
| end | ||
|
|
||
| @impl true | ||
| def handle_continue({:on_start, on_start}, state) do | ||
| tracks = on_start.() | ||
|
|
||
| if Enum.empty?(tracks) do | ||
mickel8 marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| {:noreply, state} | ||
| else | ||
| state = do_add_tracks(tracks, state) | ||
| {:noreply, state} | ||
| end | ||
| end | ||
|
|
||
| @impl true | ||
| def handle_call({:add_tracks, tracks}, _from, state) do | ||
| state = do_add_tracks(tracks, state) | ||
| {:reply, :ok, state} | ||
| end | ||
|
|
||
| @impl true | ||
| def handle_cast({:record, track_id, rid, recv_time, packet}, state) | ||
| when is_map_key(state.tracks, track_id) do | ||
| %{file: file, rid_map: rid_map} = state.tracks[track_id] | ||
|
|
||
| case rid_map do | ||
| %{^rid => rid_idx} -> | ||
| :ok = IO.binwrite(file, serialize_packet(packet, rid_idx, recv_time)) | ||
|
|
||
| _other -> | ||
| Logger.warning(""" | ||
| Tried to save packet for unknown rid. Ignoring. Track id: #{inspect(track_id)}, rid: #{inspect(rid)}.\ | ||
| """) | ||
| end | ||
|
|
||
| {:noreply, state} | ||
| end | ||
|
|
||
| @impl true | ||
| def handle_cast({:record, track_id, _rid, _recv_time, _packet}, state) do | ||
| Logger.warning(""" | ||
| Tried to save packet for unknown track id. Ignoring. Track id: #{inspect(track_id)}.\ | ||
| """) | ||
|
|
||
| {:noreply, state} | ||
| end | ||
|
|
||
| @impl true | ||
| def handle_info(_msg, state) do | ||
| {:noreply, state} | ||
| end | ||
|
|
||
| defp do_add_tracks(tracks, state) do | ||
| start_time = DateTime.utc_now() | ||
|
|
||
| tracks = | ||
| Map.new(tracks, fn track -> | ||
| path = Path.join(state.base_dir, "#{track.id}.rtpx") | ||
| file = File.open!(path, [:write]) | ||
| rid_map = (track.rids || [nil]) |> Enum.with_index() |> Map.new() | ||
|
|
||
| {track.id, | ||
| %{kind: track.kind, rid_map: rid_map, path: path, file: file, start_time: start_time}} | ||
| end) | ||
|
|
||
| state = %{state | tracks: Map.merge(state.tracks, tracks)} | ||
| report_path = Path.join(state.base_dir, "report.json") | ||
|
|
||
| report = | ||
| Map.new(state.tracks, fn {id, track} -> | ||
| track = Map.delete(track, :file) | ||
| {id, track} | ||
| end) | ||
|
|
||
| :ok = File.write!(report_path, Jason.encode!(report)) | ||
|
|
||
| %{state | tracks: tracks} | ||
| end | ||
|
|
||
| defp serialize_packet(packet, rid_idx, recv_time) do | ||
| packet = ExRTP.Packet.encode(packet) | ||
| packet_size = byte_size(packet) | ||
| <<rid_idx::8, recv_time::64, packet_size::32, packet::binary>> | ||
| end | ||
| end | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,171 @@ | ||
| defmodule ExWebRTC.Recorder.Converter do | ||
| @moduledoc """ | ||
| Processes RTP packet files saved by `Recorder`. | ||
|
|
||
| At the moment, `Converter` works only with VP8 video and Opus audio. | ||
| """ | ||
|
|
||
| require Logger | ||
|
|
||
| alias ExWebRTC.RTP.JitterBuffer.PacketStore | ||
| alias ExWebRTC.RTPCodecParameters | ||
| alias ExWebRTC.RTP.Depayloader | ||
| alias ExWebRTC.Media.{IVF, Ogg} | ||
|
|
||
| # TODO: Allow changing these values | ||
| @ivf_header_opts [ | ||
| # <<fourcc::little-32>> = "VP80" | ||
| fourcc: 808_996_950, | ||
| height: 720, | ||
| width: 1280, | ||
| num_frames: 1024, | ||
| timebase_denum: 24, | ||
| timebase_num: 1 | ||
| ] | ||
|
|
||
| # TODO: Support codecs other than VP8/Opus | ||
| @video_codec_params %RTPCodecParameters{ | ||
| payload_type: 96, | ||
| mime_type: "video/VP8", | ||
| clock_rate: 90_000 | ||
| } | ||
|
|
||
| @audio_codec_params %RTPCodecParameters{ | ||
| payload_type: 111, | ||
| mime_type: "audio/opus", | ||
| clock_rate: 48_000, | ||
| channels: 2 | ||
| } | ||
|
|
||
| @default_output_path "./converter_output" | ||
|
|
||
| @doc """ | ||
| Convert the saved dumps of tracks in the report to IVF and Ogg files. | ||
| """ | ||
| @spec convert!(Path.t(), Path.t()) :: :ok | no_return() | ||
| def convert!(report_path, output_path \\ @default_output_path) do | ||
| report_path = | ||
| report_path | ||
| |> Path.expand() | ||
| |> then( | ||
| &if(File.dir?(&1), | ||
| do: Path.join(&1, "report.json"), | ||
| else: &1 | ||
| ) | ||
| ) | ||
|
Comment on lines
+47
to
+55
There was a problem hiding this comment. I think we should take a path to the base dir and process all subdirectories. Not necessarily in this PR but in general. Optionally, we can allow to pass an option to process a single dir. There was a problem hiding this comment. I agree, but I'd leave it for when we need to use the Converter somewhere (e.g in our demo) |
||
|
|
||
| output_path = Path.expand(output_path) | ||
| File.mkdir_p!(output_path) | ||
|
|
||
| report = | ||
| report_path | ||
| |> File.read!() | ||
| |> Jason.decode!() | ||
|
|
||
| for {id, track} <- report do | ||
| %{ | ||
| "path" => path, | ||
| "kind" => kind, | ||
| "rid_map" => rid_map | ||
| } = track | ||
|
|
||
| file = File.open!(path) | ||
|
|
||
| packets = | ||
| read_packets(file, Map.new(rid_map, fn {_rid, rid_idx} -> {rid_idx, %PacketStore{}} end)) | ||
|
|
||
| case kind do | ||
| "video" -> | ||
| convert_video_track(id, rid_map, output_path, packets) | ||
|
|
||
| "audio" -> | ||
| {:ok, writer} = | ||
| output_path | ||
| |> Path.join("#{id}.ogg") | ||
| |> Ogg.Writer.open() | ||
|
|
||
| {:ok, depayloader} = Depayloader.new(@audio_codec_params) | ||
mickel8 marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| do_convert_audio_track(packets |> Map.values() |> hd(), depayloader, writer) | ||
| end | ||
| end | ||
|
|
||
| :ok | ||
| end | ||
|
|
||
| defp convert_video_track(id, rid_map, output_path, packets) do | ||
| for {rid, rid_idx} <- rid_map do | ||
| filename = if rid == "nil", do: "#{id}.ivf", else: "#{id}_#{rid}.ivf" | ||
|
|
||
| {:ok, writer} = | ||
| output_path | ||
| |> Path.join(filename) | ||
| |> IVF.Writer.open(@ivf_header_opts) | ||
|
|
||
| {:ok, depayloader} = Depayloader.new(@video_codec_params) | ||
| do_convert_video_track(packets[rid_idx], depayloader, writer) | ||
| end | ||
| end | ||
|
|
||
| defp do_convert_video_track(packets, depayloader, writer, frames_cnt \\ 0) | ||
| defp do_convert_video_track([], _depayloader, writer, _frames_cnt), do: IVF.Writer.close(writer) | ||
|
|
||
| defp do_convert_video_track([packet | rest], depayloader, writer, frames_cnt) do | ||
| case Depayloader.depayload(depayloader, packet) do | ||
| {nil, depayloader} -> | ||
| do_convert_video_track(rest, depayloader, writer, frames_cnt) | ||
|
|
||
| {vp8_frame, depayloader} -> | ||
| frame = %IVF.Frame{timestamp: frames_cnt, data: vp8_frame} | ||
| {:ok, writer} = IVF.Writer.write_frame(writer, frame) | ||
| do_convert_video_track(rest, depayloader, writer, frames_cnt + 1) | ||
| end | ||
| end | ||
|
|
||
| defp do_convert_audio_track([], _depayloader, writer), do: Ogg.Writer.close(writer) | ||
|
|
||
| defp do_convert_audio_track([packet | rest], depayloader, writer) do | ||
| {opus_packet, depayloader} = Depayloader.depayload(depayloader, packet) | ||
| {:ok, writer} = Ogg.Writer.write_packet(writer, opus_packet) | ||
| do_convert_audio_track(rest, depayloader, writer) | ||
| end | ||
|
|
||
| defp read_packets(file, stores) do | ||
| case read_packet(file) do | ||
| {:ok, rid_idx, packet} -> | ||
| stores = Map.update!(stores, rid_idx, &insert_packet_to_store(&1, packet)) | ||
| read_packets(file, stores) | ||
|
|
||
| {:error, :not_enough_data} -> | ||
| Logger.warning("Error decoding RTP packet") | ||
| read_packets(file, stores) | ||
|
|
||
| :eof -> | ||
| Map.new(stores, fn {rid_idx, store} -> | ||
| {rid_idx, store |> PacketStore.dump() |> Enum.reject(&is_nil/1)} | ||
| end) | ||
| end | ||
| end | ||
|
|
||
| defp read_packet(file) do | ||
| case IO.binread(file, 13) do | ||
| <<rid_idx::8, _recv_time::64, packet_size::32>> -> | ||
| with {:ok, packet} <- file |> IO.binread(packet_size) |> ExRTP.Packet.decode() do | ||
mickel8 marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| {:ok, rid_idx, packet} | ||
| end | ||
|
|
||
| :eof -> | ||
| :eof | ||
| end | ||
| end | ||
|
|
||
| defp insert_packet_to_store(store, packet) do | ||
| case PacketStore.insert(store, packet) do | ||
| {:ok, store} -> | ||
| store | ||
|
|
||
| {:error, :late_packet} -> | ||
|
There was a problem hiding this comment. In practice, this should only happen when a recording starts with out-of-order packets, correct? There was a problem hiding this comment. Yes, I believe that's correct |
||
| Logger.warning("Decoded late RTP packet") | ||
| store | ||
| end | ||
| end | ||
| end | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Up for discussion whether this
tryblock should go hereI don't think this is the correct place for it, I'd put it in Broadcaster logic instead
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I had the same problem some time ago - link 🙄
IMO, we can remove
try catchfor now. I don't know how someone could handle the error as it is related to this specific track, not the process of recording as a whole.Recorder is also assumed to be alive all the time so if it is not, then something bad happened and it might be better for the caller to just crash.
And we can always go back to
try catchif we find it useful in the future.