Storm High Availability and Failover - Real-time Analytics with Storm and Cassandra - page 99

at least 1");
}
this.prefetchCount = prefetchCount.intValue();
try {
this.collector = collector;
setupAMQP();
} catch (IOException e) {
log.error("AMQP setup failed", e);
log.warn("AMQP setup failed, will attempt to
reconnect...");
Utils.sleep(WAIT_AFTER_SHUTDOWN_SIGNAL);
reconnect();
}
}
/**
* Reconnect to an AMQP broker.in case the connection
breaks at some point
*/
private void reconnect() {
log.info("Reconnecting to AMQP broker...");
try {
setupAMQP();
} catch (IOException e) {
log.warn("Failed to reconnect to AMQP broker", e);
}
}
/**
* Set up a connection with an AMQP broker.
* @throws IOException
*This is the method where we actually connect to the
queue using AMQP client APIs
*/
private void setupAMQP() throws IOException{
final int prefetchCount = this.prefetchCount;
final ConnectionFactory connectionFactory = new
ConnectionFactory() {
public void configureSocket(Socket socket)