Fix bugs in handling opus in agora sink

Author: FelonEkonom

c_src/membrane_agora_plugin/sink.cpp

@@ -24,7 +24,7 @@ UNIFEX_TERM create(UnifexEnv *env, char *appId, char *token, char *channelId,
   scfg.enableVideo = true;
   scfg.useStringUid = false;
   if (state->service->initialize(scfg) != agora::ERR_OK) {
-    AG_LOG(ERROR, "Failed to initialize service");
+    AG_LOG(ERROR, "Failed to initialize sink service");
     unifex_release_state(env, state);
     return create_result_error(env, "Failed to initialize service");
   }
@@ -159,17 +159,10 @@ UNIFEX_TERM write_audio_data(UnifexEnv *env, UnifexPayload *payload,
 
   if (codec == CODEC_AUDIO_AAC) {
     audioFrameInfo.codec = agora::rtc::AUDIO_CODEC_TYPE::AUDIO_CODEC_AACLC;
-    printf("\nDUPA NIF AAC\n\n");
-    fflush(stdout);
   } else if (codec == CODEC_AUDIO_OPUS) {
     audioFrameInfo.codec = agora::rtc::AUDIO_CODEC_TYPE::AUDIO_CODEC_OPUS;
-    printf("\nDUPA NIF OPUS\n\n");
-    fflush(stdout);
   }
 
-  printf("\nDUPA NIF AFTER\n\n");
-  fflush(stdout);
-
   if (state->audioEncodedFrameSender->sendEncodedAudioFrame(
           reinterpret_cast<uint8_t *>(payload->data), payload->size,
           audioFrameInfo) != true) {

c_src/membrane_agora_plugin/source.cpp

@@ -19,7 +19,8 @@ UNIFEX_TERM create(UnifexEnv *env, char *appId, char *token, char *channelId,
   scfg.enableVideo = true;
   scfg.useStringUid = false;
   if (state->service->initialize(scfg) != agora::ERR_OK) {
-    AG_LOG(ERROR, "Failed to initialize service");
+    AG_LOG(ERROR, "Failed to initialize source service");
+    state->service = NULL;
     unifex_release_state(env, state);
     return create_result_error(env, "Failed to initialize service");
   }
@@ -77,21 +78,32 @@ UNIFEX_TERM create(UnifexEnv *env, char *appId, char *token, char *channelId,
 }
 
 void handle_destroy_state(UnifexEnv *env, SourceState *state) {
-  state->connection->unregisterObserver(state->connObserver.get());
-  state->connObserver.reset();
-
-  state->connection->getLocalUser()->unregisterLocalUserObserver(
-      state->localUserObserver.get());
-  state->connection->getLocalUser()->unregisterVideoEncodedFrameObserver(
-      state->videoEncodedFrameObserver.get());
-  state->connection->getLocalUser()->unregisterAudioFrameObserver(
-      state->audioFrameObserver.get());
-
-  state->localUserObserver.reset();
-  state->videoEncodedFrameObserver.reset();
-  state->audioFrameObserver.reset();
-
   UNUSED(env);
+  if (state->connection) {
+    state->connection->unregisterObserver(state->connObserver.get());
+    state->connObserver.reset();
+  }
+
+  if (state->connection && state->localUserObserver)
+    state->connection->getLocalUser()->unregisterLocalUserObserver(
+        state->localUserObserver.get());
+
+  if (state->connection && state->videoEncodedFrameObserver)
+    state->connection->getLocalUser()->unregisterVideoEncodedFrameObserver(
+        state->videoEncodedFrameObserver.get());
+  if (state->connection && state->audioFrameObserver)
+    state->connection->getLocalUser()->unregisterAudioFrameObserver(
+        state->audioFrameObserver.get());
+
+  if (state->localUserObserver)
+    state->localUserObserver.reset();
+  
+  if (state->videoEncodedFrameObserver)
+    state->videoEncodedFrameObserver.reset();
+  
+  if (state->audioFrameObserver)
+    state->audioFrameObserver.reset();
+  
   if (state->connection) {
     if (state->connection->disconnect()) {
       AG_LOG(ERROR, "Failed to disconnect from Agora channel!");

lib/agora/agora_sink.ex

@@ -7,6 +7,7 @@ defmodule Membrane.Agora.Sink do
 
   require Membrane.Pad, as: Pad
 
+  alias Membrane.{AAC, Opus}
   alias Membrane.Agora.Sink.Native
 
   def_input_pad :video,
@@ -17,7 +18,10 @@ defmodule Membrane.Agora.Sink do
   def_input_pad :audio,
     availability: :on_request,
     accepted_format: any_of(Membrane.AAC, Membrane.Opus),
-    flow_control: :auto
+    flow_control: :auto,
+    options: [
+      sample_rate: [default: nil]
+    ]
 
   def_options app_id: [
                 spec: String.t(),
@@ -54,7 +58,8 @@ defmodule Membrane.Agora.Sink do
       token: opts.token,
       channel_name: opts.channel_name,
       user_id: opts.user_id,
-      native_state: nil
+      native_state: nil,
+      last_frame_duration: nil
     }
 
     {[], state}
@@ -109,37 +114,13 @@ defmodule Membrane.Agora.Sink do
   end
 
   @impl true
-  def handle_stream_format(Pad.ref(:audio, _id) = pad, %Membrane.Opus{} = opus, ctx, state) do
-    {:ok, native_state} =
-      case ctx.pads[pad].options do
-        %{frame_duration: frame_duration, sample_rate: sample_rate} ->
-          samples_per_frame =
-            (Membrane.Time.as_milliseconds(frame_duration, :round) * sample_rate)
-            |> div(1000)
-
-          IO.inspect({sample_rate, opus.channels, samples_per_frame},
-            label: "HANDLE STREAM FORMAT"
-          )
-
-          Native.update_audio_stream_format(
-            sample_rate,
-            opus.channels,
-            samples_per_frame,
-            state.native_state
-          )
-
-        pad_options ->
-          raise """
-          Pad options has to have :samples_per_frame and :sample_rate options when the stream format is \
-          Membrane.Opus, but pad #{inspect(pad)} has options #{inspect(pad_options)}
-          """
-      end
-
-    {[], %{state | native_state: native_state}}
+  def handle_stream_format(Pad.ref(:audio, _id), %Opus{}, ctx, state) do
+    # audio stream format will be updated in handle_buffer/4
+    {[], state}
   end
 
   @impl true
-  def handle_buffer(Pad.ref(:video, _id), buffer, _ctx, state) do
+  def handle_buffer(Pad.ref(:video, _id) = pad, buffer, ctx, state) do
     :ok =
       Native.write_video_data(
         buffer.payload,
@@ -154,13 +135,35 @@ defmodule Membrane.Agora.Sink do
   def handle_buffer(Pad.ref(:audio, _id) = pad, buffer, ctx, state) do
     stream_format =
       case ctx.pads[pad].stream_format do
-        %Membrane.Opus{} -> :opus
-        %Membrane.AAC{} -> :aac
+        %Opus{} -> :opus
+        %AAC{} -> :aac
       end
 
-    {stream_format, buffer} |> IO.inspect(label: "HANDLE BUFFER")
+    state =
+      if stream_format == :opus and buffer.metadata.duration != state.last_frame_duration do
+        update_frame_duration(buffer.metadata.duration, pad, ctx, state)
+      else
+        state
+      end
 
     :ok = Native.write_audio_data(buffer.payload, stream_format, state.native_state)
     {[], state}
   end
+
+  defp update_frame_duration(frame_duration, pad, ctx, state) do
+    pad_data = ctx.pads[pad]
+
+    sample_rate = pad_data.options.sample_rate
+    samples_per_frame = (frame_duration * sample_rate) |> div(1000)
+
+    {:ok, native_state} =
+      Native.update_audio_stream_format(
+        sample_rate,
+        pad_data.stream_format.channels,
+        samples_per_frame,
+        state.native_state
+      )
+
+    %{state | native_state: native_state, last_frame_duration: frame_duration}
+  end
 end

mix.exs

@@ -43,6 +43,7 @@ defmodule Membrane.Agora.Mixfile do
       {:membrane_aac_format, "~> 0.8.0"},
       {:membrane_opus_format, "~> 0.3.0"},
       {:membrane_raw_audio_format, "~> 0.12.0"},
+      {:membrane_ogg_plugin, "~> 0.5.0"},
       {:unifex, "~> 1.1"},
       {:membrane_file_plugin, "~> 0.16.0", only: :test},
       {:membrane_h26x_plugin, "~> 0.10.0", only: :test},

mix.lock

@@ -5,9 +5,11 @@
   "bundlex": {:hex, :bundlex, "1.5.4", "3726acd463f4d31894a59bbc177c17f3b574634a524212f13469f41c4834a1d9", [:mix], [{:bunch, "~> 1.0", [hex: :bunch, repo: "hexpm", optional: false]}, {:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}, {:qex, "~> 0.5", [hex: :qex, repo: "hexpm", optional: false]}, {:req, ">= 0.4.0", [hex: :req, repo: "hexpm", optional: false]}, {:zarex, "~> 1.0", [hex: :zarex, repo: "hexpm", optional: false]}], "hexpm", "e745726606a560275182a8ac1c8ebd5e11a659bb7460d8abf30f397e59b4c5d2"},
   "bunt": {:hex, :bunt, "1.0.0", "081c2c665f086849e6d57900292b3a161727ab40431219529f13c4ddcf3e7a44", [:mix], [], "hexpm", "dc5f86aa08a5f6fa6b8096f0735c4e76d54ae5c9fa2c143e5a1fc7c1cd9bb6b5"},
   "coerce": {:hex, :coerce, "1.0.1", "211c27386315dc2894ac11bc1f413a0e38505d808153367bd5c6e75a4003d096", [:mix], [], "hexpm", "b44a691700f7a1a15b4b7e2ff1fa30bebd669929ac8aa43cffe9e2f8bf051cf1"},
+  "crc": {:hex, :crc, "0.10.5", "ee12a7c056ac498ef2ea985ecdc9fa53c1bfb4e53a484d9f17ff94803707dfd8", [:mix, :rebar3], [{:elixir_make, "~> 0.6", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "3e673b6495a9525c5c641585af1accba59a1eb33de697bedf341e247012c2c7f"},
   "credo": {:hex, :credo, "1.7.10", "6e64fe59be8da5e30a1b96273b247b5cf1cc9e336b5fd66302a64b25749ad44d", [:mix], [{:bunt, "~> 0.2.1 or ~> 1.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "71fbc9a6b8be21d993deca85bf151df023a3097b01e09a2809d460348561d8cd"},
   "dialyxir": {:hex, :dialyxir, "1.4.4", "fb3ce8741edeaea59c9ae84d5cec75da00fa89fe401c72d6e047d11a61f65f70", [:mix], [{:erlex, ">= 0.2.7", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "cd6111e8017ccd563e65621a4d9a4a1c5cd333df30cebc7face8029cacb4eff6"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.41", "ab34711c9dc6212dda44fcd20ecb87ac3f3fce6f0ca2f28d4a00e4154f8cd599", [:mix], [], "hexpm", "a81a04c7e34b6617c2792e291b5a2e57ab316365c2644ddc553bb9ed863ebefa"},
+  "elixir_make": {:hex, :elixir_make, "0.9.0", "6484b3cd8c0cee58f09f05ecaf1a140a8c97670671a6a0e7ab4dc326c3109726", [:mix], [], "hexpm", "db23d4fd8b757462ad02f8aa73431a426fe6671c80b200d9710caf3d1dd0ffdb"},
   "elixir_uuid": {:hex, :elixir_uuid, "1.2.1", "dce506597acb7e6b0daeaff52ff6a9043f5919a4c3315abb4143f0b00378c097", [:mix], [], "hexpm", "f7eba2ea6c3555cea09706492716b0d87397b88946e6380898c2889d68585752"},
   "erlex": {:hex, :erlex, "0.2.7", "810e8725f96ab74d17aac676e748627a07bc87eb950d2b83acd29dc047a30595", [:mix], [], "hexpm", "3ed95f79d1a844c3f6bf0cea61e0d5612a42ce56da9c03f01df538685365efb0"},
   "ex_doc": {:hex, :ex_doc, "0.34.2", "13eedf3844ccdce25cfd837b99bea9ad92c4e511233199440488d217c92571e8", [:mix], [{:earmark_parser, "~> 1.4.39", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "5ce5f16b41208a50106afed3de6a2ed34f4acfd65715b82a0b84b49d995f95c1"},
@@ -26,6 +28,7 @@
   "membrane_h264_format": {:hex, :membrane_h264_format, "0.6.1", "44836cd9de0abe989b146df1e114507787efc0cf0da2368f17a10c47b4e0738c", [:mix], [], "hexpm", "4b79be56465a876d2eac2c3af99e115374bbdc03eb1dea4f696ee9a8033cd4b0"},
   "membrane_h265_format": {:hex, :membrane_h265_format, "0.2.0", "1903c072cf7b0980c4d0c117ab61a2cd33e88782b696290de29570a7fab34819", [:mix], [], "hexpm", "6df418bdf242c0d9f7dbf2e5aea4c2d182e34ac9ad5a8b8cef2610c290002e83"},
   "membrane_h26x_plugin": {:hex, :membrane_h26x_plugin, "0.10.2", "caf2790d8c107df35f8d456b45f4e09fb9c56ce6c7669a3a03f7d59972e6ed82", [:mix], [{:bunch, "~> 1.4", [hex: :bunch, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}, {:membrane_h264_format, "~> 0.6.0", [hex: :membrane_h264_format, repo: "hexpm", optional: false]}, {:membrane_h265_format, "~> 0.2.0", [hex: :membrane_h265_format, repo: "hexpm", optional: false]}], "hexpm", "becf1ac4a589adecd850137ccd61a33058f686083a514a7e39fcd721bcf9fb2e"},
+  "membrane_ogg_plugin": {:hex, :membrane_ogg_plugin, "0.5.0", "1e910cee8f311e186f8ee26cd4ab06bde7e9d0072eb04eb1abb890372463308f", [:mix], [{:crc, "~> 0.10", [hex: :crc, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}, {:membrane_opus_format, "~> 0.3.0", [hex: :membrane_opus_format, repo: "hexpm", optional: false]}], "hexpm", "dd009453c1e01c4da0e6afc23f0037d5cf6b1d3fcdf79959a61768588e725186"},
   "membrane_opus_format": {:hex, :membrane_opus_format, "0.3.0", "3804d9916058b7cfa2baa0131a644d8186198d64f52d592ae09e0942513cb4c2", [:mix], [], "hexpm", "8fc89c97be50de23ded15f2050fe603dcce732566fe6fdd15a2de01cb6b81afe"},
   "membrane_opus_plugin": {:hex, :membrane_opus_plugin, "0.20.4", "0b018ab0cdd02e6fd1f27590a7382a81c0e2849212add0d76b2465385a7cfea7", [:mix], [{:bunch, "~> 1.3", [hex: :bunch, repo: "hexpm", optional: false]}, {:bundlex, "~> 1.2", [hex: :bundlex, repo: "hexpm", optional: false]}, {:membrane_common_c, "~> 0.16.0", [hex: :membrane_common_c, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}, {:membrane_opus_format, "~> 0.3.0", [hex: :membrane_opus_format, repo: "hexpm", optional: false]}, {:membrane_precompiled_dependency_provider, "~> 0.1.0", [hex: :membrane_precompiled_dependency_provider, repo: "hexpm", optional: false]}, {:membrane_raw_audio_format, "~> 0.12.0", [hex: :membrane_raw_audio_format, repo: "hexpm", optional: false]}, {:unifex, "~> 1.0", [hex: :unifex, repo: "hexpm", optional: false]}], "hexpm", "a3c81fc2648887b601448f6dcaf195063a30585d7f0d3190dff2c5316d759af8"},
   "membrane_precompiled_dependency_provider": {:hex, :membrane_precompiled_dependency_provider, "0.1.2", "8af73b7dc15ba55c9f5fbfc0453d4a8edfb007ade54b56c37d626be0d1189aba", [:mix], [{:bundlex, "~> 1.4", [hex: :bundlex, repo: "hexpm", optional: false]}], "hexpm", "7fe3e07361510445a29bee95336adde667c4162b76b7f4c8af3aeb3415292023"},

test/fixtures/in_audio.opus

@@ -3,8 +3,9 @@ defmodule Membrane.Agora.IntegrationTest do
   import Membrane.Testing.Assertions
   alias Membrane.Agora.Support.{ReceiverPipeline, SenderPipeline}
 
-  for audio_codec <- [:opus] do
+  for audio_codec <- [:opus, :aac] do
     @tag :tmp_dir
+    @tag audio_codec
     test "if the data is sent to Agora properly when audio codec is #{audio_codec}", %{
       tmp_dir: dir
     } do
@@ -14,9 +15,15 @@ defmodule Membrane.Agora.IntegrationTest do
       input_video = "test/fixtures/in_video.h264"
       output_video = "#{dir}/video.h264"
       reference_video = input_video
+      audio_codec = unquote(audio_codec)
 
-      input_audio = "test/fixtures/in_audio.#{unquote(audio_codec)}"
-      output_audio = "#{dir}/audio_#{unquote(audio_codec)}.pcm"
+      input_audio =
+        case audio_codec do
+          :aac -> "test/fixtures/in_audio.aac"
+          :opus -> "test/fixtures/in_audio_opus.ogg"
+        end
+
+      output_audio = "#{dir}/audio_#{audio_codec}.pcm"
       reference_audio = "test/fixtures/in_audio.pcm"
 
       {:ok, _supervisor, receiver_pipeline} =
@@ -31,6 +38,8 @@ defmodule Membrane.Agora.IntegrationTest do
           custom_args: [audio: input_audio, video: input_video, framerate: framerate]
         )
 
+      Process.sleep(3000)
+
       assert_start_of_stream(receiver_pipeline, :video_sink, :input, 10_000)
       assert_start_of_stream(receiver_pipeline, :audio_sink)
 

test/fixtures/in_audio_opus.ogg

@@ -12,20 +12,25 @@ defmodule Membrane.Agora.Support.SenderPipeline do
   def handle_init(_ctx, opts) do
     user_id = "12"
 
-    {audio_parser, sink_audio_options} =
+    {plug_audio_parsing, sink_audio_options} =
       opts[:audio]
       |> String.split(".")
       |> List.last()
       |> case do
         "aac" ->
-          {Membrane.AAC.Parser, []}
-
-        "opus" ->
-          {Membrane.Opus.Parser,
-           [
-             sample_rate: 48_000,
-             frame_duration: Membrane.Time.milliseconds(20)
-           ]}
+          {&child(&1, Membrane.AAC.Parser), []}
+
+        "ogg" ->
+          plug_audio_parser =
+            fn spec ->
+              spec
+              |> child(:ogg_demuxer, Membrane.Ogg.Demuxer)
+              |> child(:opus_parser, %Membrane.Opus.Parser{
+                generate_best_effort_timestamps?: true
+              })
+            end
+
+          {plug_audio_parser, [sample_rate: 48_000]}
       end
 
     spec =
@@ -43,16 +48,7 @@ defmodule Membrane.Agora.Support.SenderPipeline do
           user_id: user_id
         }),
         child(%Membrane.File.Source{location: opts[:audio]})
-        |> child(audio_parser)
-        |> case do
-          spec when audio_parser == Membrane.AAC.Parser ->
-            spec
-
-          spec when audio_parser == Membrane.Opus.Parser ->
-            spec
-            |> child(Membrane.Opus.Decoder)
-            |> child(Membrane.Opus.Encoder)
-        end
+        |> plug_audio_parsing.()
         |> child(Membrane.Realtimer)
         |> via_in(:audio, options: sink_audio_options)
         |> get_child(:sink)

test/integration_test.exs

@@ -0,0 +1 @@
+rtc.enable_xdump|1|rtc.enable_xdump_file|1|rtc.enable_xdump_upload|1|