storm-amqp-spout: AMQP input source for Storm
storm-amqp-spout allows a Storm topology to consume an AMQP queue as an input source. It currently provides:
- AMQPSpout: an implementation of
backtype.storm.topology.IRichSpoutthat connects to an AMQP broker, consumes the messages routed to a specified AMQP queue and emits them as Storm tuples.
- QueueDeclaration: an interface that encapsulates declaring an AMQP queue and setting up any exchange bindings it requires, used by AMQPSpout to set up the queue to consume.
- ExclusiveQueueWithBinding: a QueueDeclaration suitable for prototyping and one-off analytics use cases.
- SharedQueueWithBinding: a QueueDeclaration suitable for production use cases needing guaranteed processing.
The Javadocs can be found at http://code.rapportive.com/storm-amqp-spout.
To produce a jar:
$ mvn package
To install in your local Maven repository:
$ mvn install
To use in your
<project> <!-- ... --> <dependencies> <!-- ... --> <dependency> <groupId>com.rapportive</groupId> <artifactId>storm-amqp-spout</artifactId> <version>0.0.3-SNAPSHOT</version> </dependency> <!-- ... --> </dependencies> <!-- ... --> </project>
This is very early software. It may break and the API is liable to change completely between releases. Pull requests, patches and bug reports are very welcome.
N.B. if you need to guarantee all messages are reliably processed, you should have AMQPSpout consume from a queue that is not set as 'exclusive' or 'auto-delete': otherwise if the spout task crashes or is restarted, the queue will be deleted and any messages in it lost, as will any messages published while the task remains down. See SharedQueueWithBinding to declare a shared queue that allows for guaranteed processing. (For prototyping, an ExclusiveQueueWithBinding may be simpler to manage.)
AMQPSpout has been tested with RabbitMQ 2.3.1 and 2.6.1. It should probably work with other versions and other AMQP brokers.