Meltwater replaced its message queueing tech stack in 2020 and are no longer using RabbitMQ. We have therefore decided to stop maintaining this libabry as it is no longer in use internally at Meltwater.
RxRabbit is a RabbitMQ java 8 client library that extends and enhances the rabbitmq-java-client by providing a resilient, auto-connecting ReactiveX styled API.
- Simple, reactive API based on RxJava that fits the RabbitMQ model.
- Automatic error handling and recovery that 'just works', both for publishing and consuming.
Have a look at the RxRabbit Tutorial to get a feel for how to use the API in a real application.
You can also look at the integration tests and the ExampleCode class.
The javadoc of the core API classes and interfaces is also a good source of reference.
Gradle:
compile 'com.meltwater:rxrabbit:$RXRABBIT_VERSION' Maven:
<dependency>
<groupId>com.meltwater</groupId>
<artifactId>rxrabbit</artifactId>
<version>$RXRABBIT_VERSION</version>
<type>jar</type>
</dependency>The API has strong opinions on how RabbitMQ should be used. It hides a lot of functionality from the Channel interface and also introduces a concept of channel types that are made for a specific purpose (such as publish, consume or admin operations). This means of course that there are several things that you can't do with this api, but keep in mind that it is a conscious decision made by the API developers.
Main supported use cases by rxrabbit:
- Continuously 'infinite' consume (with manual acknowledgment) from an already existing queue on the rabbit broker.
- Continuously 'infinite' consume (with manual acknowledgment) from a server created, temporary, exclusive queue bound to an existing exchange on the rabbit broker.
- Publish messages to an exchange with (or without) publisher confirmation but with mandatory=false and immediate=false.
- Perform basic 'admin' operations (declare, remove and purge queues, exchanges and bindings)
The official rabbitmq Java client provides some basic but non-complete error handling and recovery mechanisms. A number of wrapper libraries already exists today which, with varying success, automatically handles connection recovery in more error scenarios than the official java client. The best ones we have found so far were Spring-amqp which we discovered suffered from excessive and erroneous channel handling, and lyra which, in our view, implements and overly complex error handling logic that still has issues with recovering from some extreme error cases, such as hard broker restarts.
RxRabbit instead uses a very basic but effective approach when it comes to error handling and recovery:
No matter the error, the code assumes that the connection is broken then attempts to re-connect to the broker with exponential back-off
The goal we have is that consumers and producers should not have to care about the underlying rabbit channels and connections. Re-connects should remain hidden from the API user unless specifically asked for (currently achieved by adding listeners).
Pre-requisites
- JDK8
- docker (version 1.9 or later)
- docker-compose (version 1.6 or later)
Build (including running the tests) by running
./gradlew clean build
Also note that it is currently NOT supported to run the test on OSX using docker machine, you need to be able to connect to docker containers using localhost:.
We happily accept contributions in the form of Github PRs or in the form of bug reports, comments/suggestions or usage questions by creating a github issue.
This library was developed as an Meltwater internal project starting in October 2015. Over the past year it has been battle tested in production by an (ever increasing) majority of our jvm based components and is now an integral part of our technology stack.
The MIT License (MIT)
Copyright (c) 2016 Meltwater Inc. http://underthehood.meltwater.com/