From 82bef2a3372e1e394b592d80a4e973368b858e3e Mon Sep 17 00:00:00 2001
From: Noah Levitt <nlevitt@archive.org>
Date: Thu, 6 Nov 2014 19:33:35 -0800
Subject: [PATCH] only release the lock if we actually own it (avoid
 java.lang.IllegalMonitorStateException at
 java.util.concurrent.locks.ReentrantLock.tryRelease(ReentrantLock.java:155))

---
 .../crawler/frontier/AMQPUrlReceiver.java     | 32 ++++++++++---------
 1 file changed, 17 insertions(+), 15 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 faec6615..46ee18e6 100644
--- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
+++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
@@ -119,25 +119,27 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
             while (!Thread.interrupted()) {
                 try {
                     lock.lockInterruptibly();
-                    if (!isRunning) {
-                        // start up again
-                        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);
+                    try {
+                        if (!isRunning) {
+                            // start up again
+                            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);
+                            }
                         }
-                    }
 
-                    Thread.sleep(30000);
+                        Thread.sleep(30000);
+                    } finally {
+                        lock.unlock();
+                    }
                 } catch (InterruptedException e) {
                     return;
-                } finally {
-                    lock.unlock();
                 }
             }
         }