fix: create_worker to async

lib/nif.ex

@@ -8,12 +8,11 @@ defmodule Mediasoup.Nif do
   alias Mediasoup.{Worker, Router}
 
   # construct worker
-  @spec create_worker() :: {:ok, reference()} | {:error, String.t()}
-  def create_worker(), do: :erlang.nif_error(:nif_not_loaded)
+  defp create_worker_async(), do: :erlang.nif_error(:nif_not_loaded)
+  defp create_worker_async(_option), do: :erlang.nif_error(:nif_not_loaded)
 
-  @spec create_worker(Worker.Settings.t()) ::
-          {:ok, reference()} | {:error, String.t()}
-  def create_worker(_option), do: :erlang.nif_error(:nif_not_loaded)
+  def create_worker(), do: create_worker_async() |> handle_async_nif_result()
+  def create_worker(option), do: create_worker_async(option) |> handle_async_nif_result()
 
   # worker
   @spec worker_create_router(reference, Router.create_option()) :: {:ok, reference()} | {:error}
@@ -185,4 +184,16 @@ defmodule Mediasoup.Nif do
   @spec producer_event(reference, pid, [atom()]) :: {:ok} | {:error}
   def producer_event(_producer, _pid, _event_types), do: :erlang.nif_error(:nif_not_loaded)
   def producer_dump(_producer), do: :erlang.nif_error(:nif_not_loaded)
+
+  def handle_async_nif_result(result) do
+    case result do
+      {:ok, result_key} ->
+        receive do
+          {^result_key, msg} -> msg
+        end
+
+      error ->
+        error
+    end
+  end
 end

native/mediasoup_elixir/src/atoms.rs

@@ -27,4 +27,5 @@ rustler::atoms! {
     on_layers_change,
     audio,
     video,
+    create_worker,
 }

native/mediasoup_elixir/src/lib.rs

@@ -48,13 +48,14 @@ use crate::worker::{
     worker_dump, worker_event, worker_id, worker_update_settings,
 };
 
+use futures_lite::future;
 use mediasoup::consumer::Consumer;
 use mediasoup::pipe_transport::PipeTransport;
 use mediasoup::producer::Producer;
 use mediasoup::router::Router;
 use mediasoup::webrtc_transport::WebRtcTransport;
 use mediasoup::worker::Worker;
-use rustler::{Env, LocalPid, OwnedEnv, Term};
+use rustler::{Atom, Encoder, Env, LocalPid, OwnedEnv, Term};
 
 pub fn send_msg_from_other_thread<T>(pid: LocalPid, value: T)
 where
@@ -66,6 +67,25 @@ where
     });
 }
 
+pub fn send_async_nif_result<T, E, Fut>(env: Env, msg: Atom, future: Fut)
+where
+    T: Encoder,
+    E: Encoder,
+    Fut: future::Future<Output = Result<T, E>> + Send + 'static,
+{
+    let pid = env.pid();
+    let mut my_env = OwnedEnv::new();
+    std::thread::spawn(move || {
+        let result = future::block_on(future);
+        match result {
+            Ok(worker) => {
+                my_env.send_and_clear(&pid, |env| (msg, (atoms::ok(), worker)).encode(env))
+            }
+            Err(err) => my_env.send_and_clear(&pid, |env| (msg, (atoms::error(), err)).encode(env)),
+        }
+    });
+}
+
 rustler::init! {
     "Elixir.Mediasoup.Nif",
     [

native/mediasoup_elixir/src/worker.rs

@@ -1,14 +1,14 @@
 use crate::atoms;
 use crate::json_serde::JsonSerdeWrap;
 use crate::router::RouterOptionsStruct;
-use crate::{send_msg_from_other_thread, RouterRef, WorkerRef};
+use crate::{send_async_nif_result, send_msg_from_other_thread, RouterRef, WorkerRef};
 use futures_lite::future;
 use mediasoup::worker::{
     WorkerDtlsFiles, WorkerDump, WorkerId, WorkerLogLevel, WorkerLogTag, WorkerSettings,
     WorkerUpdateSettings,
 };
 use mediasoup::worker_manager::WorkerManager;
-use rustler::{Error, NifResult, NifStruct, ResourceArc};
+use rustler::{Env, Error, NifResult, NifStruct, ResourceArc};
 use std::path::PathBuf;
 
 #[rustler::nif]
@@ -111,28 +111,33 @@ pub fn worker_event(
 }
 
 fn create_worker_impl(
+    env: Env,
     settings: WorkerSettings,
-) -> NifResult<(rustler::Atom, ResourceArc<WorkerRef>)> {
-    let worker_manager = WorkerManager::new();
+) -> NifResult<(rustler::Atom, rustler::Atom)> {
+    send_async_nif_result(env, atoms::create_worker(), async move {
+        let worker_manager = WorkerManager::new();
+        worker_manager
+            .create_worker(settings)
+            .await
+            .map(WorkerRef::resource)
+            .map_err(|error| format!("{}", error))
+    });
 
-    let worker = future::block_on(async move {
-        return worker_manager.create_worker(settings).await;
-    })
-    .map_err(|error| Error::Term(Box::new(format!("{}", error))))?;
-    Ok((atoms::ok(), WorkerRef::resource(worker)))
+    Ok((atoms::ok(), atoms::create_worker()))
 }
 
-#[rustler::nif(name = "create_worker")]
-pub fn create_worker_no_arg() -> NifResult<(rustler::Atom, ResourceArc<WorkerRef>)> {
-    create_worker_impl(WorkerSettings::default())
+#[rustler::nif(name = "create_worker_async")]
+pub fn create_worker_no_arg(env: Env) -> NifResult<(rustler::Atom, rustler::Atom)> {
+    create_worker_impl(env, WorkerSettings::default())
 }
 
-#[rustler::nif]
+#[rustler::nif(name = "create_worker_async")]
 pub fn create_worker(
+    env: Env,
     settings: WorkerSettingsStruct,
-) -> NifResult<(rustler::Atom, ResourceArc<WorkerRef>)> {
+) -> NifResult<(rustler::Atom, rustler::Atom)> {
     let settings = settings.try_to_setting()?;
-    create_worker_impl(settings)
+    create_worker_impl(env, settings)
 }
 
 #[derive(NifStruct)]

test/integration/test_worker.ex

@@ -128,4 +128,15 @@ defmodule IntegrateTest.WorkerTest do
 
     refute Worker.closed?(worker)
   end
+
+  def create_many_worker() do
+    1..200
+    |> Enum.map(fn _ ->
+      {:ok, worker} = Worker.start_link()
+      worker
+    end)
+    |> Enum.map(fn worker ->
+      Worker.close(worker)
+    end)
+  end
 end

test/worker_test.exs

@@ -37,4 +37,8 @@ defmodule WorkerTest do
   test "close_router" do
     IntegrateTest.WorkerTest.close_router()
   end
+
+  test "create_many_worker" do
+    IntegrateTest.WorkerTest.create_many_worker()
+  end
 end