Kazoo Elixir Port

kzx.ex

defmodule Kazoo do
  require Logger
  # original from https://github.com/2600hz/kazoo/blob/4.3.142/applications/ecallmgr/src/mod_kazoo.erl
  @timeout 3000

  @type freeswitch_node :: atom
  @type freeswitch_section_binding ::
          :config | :directory | :dialplan | :languages | :chatplan | :channels
  @type freeswitch_event_binding :: atom
  @type freeswitch_api_ok :: {:ok, binary()} | :ok
  @type freeswitch_api_error :: {:error, :timeout | :exception | :badarg | binary()}
  @type freeswitch_api_return :: freeswitch_api_ok() | freeswitch_api_error()

  @doc "Returns the current mod_kazoo version running on FreeSWITCH."
  @spec version(node :: freeswitch_node) :: freeswitch_api_return()
  def version(node), do: call(node, :version)

  @doc "Instructs FreeSWITCH to stop sending events."
  @spec no_events(node :: freeswitch_node) :: freeswitch_api_return()
  def no_events(node), do: call(node, :noevents)

  @doc "Closes the connection to the FreeSWITCH node."
  @spec close(node :: freeswitch_node) :: :ok
  def close(node), do: call(node, :exit)

  @doc "Returns the PID for the FreeSWITCH node."
  @spec get_pid(node :: freeswitch_node) :: {:ok, pid}
  def get_pid(node), do: call(node, :getpid)

  @doc """
  Instructs the FreeSWITCH node to begin requesting configuration from
  the calling process.

  ## Examples

    iex> bind(:"[email protected]", :dialplan)
  """
  @spec bind(node :: freeswitch_node, binding :: freeswitch_section_binding) ::
          freeswitch_api_return()
  def bind(node, binding), do: call(node, {:bind, binding})

  @doc """
  Reply to a configuration request sent from a FreeSWITCH node.
  """
  @spec fetch_reply(
          node :: freeswitch_node,
          fetch_id :: String.t(),
          section :: String.t(),
          reply :: String.t()
        ) :: freeswitch_api_return()
  def fetch_reply(node, fetch_id, section, reply),
    do: cast(node, {:fetch_reply, section, fetch_id, reply})

  @doc """
  Instructs FreeSWITCH to start sending events. Returns a tuple with
  the IP and port that we can listen on to consume the events.
  """
  @spec subscribe_events(node :: freeswitch_node, bindings :: [freeswitch_event_binding]) ::
          {:ok, :inet.ip(), :inet.port()}
  def subscribe_events(node, bindings), do: call(node, {:event, List.wrap(bindings)})

  @doc """
  Call an API on FreeSWITCH.
  """
  @spec api(node :: atom, cmd :: atom, args :: String.t()) :: freeswitch_api_return()
  def api(node, cmd, args \\ ""), do: call(node, {:api, cmd, args})

  @doc """
  Calls an API command in a background thread, returning the job id. The caller is responsible for receiving the message 
  """
  def async_bgapi(node, cmd, args \\ "") do
    call(node, {:bgapi, cmd, args})
  end

  @doc """
  Calls an API command in a background thread.
  """
  @spec bgapi(node :: atom, cmd :: atom, args :: String.t()) :: freeswitch_api_return()
  def bgapi(node, cmd, args \\ "") do
    parent = self()

    spawn(fn ->
      try do
        call(node, {:bgapi, cmd, args})
      catch
        e, r ->
          _ =
            Logger.info(fn ->
              formatted = Exception.format(e, r)
              {"exec bgapi #{to_string(cmd)} #{args} failed with #{formatted}", [node: node]}
            end)

          send(parent, {:api, {:error, :exception}})
      else
        {:ok, job_id} = job_ok ->
          send(parent, {:api, job_ok})

          receive do
            {:bgok, ^job_id, _} = bg_ok -> send(parent, bg_ok)
            {:bgerror, ^job_id, _} = bg_error -> send(parent, bg_error)
          after
            @timeout -> send(parent, {:bgerror, job_id, :timeout})
          end

        {:error, reason} ->
          send(parent, {:api, {:error, reason}})

        :timeout ->
          send(parent, {:api, {:error, :timeout}})
      end
    end)

    receive do
      {:api, result} -> result
    end
  end

  def bgapi4(node, cmd, args, fun, params) do
    parent = self()

    spawn(fn ->
      case call(node, {:bgapi4, cmd, args}) do
        {:ok, "-ERR " <> reason} ->
          send(parent, {:api, internal_fs_error(reason)})

        {:ok, job_id} = job_ok ->
          send(parent, {:api, job_ok})

          receive do
            {:bgok, ^job_id, reply} when is_function(fun, 3) ->
              send(parent, fun.(:ok, reply, [job_id | params]))

            {:bgerror, ^job_id, reply} when is_function(fun, 3) ->
              send(parent, fun.(:error, reply, [job_id | params]))

            {:bgok, ^job_id, reply, data} when is_function(fun, 4) ->
              fun.(:ok, reply, data, [job_id | params])

            {:bgerror, ^job_id, reply, data} when is_function(fun, 4) ->
              fun.(:error, reply, data, [job_id | params])
          end

        {:error, _} = err ->
          send(parent, {:api, err})

        :timeout ->
          send(parent, {:api, {:error, :timeout}})
      end
    end)

    receive do
      {:api, result} -> result
    end
  end

  def json_api(node, cmd, args \\ nil, uuid \\ nil) do
    json =
      Jason.encode!(%{
        "command" => cmd,
        "data" => args,
        "uuid" => uuid
      })

    case call(node, {:json_api, json}) do
      {:ok, response} when is_binary(response) -> 
        {:ok, Jason.decode!(response)}
      {:error, response} when is_binary(response) ->
        {:error, Jason.decode!(response)}
      other ->
        other
    end
  end
  
  @doc false
  def call(node, args, timeout \\ @timeout) do
    GenServer.call({:mod_kazoo, node}, args, timeout)
  catch
    :exit, {{:nodedown, node}, _} ->
      {:error, {:nodedown, node}}

    error, reason ->
      Logger.warning(fn ->
        {
          "Attempt to call #{inspect(args)} failed:\n" <> Exception.format(error, reason),
          node: node
        }
      end)

      {:error, :exception}
  end

  def cast(node, args), do: GenServer.cast({:mod_kazoo, node}, args)

  def internal_fs_error(reason) do
    error = String.replace(reason, "\n", "")
    {:error, error}
  end
