Merge pull request #172 from criticalbh/fix_open_port_dangling

fix issue when port is left dangling and catch poolboy timeout errors

lib/faktory_worker.ex

@@ -24,6 +24,7 @@ defmodule FaktoryWorker do
   For a full list of configuration options see the [Configuration](configuration.html) documentation.
   """
 
+  require Logger
   alias FaktoryWorker.{ConnectionManager, Pool}
 
   @default_timeout 5_000
@@ -88,14 +89,20 @@ defmodule FaktoryWorker do
   - `timeout` how long to wait for a response, in ms (default: #{@default_timeout})
 
   """
-  @spec send_command(command(), [send_command_opt()]) :: FaktoryWorker.Connection.response()
+  @spec send_command(command(), [send_command_opt()]) :: FaktoryWorker.Connection.response() | {:error, :timeout}
   def send_command(command, opts \\ []) do
-    opts
-    |> Keyword.get(:faktory_name, __MODULE__)
-    |> Pool.format_pool_name()
-    |> :poolboy.transaction(
-      &ConnectionManager.Server.send_command(&1, command),
-      Keyword.get(opts, :timeout, @default_timeout)
-    )
+    try do
+      opts
+      |> Keyword.get(:faktory_name, __MODULE__)
+      |> Pool.format_pool_name()
+      |> :poolboy.transaction(
+        &ConnectionManager.Server.send_command(&1, command),
+        Keyword.get(opts, :timeout, @default_timeout)
+      )
+    catch
+      :exit, error ->
+        Logger.error(inspect(error))
+        {:error, :timeout}
+    end
   end
 end

lib/faktory_worker/connection.ex

@@ -26,6 +26,11 @@ defmodule FaktoryWorker.Connection do
     end
   end
 
+  @spec close(conn :: __MODULE__.t()) :: :ok
+  def close(%{socket_handler: socket_handler} = conn) do
+    socket_handler.close(conn)
+  end
+
   @spec send_command(connection :: __MODULE__.t(), Protocol.protocol_command()) :: response()
   def send_command(%{socket_handler: socket_handler} = connection, :end) do
     with {:ok, payload} <- Protocol.encode_command(:end),
@@ -74,9 +79,7 @@ defmodule FaktoryWorker.Connection do
 
   defp send_handshake({:ok, %{"v" => version}}, _, _) when version != @faktory_version do
     {:error,
-     "Only Faktory version '#{@faktory_version}' is supported (connected to Faktory version '#{
-       version
-     }')."}
+     "Only Faktory version '#{@faktory_version}' is supported (connected to Faktory version '#{version}')."}
   end
 
   defp send_handshake({:ok, response}, connection, opts) do

lib/faktory_worker/connection_manager.ex

@@ -33,6 +33,8 @@ defmodule FaktoryWorker.ConnectionManager do
   def send_command(state, command, allow_retry \\ true) do
     case try_send_command(state, command) do
       {{:error, reason}, _} when reason in @connection_errors ->
+        # Close dangling port
+        close_connection(state)
         error = {:error, "Failed to connect to Faktory"}
         state = %{state | conn: nil}
 
@@ -78,6 +80,10 @@ defmodule FaktoryWorker.ConnectionManager do
     end
   end
 
+  defp close_connection(%{conn: conn}) do
+    Connection.close(conn)
+  end
+
   defp log_error(reason, {_, %{jid: jid}}) do
     Logger.warn("[#{jid}] #{reason}")
   end

lib/faktory_worker/socket.ex

@@ -13,4 +13,6 @@ defmodule FaktoryWorker.Socket do
 
   @callback recv(connection :: Connection.t(), length :: pos_integer()) ::
               {:ok, String.t()} | {:error, term()}
+
+  @callback close(connection :: Connection.t()) :: :ok
 end

lib/faktory_worker/socket/ssl.ex

@@ -31,6 +31,11 @@ defmodule FaktoryWorker.Socket.Ssl do
     result
   end
 
+  @impl true
+  def close(%{socket: socket}) do
+    :ssl.close(socket)
+  end
+
   defp try_connect(host, port, opts) do
     host = String.to_charlist(host)
     tls_verify = Keyword.get(opts, :tls_verify, true)

lib/faktory_worker/socket/tcp.ex

@@ -31,6 +31,11 @@ defmodule FaktoryWorker.Socket.Tcp do
     result
   end
 
+  @impl true
+  def close(%{socket: socket}) do
+    :gen_tcp.close(socket)
+  end
+
   defp try_connect(host, port) do
     host = String.to_charlist(host)
 

lib/faktory_worker/telemetry.ex

@@ -120,5 +120,4 @@ defmodule FaktoryWorker.Telemetry do
   defp log_error(outcome, jid, args, worker_module) do
     log_error("#{outcome} (#{worker_module}) jid-#{jid} #{inspect(args)}")
   end
-
 end