JarCastingQueueBox - async queue engine based on MongoDB
Async java message queue using MongoDB as a backend.
This fork use the latest MongoDB version with the latest Java Driver and contains wrapper that hide MongoDB driver API and allow using plain java objects in queue.
Unit tests based on real MongoDB with testcontainers.
issues and feature requests
- security issues - hackerone
- other issues and features - github issues
Features
- totally async and non-blocking multithreading (maybe used in clustered software)
- Message selection and/or count via MongoDB query
- Distributes across machines via MongoDB
- Message priority
- Delayed messages
- Message routing
- Running message timeout and redeliver
- Atomic acknowledge and send together
- Easy index creation based only on payload
- work with the latest MongoDB (4.2)
- you can use any other storage system by implementing interface
Jar
To add the library as a jar simply Build the project and use the queue-box-0.0.1.jar from the created target directory!
Maven
To add the library as a local, per-project dependency use Maven! Simply add a dependency on to your project's pom.xml file such as:
<dependency>
<groupId>net.c0f3.labs</groupId>
<artifactId>queue-box</artifactId>
<version>0.2.1</version>
</dependency>
Usage example
- starting QueueBox instance
- creating listener for specific "destination"
- creating and sending some simple message presented as POJO
public final class Main {
public static void main(String[] args) throws InterruptedException, IOException {
final String defaultSource = "just_source";
final String defaultDestination = "just_destination";
Properties properties = new Properties();
properties.load(ExampleWithMain.class.getResourceAsStream("mongodb.properties"));
MongoRoutedQueueBox<JustPojoRouted> queueBox = new MongoRoutedQueueBox<>(
properties,
JustPojoRouted.class
);
queueBox.start(); // init internal thread pool ant begin periodic query to db
final JustPojoRouted pojo = new JustPojoRouted(13, "string message for 13");
pojo.setSource(defaultSource);
pojo.setDestination(defaultDestination);
queueBox.subscribe(new QueueConsumer<JustPojoRouted>() {
@Override
public void onPacket(MessageContainer<JustPojoRouted> message) {
JustPojoRouted recvPojo = message.getMessage();
System.out.println("received packet:"+recvPojo);
message.done(); // accepting message
}
@Override
public String getConsumerId() {
return defaultDestination; // destinations that this consumer accepts
}
});
Future future = queueBox.queue(pojo);
while (!future.isDone()) {
Thread.sleep(5);
}
System.out.println("send packet: "+pojo);
}
}
Also you can use just core library without wrapper, as it described in original README.
public final class Main {
public static void main(final String[] args) throws UnknownHostException {
final Queue queue = new Queue(new MongoClient().getDB("testing").getCollection("messages"));
queue.send(new BasicDBObject());
final BasicDBObject message = queue.get(new BasicDBObject(), 60);
queue.ack(message);
}
}
Documentation
Found in the source itself, take a look!
Project Build
For testing install docker.
mvn clean install
We must know our heroes!
This version based on the original version authored by Gaillard from here and impoved by Uromahn
