v0

Author: sgfn

lib/ex_webrtc/recorder.ex

@@ -8,33 +8,17 @@ defmodule ExWebRTC.Recorder do
   See `ExWebRTC.Recorder.S3` and `t:options/0` for more info.
   """
 
-  use GenServer
-
   alias ExWebRTC.MediaStreamTrack
-  alias __MODULE__.S3.UploadHandler
   alias __MODULE__.S3
 
   require Logger
 
+  use GenServer
+
   @default_base_dir "./recordings"
 
   @type recorder :: GenServer.server()
 
-  @typep track_manifest :: %{
-           start_time: DateTime.t(),
-           kind: :video | :audio,
-           streams: [MediaStreamTrack.stream_id()],
-           rid_map: %{MediaStreamTrack.rid() => integer()},
-           location: String.t()
-         }
-
-  # XXX really opaque?
-  @typedoc """
-  Contains metadata about the recordings.
-  WRITEME
-  """
-  @opaque manifest :: %{MediaStreamTrack.id() => track_manifest()}
-
   @typedoc """
   Options that can be passed to `start_link/1`.
 
@@ -57,12 +41,12 @@ defmodule ExWebRTC.Recorder do
   Messages sent by the `ExWebRTC.Recorder` process to its controlling process.
 
   * `:upload_complete`, `:upload_failed` - Sent after the completion of the upload task, identified by its reference.
-    Contains the updated manifest with `s3://` schema URLs to uploaded files.
+    Contains the updated manifest with `s3://` scheme URLs to uploaded files.
   """
   @type message ::
           {:ex_webrtc_recorder, pid(),
-           {:upload_complete, S3.upload_task_ref(), manifest()}
-           | {:upload_failed, S3.upload_task_ref(), manifest()}}
+           {:upload_complete, S3.upload_task_ref(), __MODULE__.Manifest.t()}
+           | {:upload_failed, S3.upload_task_ref(), __MODULE__.Manifest.t()}}
 
   # Necessary to start Recorder under a supervisor using `{Recorder, [recorder_opts, gen_server_opts]}`
   @doc false
@@ -107,9 +91,9 @@ defmodule ExWebRTC.Recorder do
   Adds new tracks to the recording.
 
   Returns the part of the recording manifest that's relevant to the freshly added tracks.
-  See `t:manifest/0` for more info.
+  See `t:ExWebRTC.Recorder.Manifest.t/0` for more info.
   """
-  @spec add_tracks(recorder(), [MediaStreamTrack.t()]) :: {:ok, manifest()}
+  @spec add_tracks(recorder(), [MediaStreamTrack.t()]) :: {:ok, __MODULE__.Manifest.t()}
   def add_tracks(recorder, tracks) do
     GenServer.call(recorder, {:add_tracks, tracks})
   end
@@ -143,18 +127,18 @@ defmodule ExWebRTC.Recorder do
   Finishes the recording for the given tracks and optionally uploads the result files.
 
   Returns the part of the recording manifest that's relevant to the freshly ended tracks.
-  See `t:manifest/0` for more info.
+  See `t:ExWebRTC.Recorder.Manifest.t/0` for more info.
 
   If uploads are configured:
   * Returns the reference to the upload task that was spawned.
   * Will send the `:upload_complete`/`:upload_failed` message with this reference
     to the controlling process when the task finishes.
 
   Note that the manifest returned by this function always contains local paths to files.
