From 10e0f262b31a8110c0d009a99f0d840c1003941e Mon Sep 17 00:00:00 2001
From: Noah Levitt <nlevitt@archive.org>
Date: Thu, 24 Apr 2014 20:31:40 -0700
Subject: [PATCH 1/2] ARI-3765 - make AMQPPublishProcessor gracefully handle
 amqp server going up and down

---
 .../archive/modules/AMQPPublishProcessor.java | 19 +++++++++++++++++--
 1 file changed, 17 insertions(+), 2 deletions(-)

diff --git a/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java b/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java
index 02935cb8..3646f399 100644
--- a/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java
+++ b/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java
@@ -24,6 +24,7 @@ import static org.archive.modules.CoreAttributeConstants.A_HERITABLE_KEYS;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -176,8 +177,12 @@ public class AMQPPublishProcessor extends Processor {
     }
 
     protected synchronized Channel channel() {
+        if (threadChannel.get() != null && !threadChannel.get().isOpen()) {
+            threadChannel.set(null);
+        }
+
         if (threadChannel.get() == null) {
-            if (connection == null) {
+            if (connection == null || !connection.isOpen()) {
                 connect();
             }
             try {
@@ -190,14 +195,24 @@ public class AMQPPublishProcessor extends Processor {
         }
         return threadChannel.get();
     }
+    
+    private AtomicBoolean serverLooksDown = new AtomicBoolean(false);
 
     private synchronized void connect() {
         ConnectionFactory factory = new ConnectionFactory();
         try {
             factory.setUri(getAmqpUri());
             connection =  factory.newConnection();
+            boolean wasDown = serverLooksDown.getAndSet(false);
+            if (wasDown) {
+                logger.info(getAmqpUri() + " is back up, connected successfully!");
+            }
         } catch (Exception e) {
-            logger.log(Level.SEVERE, "Attempting to connect to AMQP server failed!", e);
+            connection = null;
+            boolean wasAlreadyDown = serverLooksDown.getAndSet(true);
+            if (!wasAlreadyDown) {
+                logger.log(Level.SEVERE, "Attempting to connect to AMQP server failed!", e);
+            }
         }
     }
 

From 381442dfb60ac2157795177555529c97d4b84136 Mon Sep 17 00:00:00 2001
From: Noah Levitt <nlevitt@archive.org>
Date: Thu, 24 Apr 2014 21:23:03 -0700
Subject: [PATCH 2/2] ARI-3765 - make AMQPUrlReceiver gracefully handle amqp
 server going up and down

---
 .../crawler/frontier/AMQPUrlReceiver.java     | 72 ++++++++++++-------
 1 file changed, 47 insertions(+), 25 deletions(-)

diff --git a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
index c1c37f07..4074a842 100644
--- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
+++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
@@ -51,6 +51,7 @@ import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.Consumer;
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
+import com.rabbitmq.client.ShutdownSignalException;
 
 /**
  * @contributor nlevitt
@@ -104,32 +105,43 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
         return isRunning;
     }
 
-    @Override
-    synchronized public void start() {
-        // spawn off a thread for to start up the amqp consumer, otherwise the crawl launch thread can get stuck here 
-        if (!isRunning) {
-            new Thread(AMQPUrlReceiver.class.getSimpleName() + "-start") {
-                @Override
-                public void run() {
-                    synchronized (AMQPUrlReceiver.this) {
-                        while (!isRunning) {
-                            try {
-                                Consumer consumer = new UrlConsumer(channel());
-                                channel.queueDeclare(getQueueName(), false, false, true, null);
-                                channel().queueBind(getQueueName(), exchange, getQueueName());
-                                channel().basicConsume(getQueueName(), false, consumer);
-                                isRunning = true;
-                            } catch (IOException e) {
-                                logger.log(Level.SEVERE, "problem starting AMQP consumer (will try again after 30 seconds)", e);
-                                try {
-                                    Thread.sleep(30000);
-                                } catch (InterruptedException e1) {
-                                }
-                            }
+    private class StarterRestarter extends Thread {
+        public StarterRestarter(String name) {
+            super(name);
+        }
+
+        @Override
+        public void run() {
+            while (true) {
+                synchronized (AMQPUrlReceiver.this) {
+                    if (!isRunning) {
+                        try {
+                            Consumer consumer = new UrlConsumer(channel());
+                            channel.queueDeclare(getQueueName(), false, false, true, null);
+                            channel().queueBind(getQueueName(), getExchange(), getQueueName());
+                            channel().basicConsume(getQueueName(), false, consumer);
+                            isRunning = true;
+                            logger.info("started AMQP consumer uri=" + getAmqpUri() + " exchange=" + getExchange() + " queueName=" + getQueueName());
+                        } catch (IOException e) {
+                            logger.log(Level.SEVERE, "problem starting AMQP consumer (will try again after 30 seconds)", e);
                         }
                     }
                 }
-            }.start();
+
+                try {
+                    Thread.sleep(30000);
+                } catch (InterruptedException e1) {
+                }
+            }
+        }
+    }
+
+    @Override
+    synchronized public void start() {
+        // spawn off a thread to start up the amqp consumer, and try to restart it if it dies 
+        if (!isRunning) {
+            StarterRestarter t = new StarterRestarter(AMQPUrlReceiver.class.getSimpleName() + "-starter-restarter");
+            t.start();
         }
     }
 
@@ -143,6 +155,9 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
                 logger.log(Level.SEVERE, "problem closing AMQP connection", e);
             }
         }
+        connection = null;
+        channel = null;
+        isRunning = false;
     }
 
     transient protected Connection connection = null;
@@ -226,6 +241,13 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
 
             this.getChannel().basicAck(envelope.getDeliveryTag(), false);
         }
+        
+        @Override
+        public void handleShutdownSignal(String consumerTag,
+                ShutdownSignalException sig) {
+            logger.log(Level.SEVERE, "amqp channel/connection shut down consumerTag=" + consumerTag, sig);
+            isRunning = false;
+        }
 
         // {
         //  "headers": {
@@ -295,7 +317,7 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
     public void onApplicationEvent(CrawlStateEvent event) {
         switch(event.getState()) {
         case PAUSING: case PAUSED:
-            if (channel != null) {
+            if (channel != null && channel.isOpen()) {
                 try {
                     channel.flow(false);
                 } catch (IOException e) {
@@ -305,7 +327,7 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
             break;
 
         case RUNNING: case EMPTY: case PREPARING:
-            if (channel != null) {
+            if (channel != null && channel.isOpen()) {
                 try {
                     channel.flow(true);
                 } catch (IOException e) {