Reconnect on send for tcp

If the listening socket is closed when :tcp_closed is received, then try
to reconnect again on each send.

Author: KronicDeth

lib/logger_logstash_backend.ex

@@ -63,8 +63,8 @@ defmodule LoggerLogstashBackend do
   end
 
   def handle_info({:tcp_closed, socket}, state = %__MODULE__{host: host, port: port, socket: socket}) do
-    with {:error, _} <- configure_socket(state, host, port) do
-      {:ok, state}
+    with {:error, reason} <- configure_socket(state, host, port) do
+      {:ok, %{state | socket: nil}}
     end
   end
 
@@ -103,15 +103,7 @@ defmodule LoggerLogstashBackend do
       fields: fields
     }
 
-    with {:error, reason} <- Socket.send(state, [json, "\n"]) do
-      # fallback in case TCP configuration is bad
-      IO.puts :stderr,
-              "Could not log message (#{json}) to socket (#{url(state)}) due to #{inspect reason}.  " <>
-              "Check that state (#{inspect state}) is correct."
-    end
-
-
-    {:ok, state}
+    send_with_retry(state, json)
   end
 
   defp configure(name, opts) when is_atom(name) and is_list(opts) do
@@ -131,15 +123,22 @@ defmodule LoggerLogstashBackend do
     configure_socket(state, host, port)
   end
 
-  defp configure_socket(state, host, port) when is_nil(host) or is_nil(port), do: {:ok, state}
-  defp configure_socket(state, host, port) do
+  defp configure_socket(state = %__MODULE__{host: host, port: port}) do
     with reply = {:error, reason} <- Socket.open %{state | host: to_char_list(host), port: port} do
       IO.puts :stderr, "Could not open socket (#{url(state)}) due to reason #{inspect reason}"
 
       reply
     end
   end
 
+  defp configure_socket(state, host, port) when is_nil(host) or is_nil(port), do: {:ok, state}
+  defp configure_socket(state, host, port), do: configure_socket %{state | host: to_char_list(host), port: port}
+
+  defp fallback_log(state, json, reason) do
+    IO.puts :stderr,
+            "Could not log message (#{json}) to socket (#{url(state)}) due to #{inspect reason}.  " <>
+            "Check that state (#{inspect state}) is correct."
+  end
 
   # inspects the argument only if it is a pid
   defp inspect_pid(pid) when is_pid(pid), do: inspect(pid)
@@ -152,6 +151,31 @@ defmodule LoggerLogstashBackend do
     end
   end
 
+  defp send_with_retry(state = %__MODULE__{socket: nil}, json), do: send_with_new_socket(state, json)
+  defp send_with_retry(state, json) do
+    case Socket.send(state, [json, "\n"]) do
+      :ok ->
+        {:ok, state}
+      {:error, :closed} ->
+        send_with_new_socket(state, json)
+      {:error, reason} ->
+        fallback_log(state, json, reason)
+
+        {:ok, state}
+    end
+  end
+
+  defp send_with_new_socket(state, json) do
+    case configure_socket(state) do
+      {:error, reason} ->
+        fallback_log(state, json, reason)
+
+        {:ok, state}
+      {:ok, new_state} ->
+        send_with_retry(new_state, json)
+    end
+  end
+
   defp url(%__MODULE__{host: host, port: port, protocol: protocol}) do
     "#{protocol}://#{host}:#{port}"
   end

test/logger_logstash_backend/tcp_test.exs

@@ -19,12 +19,21 @@ defmodule LoggerLogstashBackend.TCPTest do
   use Timex
 
   import ExUnit.CaptureIO
+  import ExUnit.CaptureLog
+
+  # Functions
+
+  def accept(listen_socket), do: :gen_tcp.accept(listen_socket, 1_000)
+
+  def listen(port \\ 0) do
+    :gen_tcp.listen port, [:binary, {:active, true}, {:ip, {127, 0, 0, 1}}, {:reuseaddr, true}]
+  end
 
   # Callbacks
 
   setup context = %{line: line} do
     # have to open socket before configure_backend, so that it is listening when connect happens
-    {:ok, listen_socket} = :gen_tcp.listen 0, [:binary, {:active, true}, {:ip, {127, 0, 0, 1}}, {:reuseaddr, true}]
+    {:ok, listen_socket} = listen
     {:ok, port} = :inet.port(listen_socket)
 
     backend = {LoggerLogstashBackend, String.to_atom("#{inspect __MODULE__}#{line}")}
@@ -41,7 +50,7 @@ defmodule LoggerLogstashBackend.TCPTest do
       protocol: :tcp
     ]
 
-    {:ok, accept_socket} = :gen_tcp.accept(listen_socket, 1_000)
+    {:ok, accept_socket} = accept(listen_socket)
 
     on_exit fn ->
       Logger.remove_backend backend
@@ -52,6 +61,7 @@ defmodule LoggerLogstashBackend.TCPTest do
     full_context = context
                    |> Map.put(:accept_socket, accept_socket)
                    |> Map.put(:listen_socket, listen_socket)
+                   |> Map.put(:port, port)
 
     {:ok, full_context}
   end
@@ -68,7 +78,7 @@ defmodule LoggerLogstashBackend.TCPTest do
     assert fields["function"] == "test can log/1"
     assert fields["key1"] == "field1"
     assert fields["level"] == "info"
-    assert fields["line"] == 60
+    assert fields["line"] == 70
     assert fields["module"] == to_string(__MODULE__)
     assert fields["pid"] == inspect(self)
     assert fields["some_metadata"] == "go here"
@@ -92,7 +102,7 @@ defmodule LoggerLogstashBackend.TCPTest do
     assert fields["function"] == "test can log pids/1"
     assert fields["pid_key"] == inspect(self)
     assert fields["level"] == "info"
-    assert fields["line"] == 84
+    assert fields["line"] == 94
     assert fields["module"] == to_string(__MODULE__)
     assert fields["pid"] == inspect(self)
     assert fields["some_metadata"] == "go here"
@@ -115,6 +125,42 @@ defmodule LoggerLogstashBackend.TCPTest do
     assert {:ok, _} = :gen_tcp.accept(listen_socket, 1_000)
   end
 
+  test "it reconnects on send if disconnected and listening socket is closed " <>
+       "when handle_info({:tcp_closed, _}, _) is called",
+       %{accept_socket: accept_socket, listen_socket: listen_socket, port: port} do
+    :ok = :gen_tcp.close listen_socket
+    :ok = :gen_tcp.close accept_socket
+
+    Logger.info "Can't connect"
+    :timer.sleep 100
+
+    {:ok, new_listen_socket} = listen(port)
+
+    # Need to accept and wait in a separate process because connect won't come until the Logger.info call tries to
+    # establish a new socket
+    pid = self
+    spawn_link fn ->
+      {:ok, new_accept_socket} = accept(new_listen_socket)
+
+      send pid, {:new_accept_socket, new_accept_socket}
+
+      receive do
+        # forward to pid so that get_log! works as normal
+        message = {:tcp, _, _} -> send pid, message
+      end
+    end
+
+    Logger.info "I reconnected"
+
+    assert_receive {:new_accept_socket, _}
+
+    :timer.sleep 100
+
+    {:ok, data} = JSX.decode get_log!
+
+    assert data["message"] == "I reconnected"
+  end
+
   test "if it can't reconnect, then is prints to :stderr",
        %{accept_socket: accept_socket, listen_socket: listen_socket} do
     :ok = :gen_tcp.close listen_socket