Improve error handling when server goes down

Rearranged children order so consumers don't go down due to rest_for_one strategy.
inside-track · Aug 4, 2020 · 877b952 · 877b952
1 parent ce729a5
commit 877b952
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 9 deletions.
diff --git a/lib/itk/queue.ex b/lib/itk/queue.ex
@@ -38,10 +38,7 @@ defmodule ITKQueue do
     [
       ITKQueue.ConnectionPool,
       ITKQueue.PublisherPool,
-      %{
-        id: ITKQueue.RetryPublisher,
-        start: {ITKQueue.RetryPublisher, :start_link, [[], ITKQueue.RetryPublisher]}
-      },
+      ITKQueue.ConsumerSupervisor,
       %{
         id: ITKQueue.ConsumerConnection,
         start:
@@ -51,8 +48,11 @@ defmodule ITKQueue do
              ITKQueue.ConsumerConnection
            ]}
       },
-      ITKQueue.ConsumerSupervisor,
-      ITKQueue.Workers
+      ITKQueue.Workers,
+      %{
+        id: ITKQueue.RetryPublisher,
+        start: {ITKQueue.RetryPublisher, :start_link, [[], ITKQueue.RetryPublisher]}
+      }
     ]
   end
 

diff --git a/lib/itk/queue/consumer.ex b/lib/itk/queue/consumer.ex
@@ -93,7 +93,7 @@ defmodule ITKQueue.Consumer do
       connection -> {:ok, connection}
     end
   catch
-    _ ->
+    _, _ ->
       {:error, :connection_lost}
   end
 
@@ -107,8 +107,8 @@ defmodule ITKQueue.Consumer do
 
     {:ok, _} = AMQP.Basic.consume(channel, queue_name, self())
     {:ok, channel}
-  rescue
-    e -> e
+  catch
+    _, e -> e
   end
 
   defp subscribe(subscription = %Subscription{queue_name: queue_name, routing_key: routing_key}) do

diff --git a/lib/itk/queue/pub_channel.ex b/lib/itk/queue/pub_channel.ex
@@ -49,6 +49,10 @@ defmodule ITKQueue.PubChannel do
       Logger.info("Publisher channel #{inspect(self())} opened on conn #{inspect(conn)}")
       {:noreply, %{state | chan: chan, status: :connected}}
     end)
+  catch
+    _, e ->
+      Process.send_after(self(), :connect, @reconnect_interval)
+      {:noreply, state}
   end
 
   def handle_info(:confirm, state = %{status: :connected, chan: chan, pending: pending}) do

diff --git a/lib/itk/queue/retry_publisher.ex b/lib/itk/queue/retry_publisher.ex
@@ -44,6 +44,11 @@ defmodule ITKQueue.RetryPublisher do
       Logger.info("RetryPublisher channel #{inspect(self())} opened on conn #{inspect(conn)}")
       {:noreply, %{state | chan: chan, status: :connected}}
     end)
+  catch
+    _, e ->
+      Logger.warn("RetryPublisher connect error: #{inspect(e)}")
+      Process.send_after(self(), :connect, @reconnect_interval)
+      {:noreply, state}
   end
 
   def handle_info(:confirm, state = %{status: :connected, chan: chan}) do