lee/heritrix3
1
0
Fork 0
mirror of https://github.com/internetarchive/heritrix3.git synced 2026-09-25 23:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #62 from nlevitt/ari-3765

Browse Source 
Ari 3765 gracefully handle amqp server going up and down
This commit is contained in:
vonrosen
2014-04-28 09:39:37 -07:00
parent 50a38e1376 381442dfb6
commit 747bbeba97
2 changed files with 64 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
+47 -25
View File
@@ -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) {
contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java
+17 -2
View File
@@ -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);
}
}
}