-  The updated manifest with `s3://` schema URLs is sent in the aforementioned message.
+  The updated manifest with `s3://` scheme URLs is sent in the aforementioned message.
   """
   @spec end_tracks(recorder(), [MediaStreamTrack.id()]) ::
-          {:ok, manifest(), S3.upload_task_ref() | nil} | {:error, :tracks_not_found}
+          {:ok, __MODULE__.Manifest.t(), S3.upload_task_ref() | nil} | {:error, :tracks_not_found}
   def end_tracks(recorder, track_ids) do
     GenServer.call(recorder, {:end_tracks, track_ids})
   end
@@ -172,7 +156,7 @@ defmodule ExWebRTC.Recorder do
     upload_handler =
       if config[:s3_upload_config] do
         Logger.info("Recordings will be uploaded to S3")
-        UploadHandler.new(config[:s3_upload_config])
+        S3.UploadHandler.new(config[:s3_upload_config])
       end
 
     state = %{
@@ -264,7 +248,7 @@ defmodule ExWebRTC.Recorder do
   def handle_info({ref, _res} = task_result, state) when is_reference(ref) do
     if state.upload_handler do
       {result, manifest, handler} =
-        UploadHandler.process_result(state.upload_handler, task_result)
+        S3.UploadHandler.process_result(state.upload_handler, task_result)
 
       case result do
         :ok ->
@@ -332,7 +316,7 @@ defmodule ExWebRTC.Recorder do
         {manifest_diff, nil, state}
 
       handler ->
-        {ref, handler} = UploadHandler.spawn_task(handler, manifest_diff)
+        {ref, handler} = S3.UploadHandler.spawn_task(handler, manifest_diff)
 
         {manifest_diff, ref, %{state | upload_handler: handler}}
     end

lib/ex_webrtc/recorder/converter.ex

@@ -1,6 +1,7 @@
 defmodule ExWebRTC.Recorder.Converter do
   @moduledoc """
   Processes RTP packet files saved by `ExWebRTC.Recorder`.
+
   Requires the `ffmpeg` binary with the relevant libraries present in `PATH`.
 
   At the moment, `ExWebRTC.Recorder.Converter` works only with VP8 video and Opus audio.
@@ -9,14 +10,17 @@ defmodule ExWebRTC.Recorder.Converter do
   See `ExWebRTC.Recorder.S3` and `t:options/0` for more info.
   """
 
-  require Logger
-
   alias ExWebRTC.RTP.JitterBuffer.PacketStore
   alias ExWebRTC.RTP.Depayloader
   alias ExWebRTC.Media.{IVF, Ogg}
+
   alias ExWebRTC.Recorder.S3
   alias ExWebRTC.{Recorder, RTPCodecParameters}
 
+  alias __MODULE__.FFmpeg
+
+  require Logger
+
   # TODO: Allow changing these values
   @ivf_header_opts [
     # <<fourcc::little-32>> = "VP80"
@@ -45,32 +49,23 @@ defmodule ExWebRTC.Recorder.Converter do
   @default_output_path "./converter/output"
   @default_download_path "./converter/download"
   @default_thumbnail_width 640
-  # -1 means fit to aspect ratio
   @default_thumbnail_height -1
 
-  @typep file_manifest :: %{
-           :location => String.t(),
-           :duration_seconds => non_neg_integer(),
-           optional(:thumbnail_location) => String.t()
-         }
-
-  # XXX this probably shouldn't be opaque
-  # WRITEME typedoc
-  @opaque manifest :: %{ExWebRTC.MediaStreamTrack.stream_id() => file_manifest()}
-
   @typedoc """
-  Context for the thumbnail generation
+  Context for the thumbnail generation.
 
-  WRITEME
+  * `:width` - Thumbnail width. #{@default_thumbnail_width} by default.
+  * `:height` - Thumbnail height. #{@default_thumbnail_height} by default.
+
+  Setting either of the values to `-1` will fit the size to the aspect ratio.
   """
   @type thumbnails_ctx :: %{
           optional(:width) => pos_integer() | -1,
           optional(:height) => pos_integer() | -1
         }
 
-  # XXX as well as convert_from_file...?
   @typedoc """
-  Options that can be passed to `convert_manifest!/2`.
+  Options that can be passed to `convert_manifest!/2` and `convert_path!/2`.
 
   * `:output_path` - Directory where Converter will save its artifacts. `#{@default_output_path}` by default.
   * `:s3_upload_config` - If passed, processed recordings will be uploaded to S3-compatible storage.
@@ -91,13 +86,12 @@ defmodule ExWebRTC.Recorder.Converter do
   @type options :: [option()]
 
   @doc """
-  Convert the saved dumps of tracks in the report to IVF and Ogg files.
+  Loads the recording manifest from file, then proceeds with `convert_manifest!/2`.
   """
