Flake: HM module

lib/peer.ex

@@ -17,18 +17,8 @@ defmodule Frajtano.Peer do
     {:ok, %{conn: conn, clients: :queue.new(), buffer: <<>>}}
   end
 
-  @impl true
-  def handle_call(packet, from, %{conn: conn, clients: clients} = state) do
-    case :gen_tcp.send(conn, Proto.encode(packet)) do
-      :ok ->
-        {:noreply, %{state | clients: :queue.in(from, clients)}}
-
-      {:error, :closed} ->
-        {:noreply, state, {:continue, :closed}}
-
-      {:error, e} ->
-        raise(e)
-    end
+  def reply({client, ref}, msg) do
+    send(client, {ref, msg})
   end
 
   defp handle_messages(%{clients: clients, buffer: buffer} = state) do
@@ -38,42 +28,85 @@ defmodule Frajtano.Peer do
 
       {msg, buffer} ->
         {{:value, client}, clients} = :queue.out(clients)
-        GenServer.reply(client, {:ok, msg})
+        reply(client, {:ok, msg})
         handle_messages(%{state | clients: clients, buffer: buffer})
     end
   end
 
+  @impl true
+  def handle_info({:send, packet, from}, %{conn: conn, clients: clients} = state) do
+    case :gen_tcp.send(conn, Proto.encode(packet)) do
+      :ok ->
+        {:noreply, %{state | clients: :queue.in(from, clients)}}
+
+      {:error, e} ->
+        {:noreply, state, {:continue, {:error, e}}}
+    end
+  end
+
   @impl true
   def handle_info({:tcp, conn, msg}, %{conn: conn, buffer: buffer} = state) do
     :inet.setopts(conn, active: :once)
     buffer = buffer <> msg
 
-    handle_messages(%{ state | buffer: buffer })
+    handle_messages(%{state | buffer: buffer})
   end
 
   @impl true
   def handle_info({:tcp_closed, _}, state) do
-    {:noreply, state, {:continue, :closed}}
+    {:noreply, state, {:continue, {:error, :closed}}}
   end
 
   @impl true
-  def handle_continue(:closed, %{clients: clients}) do
+  def handle_info(:timeout, state) do
+    {:noreply, state, {:continue, {:error, :closed}}}
+  end
+
+  @impl true
+  def handle_continue({:error, e}, %{clients: clients}) do
     clients
     |> :queue.to_list()
-    |> Enum.each(&GenServer.reply(&1, {:error, :closed}))
+    |> Enum.each(&reply(&1, {:error, e}))
 
-    {:stop, :closed, %{}}
+    {:stop, {:error, e}, %{}}
   end
 
   def identities(peer) do
-    with {:ok, {:agent_identities_answer, identities}} <-
-           GenServer.call(peer, {:agentc_request_identities, nil}),
-         do: {:ok, identities}
+    ref = make_ref()
+    send(peer, {:send, {:agentc_request_identities, nil}, {self(), ref}})
+    # Needs to be less than the timeout in Frajtano.Agent.identities on the Task.async_stream call
+    # That's 5000 by default
+    timer = Process.send_after(peer, :timeout, 4500)
+
+    receive do
+      {^ref, msg} ->
+        Process.cancel_timer(timer)
+        msg
+    end
+    |> case do
+      {:ok, {:agent_identities_answer, identities}} -> {:ok, identities}
+      {:ok, {:agent_failure, nil}} -> {:error, :agent_failure}
+      {:ok, msg} -> raise("Unexpected message #{inspect(msg)}")
+      {:error, e} -> {:error, e}
+    end
   end
 
   def sign(peer, request) do
-    with {:ok, {:agent_sign_response, signature}} <-
-           GenServer.call(peer, {:agentc_sign_request, request}),
-         do: {:ok, signature}
+    # Signing may take some time, as a password may need to be entered or similar
+    # There is therefore no timeout
+    # If something requests identities afterwards, it will timeout, which also kills this signature request
+    # The SSH agent protocol strict ordering leaves fun problems with timeouts, as it turns out
+    ref = make_ref()
+    send(peer, {:send, {:agentc_sign_request, request}, {self(), ref}})
+
+    receive do
+      {^ref, msg} -> msg
+    end
+    |> case do
+      {:ok, {:agent_sign_response, signature}} -> {:ok, signature}
+      {:ok, {:agent_failure, nil}} -> {:error, :agent_failure}
+      {:ok, msg} -> raise("Unexpected message #{inspect(msg)}")
+      {:error, e} -> {:error, e}
+    end
   end
 end