Ability to connect/disconnect ports

lib/ex_jack/native.ex

@@ -39,6 +39,8 @@ defmodule ExJack.Native do
   end
 
   def _start(_opts), do: error()
+  def connect_ports(_resource, _port_from_name, _port_to_name), do: error()
+  def disconnect_ports(_resource, _port_from_name, _port_to_name), do: error()
   def stop(_resource), do: error()
   def send_frames(_resource, _frames), do: error()
 

lib/ex_jack/server.ex

@@ -24,6 +24,7 @@ defmodule ExJack.Server do
 
   defstruct handler: nil,
             shutdown_handler: nil,
+            port_handler: nil,
             current_frame: 0,
             buffer_size: 0,
             sample_rate: 44100,
@@ -35,6 +36,7 @@ defmodule ExJack.Server do
   @type t :: %__MODULE__{
           handler: any(),
           shutdown_handler: any(),
+          port_handler: any(),
           current_frame: pos_integer(),
           buffer_size: buffer_size_t,
           sample_rate: sample_rate_t,
@@ -120,6 +122,22 @@ defmodule ExJack.Server do
     GenServer.call(__MODULE__, :ports)
   end
 
+  @doc """
+  Connect an output port to an input port
+  """
+  @spec connect_ports(port_t, port_t) :: any()
+  def connect_ports(port_from_name, port_to_name) do
+    GenServer.call(__MODULE__, {:connect_ports, port_from_name, port_to_name})
+  end
+
+  @doc """
+  Disconnect an output port to an input port
+  """
+  @spec disconnect_ports(port_t, port_t) :: any()
+  def disconnect_ports(port_from_name, port_to_name) do
+    GenServer.call(__MODULE__, {:disconnect_ports, port_from_name, port_to_name})
+  end
+
   @doc """
   Set the callback function that will receive input data from JACK each cycle.
 
@@ -142,13 +160,14 @@ defmodule ExJack.Server do
 
   @impl true
   def init(opts) do
-    {:ok, handler, shutdown_handler, ports, %{buffer_size: buffer_size, sample_rate: sample_rate}} =
-      ExJack.Native.start(opts)
+    {:ok, handler, shutdown_handler, port_handler, ports,
+     %{buffer_size: buffer_size, sample_rate: sample_rate}} = ExJack.Native.start(opts)
 
     {:ok,
      %__MODULE__{
        handler: handler,
        shutdown_handler: shutdown_handler,
+       port_handler: port_handler,
        current_frame: 0,
        buffer_size: buffer_size,
        sample_rate: sample_rate,
@@ -185,6 +204,38 @@ defmodule ExJack.Server do
     {:reply, ports, state}
   end
 
+  @impl true
+  @spec handle_call({:connect_ports, port_t, port_t}, GenServer.from(), t()) ::
+          {:reply, ports_t(), t()}
+  def handle_call(
+        {:connect_ports, port_from_name, port_to_name},
+        _from,
+        %{ports: ports, port_handler: port_handler} = state
+      ) do
+    if Map.has_key?(ports, port_from_name) and Map.has_key?(ports, port_to_name) do
+      ret = ExJack.Native.connect_ports(port_handler, port_from_name, port_to_name)
+      {:reply, ret, state}
+    else
+      {:reply, {:error, :ports_not_found}, state}
+    end
+  end
+
+  @impl true
+  @spec handle_call({:disconnect_ports, port_t, port_t}, GenServer.from(), t()) ::
+          {:reply, ports_t(), t()}
+  def handle_call(
+        {:disconnect_ports, port_from_name, port_to_name},
+        _from,
+        %{ports: ports, port_handler: port_handler} = state
+      ) do
+    if Map.has_key?(ports, port_from_name) and Map.has_key?(ports, port_to_name) do
+      ret = ExJack.Native.disconnect_ports(port_handler, port_from_name, port_to_name)
+      {:reply, ret, state}
+    else
+      {:reply, {:error, :ports_not_found}, state}
+    end
+  end
+
   @impl true
   @spec handle_cast({:set_output_func, output_func_t}, t()) :: {:noreply, t()}
   def handle_cast({:set_output_func, output_func}, state) do

native/ex_jack/src/lib.rs

@@ -9,12 +9,14 @@ type Sample = f32;
 
 pub struct SendFramesChannel(Mutex<mpsc::Sender<Vec<Sample>>>);
 pub struct ShutdownChannel(Mutex<Option<mpsc::Sender<()>>>);
+pub struct PortActionChannel(Mutex<mpsc::Sender<PortAction>>);
 
 type StartResult = Result<
     (
         Atom,
         ResourceArc<SendFramesChannel>,
         ResourceArc<ShutdownChannel>,
+        ResourceArc<PortActionChannel>,
         Vec<String>,
         Pcm,
     ),
@@ -27,6 +29,13 @@ pub struct Pcm {
     pub sample_rate: usize,
 }
 
+#[derive(NifMap)]
+pub struct PortAction {
+    pub connect: bool,
+    pub port_from_name: String,
+    pub port_to_name: String,
+}
+
 #[derive(NifMap)]
 pub struct Config {
     pub name: String,
@@ -37,6 +46,7 @@ pub struct Config {
 pub fn load(env: Env, _: Term) -> bool {
     rustler::resource!(SendFramesChannel, env);
     rustler::resource!(ShutdownChannel, env);
+    rustler::resource!(PortActionChannel, env);
     true
 }
 
@@ -59,6 +69,7 @@ pub fn _start(env: Env, config: Config) -> StartResult {
 
     let (shutdown_tx, shutdown_rx) = mpsc::channel::<()>();
     let (frames_tx, frames_rx) = mpsc::channel::<Vec<Sample>>();
+    let (manage_ports_tx, manage_ports_rx) = mpsc::channel::<>();
 
     let use_callback = config.use_callback;
     let process = jack::ClosureProcessHandler::new(
@@ -107,19 +118,46 @@ pub fn _start(env: Env, config: Config) -> StartResult {
                 .unwrap();
         }
 
-        let ten_seconds = time::Duration::from_secs(10);
-        while let Err(_) = shutdown_rx.try_recv() {
-            thread::sleep(ten_seconds);
+        let poll_interval = time::Duration::from_secs(5);
+        loop {
+            if let Ok(_) = shutdown_rx.try_recv() {
+                break;
+            }
+
+            if let Ok(PortAction {
+                connect,
+                port_from_name,
+                port_to_name
+            }) = manage_ports_rx.try_recv() {
+                if connect {
+                    // TODO at the moment if connecting/disconnecting fails
+                    // the client is not informed.
+                    // See issue https://github.com/dulltools/ex_jack/issues/18
+                    // for possible solutions
+                    let _ = active_client
+                        .as_client()
+                        .connect_ports_by_name(&port_from_name, &port_to_name);
+                } else {
+                    let _ = active_client
+                        .as_client()
+                        .disconnect_ports_by_name(&port_from_name, &port_to_name);
+                }
+
+            }
+
+            thread::sleep(poll_interval);
         }
     });
 
     let shutdown_ref = ResourceArc::new(ShutdownChannel(Mutex::new(Some(shutdown_tx))));
     let sender_ref = ResourceArc::new(SendFramesChannel(Mutex::new(frames_tx)));
+    let manage_ports_ref = ResourceArc::new(PortActionChannel(Mutex::new(manage_ports_tx)));
 
     Ok((
         atoms::ok(),
         sender_ref,
         shutdown_ref,
+        manage_ports_ref,
         already_connected_ports,
         Pcm {
             buffer_size,
@@ -231,6 +269,28 @@ fn send_frames(resource: ResourceArc<SendFramesChannel>, frames: Vec<Sample>) ->
     atoms::ok()
 }
 
+#[rustler::nif]
+fn connect_ports(resource: ResourceArc<PortActionChannel>, port_from_name: String, port_to_name: String) -> Atom {
+    let arc = resource.0.lock().unwrap().clone();
+    let _ = arc.send(PortAction {
+        connect: true,
+        port_to_name,
+        port_from_name,
+    });
+    atoms::ok()
+}
+
+#[rustler::nif]
+fn disconnect_ports(resource: ResourceArc<PortActionChannel>, port_from_name: String, port_to_name: String) -> Atom {
+    let arc = resource.0.lock().unwrap().clone();
+    let _ = arc.send(PortAction {
+        connect: false,
+        port_to_name,
+        port_from_name,
+    });
+    atoms::ok()
+}
+
 #[rustler::nif]
 pub fn stop(resource: ResourceArc<ShutdownChannel>) -> Atom {
     let mut lock = resource.0.lock().unwrap();
@@ -244,6 +304,6 @@ pub fn stop(resource: ResourceArc<ShutdownChannel>) -> Atom {
 
 rustler::init!(
     "Elixir.ExJack.Native",
-    [_start, stop, send_frames],
+    [_start, connect_ports, disconnect_ports, stop, send_frames],
     load = load
 );