High level abstraction that represents connection in the most common sense. Implementation can use any transport to connect to any remote or VM-local resource: Agent, GenServer, different http-APIs, etc.

Callbacks

• init/2 — initialize connection with given arguments;
• resource/1 — resource for that interaction is made;
• methods!/1 — methods supported while making Conn.call/3;
• call/3 — interact via selected method;
• parse/2 — parse data in respect of connection context.

Example

Let’s implement simple generic text-base RPC protocol. Client can send server requests in the form of :COMMAND or :COMMAND:ARGS to execute COMMAND with given ARGS or without them. There is a special request :COMMANDS to ask for a list of all available commands, response for it is ok:COMMAND1,CMD2,….

Server may reply with ok:COMMAND1,CMD2,…, ok, ok:DATA or err:REASON, where REASON could be notsupported (command), badarg, parse, or arbitrary string.

Now simple test conn will look like:

defmodule TextConn, do: defstruct [:res]

Let’s implement protocol described above:

defimpl Conn, for: TextConn do
  def init(conn, server) do
    if Process.alive?(server) do
      {:ok, %TextConn{res: server}}
    else
      {:error, :dead, :closed, conn}
    end
  end

  def resource(%_{res: server}), do: server

  def methods!(%_{res: server}=conn) do
    send(server, {self(), ":COMMANDS"})

    receive do
      "ok:" <> cmds ->
        String.split(",")
    end
  end

  def call(%_{res: server} = conn, cmd, args \\ nil) when is_binary(args) do
    if Process.alive?(server) do
      send(server, {self(), args && ":#{cmd}:#{args}" || ":#{cmd}"})

      receive do
        "ok" ->
          {:noreply, conn}

        "ok:" <> reply ->
          {:reply, reply, conn}

        "err:notsupported" ->
          {:error, :notsupported, conn}

        "err:"<>reason ->
          # Suggest 50 ms timeout.
          {:error, reason, 50, conn}
      after
        5000 ->
          {:error, :timeout, conn}
      end
    else
      {:error, :closed}
    end
  end

— it’s the part of the implementation used on the client side. Let’s move to parsing:

  def parse(conn, ""), do: :ok
  def parse(conn, ":COMMANDS" <> _), do: {:ok, :methods, ""}

  def parse(conn, ":" <> data) do
    case Regex.named_captures(~r[(?<cmd>[^:]+)(?::(?<args>.*))?], data) do
      %{"cmd" => cmd, "args" => args} ->
        {:ok, {:call, cmd, args}, ""}

      _ ->
        {:error, {:parse, data}, ""}
    end
  end

  def parse(conn, malformed) do
    {:error, {:parse, malformed}, ""}
  end
end

That’s it. Now all that left is to spawn corresponding server that will interact with client using above protocol. Let us implement server that holds integer value and supports commands INC, GET and STOP.

iex> defmodule Server do
...>   def loop(state \\ 0) do
...>     receive do
...>       {client, data} ->
...>         case Conn.parse(%TextConn{}, data) do
...>           {:error, {:parse, _}, _rest} ->
...>              send(client, "err:parse")
...>              loop(state)
...>
...>           {:ok, :methods, _rest} ->
...>              send(client, "ok:INC,GET,STOP")
...>              loop(state)
...>
...>           {:ok, {:call, "INC", ""}, _rest} ->
...>              send(client, "ok:#{state+1}")
...>              loop(state+1)
...>
...>           {:ok, {:call, "GET", ""}, _rest} ->
...>              send(client, "ok:#{state}")
...>              loop(state)
...>
...>           {:ok, {:call, "STOP", ""}, _rest} ->
...>              send(client, "ok")
...>
...>           {:ok, {:call, _, ""}, _rest} ->
...>              send(client, "err:notsupported")
...>              loop(state)
...>
...>           {:ok, {:call, _, _}, _rest} ->
...>              send(client, "err:badarg")
...>              loop(state)
...>         end
...>     end
...>   end
...> end
...>
...>
iex> res = spawn_link(&Server.loop/0)
iex> {:ok, pool} = Conn.Pool.start_link()
iex> Conn.Pool.init(pool, %TextConn{}, res)
iex> Conn.Pool.call(pool, res, "GET")
{:ok, "0"}
iex> Conn.Pool.call(pool, res, "INC")
{:ok, "1"}
iex> Conn.Pool.call(pool, res, "DEC")
{:error, :method}
iex> Conn.Pool.call(pool, res, "INC", :badarg)
{:error, "badarg"}
iex> Conn.Pool.call(pool, res, "STOP")
:ok
iex> Conn.Pool.call(pool, res, "GET")
{:error, :resource}

TextConn could be used to connect to any server that implements above protocol.