Stability improvements

lib/lapin/connection.ex

@@ -358,15 +358,15 @@ defmodule Lapin.Connection do
 
     with configuration <- Keyword.merge(@connection_default_params, configuration),
          {:ok, connection} <- AMQP.Connection.open(configuration),
-         _ref = Process.monitor(connection.pid),
          {:ok, config_channel} <- AMQP.Channel.open(connection),
          {:ok, exchanges} <- declare_exchanges(configuration, config_channel),
          {:ok, queues} <- declare_queues(configuration, config_channel),
          :ok <- bind_exchanges(exchanges, config_channel),
          :ok <- bind_queues(queues, config_channel),
          {:ok, producers} <- create_producers(configuration, connection),
          {:ok, consumers} <- create_consumers(configuration, connection),
-         :ok <- AMQP.Channel.close(config_channel) do
+         :ok <- AMQP.Channel.close(config_channel),
+         :ok <- monitor_processes(connection, producers, consumers) do
       {
         :next_state,
         :connected,
@@ -443,6 +443,16 @@ defmodule Lapin.Connection do
         {:error, reason} ->
           Logger.debug("Failed rejecting message #{delivery_tag}: #{inspect(reason)}")
       end
+  catch
+    :exit, reason ->
+      case Consumer.reject_message(consumer, delivery_tag, not redelivered) do
+        :ok ->
+          Logger.error("Rejected message #{delivery_tag} due to process exit: #{inspect(reason)}")
+          :ok
+
+        {:error, reject_error} ->
+          Logger.debug("Failed rejecting message #{delivery_tag}: #{inspect(reject_error)}")
+      end
   end
 
   defp consume_ack(true = _consumer_ack, consumer, delivery_tag) do
@@ -612,4 +622,18 @@ defmodule Lapin.Connection do
       {:error, :missing_params, Enum.reject(params, &Keyword.has_key?(configuration, &1))}
     end
   end
+
+  defp monitor_processes(connection, producers, consumers) do
+    Process.monitor(connection.pid)
+
+    Enum.each(producers, fn producer ->
+      Process.monitor(producer.channel.pid)
+    end)
+
+    Enum.each(consumers, fn consumer ->
+      Process.monitor(consumer.channel.pid)
+    end)
+
+    :ok
+  end
 end