end

Thank you for this! Especially for the reminder that one can use any(?) BEAM-based language to interact with C-nodes; I keep forgetting this..:)

No problem, hope it helps!

any(?)

I am going to assume yes even though I have not personally tested every BEAM language 😄

Yes! You could even experiment with it if you wanted to try Elixir, like maybe in a different FreeSWITCH context! I'm fuzzy on the details because it's been a while, but I think Kazoo does bind(:freeswitch_node, :dialplan), which causes you to start receiving {:fetch, _} messages (I thought _ was a tuple, but maybe it has changed). You respond to those messages using fetch_reply/4.

The basic "entrypoint" (eliding all the other stuff that happens before getting to the dialplan) for calls in Kazoo is something like this:

<context name="kazoo">
  <extension name="entrypoint">
    <condition>
      <action application="park"/>
    </condition>
  </extension>
</context>

This can be the thing you give with fetch_reply/4 (last argument named reply) or maybe you don't need to bind to the dialplan at all and could just subscribe_events(node, [:HEARTBEAT, :CHANNEL_CREATE])? I am imagining you could transfer the parked call to whatever controlling application you want.

I realized this gist is incomplete, missing some sendcmd stuff.

Thanks again! I've been meaning to dive into mod_kazoo for a while now, but it's undocumented status discouraged, so I appreciate giving me an entry point:)