-  # REWRITEME
-  @spec convert!(Path.t(), Path.t()) :: term() | no_return()
-  def convert!(report_path, output_path \\ @default_output_path) do
-    report_path =
-      report_path
+  @spec convert_path!(Path.t(), options()) :: __MODULE__.Manifest.t() | no_return()
+  def convert_path!(recorder_manifest_path, options \\ []) do
+    recorder_manifest_path =
+      recorder_manifest_path
       |> Path.expand()
       |> then(
         &if(File.dir?(&1),
@@ -106,18 +100,21 @@ defmodule ExWebRTC.Recorder.Converter do
         )
       )
 
-    report =
-      report_path
+    recorder_manifest =
+      recorder_manifest_path
       |> File.read!()
       |> Jason.decode!()
+      |> Recorder.Manifest.from_json!()
 
-    # XXX no maikel this is so not right
-    convert_manifest!(report, output_path)
+    convert_manifest!(recorder_manifest, options)
   end
 
-  # XXX type options + docs
-  @spec convert_manifest!(Recorder.manifest(), keyword()) :: manifest() | no_return()
-  def convert_manifest!(manifest, options \\ [])
+  @doc """
+  Converts the saved dumps of tracks in the manifest to WEBM files.
+  """
+  @spec convert_manifest!(Recorder.Manifest.t(), options()) ::
+          __MODULE__.Manifest.t() | no_return()
+  def convert_manifest!(recorder_manifest, options \\ [])
 
   def convert_manifest!(manifest, options) when map_size(manifest) > 0 do
     thumbnails_ctx =
@@ -154,19 +151,18 @@ defmodule ExWebRTC.Recorder.Converter do
       if upload_handler != nil do
         {ref, upload_handler} =
           output_manifest
-          |> prepare_upload_handler_manifest()
+          |> __MODULE__.Manifest.to_upload_handler_manifest()
           |> then(&S3.UploadHandler.spawn_task(upload_handler, &1))
 
-        # XXX What if upload fails?
+        # FIXME: Add descriptive errors
         {:ok, upload_handler_result_manifest, _handler} =
           receive do
             {^ref, _res} = task_result ->
               S3.UploadHandler.process_result(upload_handler, task_result)
           end
 
-        # XXX this naming of functions and variables is tragic and extremely shitty
         upload_handler_result_manifest
-        |> prepare_result_manifest(output_manifest)
+        |> __MODULE__.Manifest.from_upload_handler_manifest(output_manifest)
       else
         output_manifest
       end
@@ -176,37 +172,6 @@ defmodule ExWebRTC.Recorder.Converter do
 
   def convert_manifest!(_empty_manifest, _options), do: %{}
 
-  # def convert_report!(report, output_path \\ @default_output_path) do
-  #   output_path = Path.expand(output_path)
-  #   File.mkdir_p!(output_path)
-
-  defp prepare_upload_handler_manifest(converter_result_manifest) do
-    Enum.reduce(converter_result_manifest, %{}, fn
-      {id, %{location: file, thumbnail_location: thumbnail}}, acc ->
-        acc
-        |> Map.put(id, %{location: file})
-        |> Map.put("thumbnail_#{id}", %{location: thumbnail})
-
-      {id, %{location: file}}, acc ->
-        Map.put(acc, id, %{location: file})
-    end)
-  end
-
-  defp prepare_result_manifest(upload_handler_result_manifest, original_output_manifest) do
-    Enum.reduce(upload_handler_result_manifest, original_output_manifest, fn
-      {"thumbnail_" <> id, %{location: thumbnail}}, acc ->
-        Map.update(
-          acc,
-          id,
-          %{thumbnail_location: thumbnail},
-          &Map.put(&1, :thumbnail_location, thumbnail)
-        )
-
-      {id, %{location: file}}, acc ->
-        Map.update(acc, id, %{location: file}, &Map.put(&1, :location, file))
-    end)
-  end
-
   defp fetch_remote_files!(manifest, dl_path, dl_config) do
     Map.new(manifest, fn {track_id, %{location: location} = track_data} ->
       scheme = URI.parse(location).scheme || "file"
@@ -229,7 +194,7 @@ defmodule ExWebRTC.Recorder.Converter do
          {:ok, _result} <- S3.Utils.fetch_file(bucket_name, s3_path, out_path, dl_config) do
       {:ok, out_path}
     else
-      # XXX descriptive errors
+      # FIXME: Add descriptive errors
       _other -> :error
     end
   end
@@ -268,49 +233,38 @@ defmodule ExWebRTC.Recorder.Converter do
         |> Map.update!(stream_id, &Map.put(&1, kind, output_metadata))
       end)
 
