fix thumbnailer and webrtc element (#487)

* keep rtsp connection alive

* upgrade

apps/ex_nvr/lib/ex_nvr/pipeline/output/thumbnailer.ex

@@ -86,7 +86,7 @@ defmodule ExNVR.Pipeline.Output.Thumbnailer do
   def handle_buffer(:input, _buffer, _ctx, state), do: {[], state}
 
   defp do_decode(buffer, state) do
-    with {:ok, decoded} <- state.decoder.decode(state.decoder_state, buffer),
+    with {:ok, decoded} <- decode(state, buffer),
          {:ok, scaled} <- scale(decoded, state),
          {:ok, jpeg_image} <- to_jpeg(scaled, state),
          :ok <- File.write(image_path(state.dest, buffer), jpeg_image) do
@@ -98,14 +98,25 @@ defmodule ExNVR.Pipeline.Output.Thumbnailer do
     end
   end
 
-  defp scale([], _state), do: {:ok, nil}
+  defp decode(state, buffer) do
+    with {:ok, []} <- state.decoder.decode(state.decoder_state, buffer) do
+      state.decoder.flush(state.decoder_state)
+    end
+  end
+
+  defp scale([], _state), do: {:error, nil}
   defp scale([buffer], state), do: Image.Scaler.scale(state.scaler, buffer.payload)
 
   defp to_jpeg(raw_image, state) do
     Turbojpeg.yuv_to_jpeg(raw_image, state.thumbnail_width, state.thumbnail_height, 75, :I420)
   end
 
   defp image_path(dest_folder, buffer) do
-    Path.join(dest_folder, "#{Membrane.Buffer.get_dts_or_pts(buffer)}.jpg")
+    filename =
+      if Application.get_env(:ex_nvr, :env) == :test,
+        do: "#{Membrane.Buffer.get_dts_or_pts(buffer)}.jpg",
+        else: "#{DateTime.utc_now() |> DateTime.to_unix()}.jpg"
+
+    Path.join(dest_folder, filename)
   end
 end

apps/ex_nvr/lib/ex_nvr/pipeline/output/web_rtc.ex

@@ -94,7 +94,10 @@ defmodule ExNVR.Pipeline.Output.WebRTC do
 
   @impl true
   def handle_pad_removed(Pad.ref(:input, :main_stream), _ctx, state) do
-    Engine.message_endpoint(state.rtc_engine, state.stream_endpoint_id, :remove_track)
+    if state.encoding == :H264 do
+      Engine.message_endpoint(state.rtc_engine, state.stream_endpoint_id, :remove_track)
+    end
+
     {[remove_children: :sink], %{state | encoding: nil}}
   end
 
@@ -199,11 +202,11 @@ defmodule ExNVR.Pipeline.Output.WebRTC do
     Membrane.Logger.error("Stream endpoint #{endpoint_id} crashed")
     Engine.add_endpoint(state.rtc_engine, %Output.WebRTC.StreamEndpoint{}, id: endpoint_id)
 
-    if state.media_track do
+    if state.encoding do
       Engine.message_endpoint(
         state.rtc_engine,
         endpoint_id,
-        {:media_track, state.media_track}
+        {:encoding, state.encoding}
       )
     end
 

apps/ex_nvr/mix.exs

@@ -4,7 +4,7 @@ defmodule ExNVR.MixProject do
   def project do
     [
       app: :ex_nvr,
-      version: "0.15.0",
+      version: "0.15.1",
       build_path: "../../_build",
       config_path: "../../config/config.exs",
       deps_path: "../../deps",

apps/ex_nvr_rtsp/lib/rtsp/source.ex

@@ -151,7 +151,8 @@ defmodule ExNVR.RTSP.Source do
 
   @impl true
   def handle_info(:keep_alive, _ctx, state) do
-    Task.start(fn -> ConnectionManager.keep_alive(state) end)
+    pid = self()
+    Task.start(fn -> ConnectionManager.keep_alive(state, pid) end)
     {[], state}
   end
 

apps/ex_nvr_rtsp/lib/rtsp/source/connection_manager.ex

@@ -51,20 +51,20 @@ defmodule ExNVR.RTSP.Source.ConnectionManager do
 
     case RTSP.play(state.rtsp_session) do
       {:ok, %{status: 200}} ->
-        %{state | keep_alive_timer: start_keep_alive_timer(state)}
+        %{state | keep_alive_timer: start_keep_alive_timer(state, self())}
 
       _error ->
         handle_rtsp_error(:play_rtsp_failed, state)
     end
   end
 
-  @spec keep_alive(State.t()) :: State.t()
-  def keep_alive(state) do
+  @spec keep_alive(State.t(), pid()) :: State.t()
+  def keep_alive(state, pid) do
     Membrane.Logger.debug("Send GET_PARAMETER to keep session alive")
 
     {:ok, %{status: 200}} = RTSP.get_parameter(state.rtsp_session)
 
-    %{state | keep_alive_timer: start_keep_alive_timer(state)}
+    %{state | keep_alive_timer: start_keep_alive_timer(state, pid)}
   end
 
   @spec start_rtsp_connection(State.t()) :: connection_establishment_phase_return()
@@ -112,9 +112,9 @@ defmodule ExNVR.RTSP.Source.ConnectionManager do
     end
   end
 
-  @spec start_keep_alive_timer(State.t()) :: reference()
-  defp start_keep_alive_timer(%{keep_alive_interval: interval}) do
-    Process.send_after(self(), :keep_alive, interval |> Membrane.Time.as_milliseconds(:round))
+  @spec start_keep_alive_timer(State.t(), pid()) :: reference()
+  defp start_keep_alive_timer(%{keep_alive_interval: interval}, pid) do
+    Process.send_after(pid, :keep_alive, interval |> Membrane.Time.as_milliseconds(:round))
   end
 
   @spec setup_rtsp_connection_with_tcp(RTSP.t(), [track()]) ::

apps/ex_nvr_rtsp/mix.exs

@@ -4,7 +4,7 @@ defmodule ExNvrRtsp.MixProject do
   def project do
     [
       app: :ex_nvr_rtsp,
-      version: "0.15.0",
+      version: "0.15.1",
       build_path: "../../_build",
       config_path: "../../config/config.exs",
       deps_path: "../../deps",

apps/ex_nvr_web/assets/package.json

@@ -1,6 +1,6 @@
 {
   "name": "ex_nvr",
-  "version": "0.15.0",
+  "version": "0.15.1",
   "description": "",
   "main": "index.js",
   "scripts": {

apps/ex_nvr_web/mix.exs

@@ -4,7 +4,7 @@ defmodule ExNVRWeb.MixProject do
   def project do
     [
       app: :ex_nvr_web,
-      version: "0.15.0",
+      version: "0.15.1",
       build_path: "../../_build",
       config_path: "../../config/config.exs",
       deps_path: "../../deps",

apps/ex_nvr_web/priv/static/openapi.yaml

@@ -2,7 +2,7 @@ openapi: 3.0.0
 info:
   title: ExNVR API
   description: Manage ExNVR via API endpoints
-  version: 0.15.0
+  version: 0.15.1
 servers:
   - url: '{protocol}://{host}:{port}'
     variables:

mix.exs

@@ -1,7 +1,7 @@
 defmodule ExNVR.Umbrella.MixProject do
   use Mix.Project
 
-  @version "0.15.0"
+  @version "0.15.1"
 
   def project do
     [