com.rapportive.storm.amqp

Class ExclusiveQueueWithBinding

java.lang.Object

com.rapportive.storm.amqp.ExclusiveQueueWithBinding

All Implemented Interfaces:

QueueDeclaration, Serializable

public class ExclusiveQueueWithBinding
extends Object
implements QueueDeclaration

Declares an exclusive, server-named queue and binds it to an existing exchange. This is probably the easiest way to start prototyping with an AMQPSpout: if your app already publishes to an exchange, you can just point this at the exchange and start consuming messages.

However N.B. this queue setup is not reliable, in that if the spout task crashes or restarts, messages published while the spout is down will be lost (because the spout creates the queue when it starts up, and the server deletes the queue when the spout closes).

It also cannot scale out to multiple parallel spout tasks. The semantics of an exclusive queue mean that each spout task would get its own queue bound to the exchange. That means each task would receive a copy of every message, so messages would get processed multiple times.

If you need guaranteed processing or a horizontally scalable spout, consider SharedQueueWithBinding.

See Also:

Serialized Form

Constructor Summary

Constructors

Constructor and Description
ExclusiveQueueWithBinding(String exchange, String routingKey) Create a declaration of an exclusive server-named queue bound to the specified exchange.

Method Summary

Modifier and Type	Method and Description
AMQP.Queue.DeclareOk	declare(Channel channel) Verifies the exchange exists, creates an exclusive, server-named queue and binds it to the exchange.
boolean	isParallelConsumable() Returns false as this queue is not safe for parallel consumers.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

ExclusiveQueueWithBinding

public ExclusiveQueueWithBinding(String exchange,
                                 String routingKey)

Create a declaration of an exclusive server-named queue bound to the specified exchange.

Parameters:

exchange - exchange to bind the queue to.

routingKey - routing key for the exchange binding. Use "#" to receive all messages published to the exchange.

Method Detail

declare

public AMQP.Queue.DeclareOk declare(Channel channel)
                             throws IOException

Verifies the exchange exists, creates an exclusive, server-named queue and binds it to the exchange.

Specified by:

declare in interface QueueDeclaration

Parameters:

channel - An open AMQP channel which can be used to send the declarations.

Returns:

the server's response to the successful queue declaration (you can use this to discover the name of the queue).

Throws:

IOException - if the exchange does not exist, or if the AMQP connection drops.

isParallelConsumable

public boolean isParallelConsumable()

Returns false as this queue is not safe for parallel consumers.

Specified by:

isParallelConsumable in interface QueueDeclaration

Returns:

true if safe for parallel consumers, otherwise false.