+    # FIXME: This won't work if we have audio/video only
     for {stream_id, %{video: video_files, audio: audio_files}} <- stream_map,
         {rid, %{filename: video_file, start_time: video_start}} <- video_files,
         {nil, %{filename: audio_file, start_time: audio_start}} <- audio_files,
         into: %{} do
-      {video_start_time, audio_start_time} = calculate_start_times(video_start, audio_start)
       output_id = if rid == nil, do: stream_id, else: "#{stream_id}_#{rid}"
-
       output_file = Path.join(output_path, "#{output_id}.webm")
 
-      {_io, 0} =
-        System.cmd(
-          "ffmpeg",
-          ~w(-ss #{video_start_time} -i #{Path.join(output_path, video_file)} -ss #{audio_start_time} -i #{Path.join(output_path, audio_file)} -c:v copy -c:a copy -shortest #{output_file}),
-          stderr_to_stdout: true
-        )
-
-      {duration, 0} =
-        System.cmd(
-          "ffprobe",
-          ~w(-v error -show_entries format=duration -of default=noprint_wrappers=1:nokey=1 #{output_file})
-        )
+      FFmpeg.combine_av!(
+        Path.join(output_path, video_file),
+        video_start,
+        Path.join(output_path, audio_file),
+        audio_start,
+        output_file
+      )
 
-      {duration_seconds, _rest} = Float.parse(duration)
+      # TODO: Consider deleting the `.ivf` and `.ogg` files at this point
 
       stream_manifest = %{
         location: output_file,
-        duration_seconds: round(duration_seconds)
+        duration_seconds: FFmpeg.get_duration_in_seconds!(output_file)
       }
 
-      if thumbnails_ctx do
-        thumbnail_file = "#{output_file}_thumbnail.jpg"
-
-        {_io, 0} =
-          System.cmd(
-            "ffmpeg",
-            ~w(-i #{output_file} -vf thumbnail,scale=#{thumbnails_ctx.width}:#{thumbnails_ctx.height} -frames:v 1 #{thumbnail_file}),
-            stderr_to_stdout: true
-          )
+      stream_manifest =
+        if thumbnails_ctx do
+          thumbnail_file = FFmpeg.generate_thumbnail!(output_file, thumbnails_ctx)
+          Map.put(stream_manifest, :thumbnail_location, thumbnail_file)
+        else
+          stream_manifest
+        end
 
-        {output_id, Map.put(stream_manifest, :thumbnail_location, thumbnail_file)}
-      else
-        {output_id, stream_manifest}
-      end
+      {output_id, stream_manifest}
     end
   end
 
@@ -331,6 +285,7 @@ defmodule ExWebRTC.Recorder.Converter do
         frames_cnt: 0
       }
 
+      # Returns the timestamp (in milliseconds) at which the first frame was received
       start_time = do_convert_video_track(packets[rid_idx], conversion_state)
 
       {rid, %{filename: filename, start_time: start_time}}
@@ -373,7 +328,7 @@ defmodule ExWebRTC.Recorder.Converter do
 
     {:ok, depayloader} = Depayloader.new(@audio_codec_params)
 
-    # XXX ugleh
+    # Same behaviour as in `convert_video_track/4`
     start_time = do_convert_audio_track(packets, %{depayloader: depayloader, writer: writer})
 
     %{filename: filename, start_time: start_time}
@@ -452,15 +407,4 @@ defmodule ExWebRTC.Recorder.Converter do
         store
     end
   end
-
-  defp calculate_start_times(video_start_ms, audio_start_ms) do
-    diff = abs(video_start_ms - audio_start_ms)
-    s = div(diff, 1000)
-    ms = rem(diff, 1000)
-    delayed_start_time = :io_lib.format("00:00:~2..0w.~3..0w", [s, ms]) |> to_string()
-
-    if video_start_ms > audio_start_ms,
-      do: {"00:00:00.000", delayed_start_time},
-      else: {delayed_start_time, "00:00:00.000"}
-  end
 end