in-spring-zeromq

Spring Boot enabled ZeromMQ integrations inspired by spring-rabbit and built on jzmq-api.

Getting Started

This project is built on Spring Boot. Add this project as a dependency to your Spring Boot application to get started.

Maven:

<dependency>
<groupId>io.insource</groupId>
<artifactId>in-spring-zeromq</artifactId>
<version>0.1.1</version>
</dependency> 

Gradle:

implementation("io.insource:in-spring-zeromq:0.1.1")

Publisher

To use the publisher, add the @EnableZmqPublisher annotation to your Spring Boot application:

@SpringBootApplication@EnableZmqPublisherpublicclassMyApplication {
publicstaticvoidmain(String[] args) {
SpringApplication.run(MyApplication.class, args);
}
}

To publish messages, use a ZmqTemplate. For example:

ZmqTemplatezmqTemplate = newZmqTemplate();
zmqTemplate.setTopic("events"); // Use topic "events"zmqTemplate.setRoutingKey("MyEvent"); // Use routing key "MyEvent" by defaultzmqTemplate.setMessageConverter(newVerySimpleMessageConverter()); // Use custom MessageConverter// Send using defaultszmqTemplate.send("This is a test.");

This publishes a message with the string This is a test. to the events topic using a routing key of MyEvent. Other flavors of send() look like this:

// Send using custom routing keyzmqTemplate.send("MyRoutingKey", "This is another test.");
// Send using custom routing key and headersMap<String, String> headers = newHashMap<>();
headers.put("my-header", "TEST");
zmqTemplate.send("MyRoutingKey", headers, "This is yet another test.");

Serialization to and from bytes is done using a MessageConverter. Here is the definition of a very simple MessageConverter:

publicclassVerySimpleMessageConverterimplementsMessageConverter {
@OverridepublicMessagetoMessage(Objectobj, Map<String, String> headers) {
returnnewMessage(obj.toString());
}
@OverridepublicObjectfromMessage(Messagemessage) {
returnmessage.popString();
}
}

Note: This example does not consider headers. The bundled class SimpleMessageConverter does consider headers, which changes the data format over the wire considerably. Both sides must use the same MessageConverter or risk incompatible (de)serialization.

Messages sent on any thread within the JVM will be sent to a local proxy to be forwarded to a remote host. To configure the proxy to forward messages for this topic, use the following configuration:

application.yml:

zmq:
publisher:
host: 10.0.0.123port: 1337topics:
- events

Subscriber

To use the subscriber, add the @EnableZmqSubscriber annotation to your Spring Boot application:

@SpringBootApplication@EnableZmqSubscriberpublicclassMyApplication {
publicstaticvoidmain(String[] args) {
SpringApplication.run(MyApplication.class, args);
}
}

To create a subscriber, add the @ZmqSubscriber annotation to a class that implements MessageListener:

importio.insource.framework.annotation.QueueBinding;
importio.insource.framework.annotation.ZmqSubscriber;
importio.insource.framework.zeromq.MessageListener;
importorg.zeromq.api.Message;
@ZmqSubscriber({
@QueueBinding(topic = "events", key = "MyEvent", queue = "my-events"),
@QueueBinding(topic = "events", key = "AnotherEvent", queue = "other")
})
publicclassMySubscriberimplementsMessageListener {
@OverridepublicvoidonMessage(Messagemessage) {
// Do something with messageSystem.out.println(message.popString());
}
}

This will create queues called my-events and other each with their own inbox. Each inbox can hold up to 1000 messages at a time. Additional messages will be discarded.

Both inboxes will be managed by a thread dedicated to dispatching messages to MySubscriber. Only events with routing key MyEvent and AnotherEvent on topic events will be delivered. Other topics and routing keys will be ignored.

Alternatively, you can register a MessageConverter in your Spring @Configuration to deserialize messages into Java objects:

importio.insource.framework.zeromq.MessageConverter;
importorg.springframework.context.annotation.Bean;
importorg.springframework.context.annotation.Configuration;
importorg.zeromq.api.Message;
@ConfigurationpublicclassMyConfiguration {
@BeanpublicMessageConvertermessageConverter() {
returnnewVerySimpleMessageConverter();
}
}

Then simply add a method to your @ZmqSubscriber with the @ZmqHandler annotation. Make sure it accepts exactly the type returned by your MessageConverter:

importio.insource.framework.annotation.QueueBinding;
importio.insource.framework.annotation.ZmqHandler;
importio.insource.framework.annotation.ZmqSubscriber;
@ZmqSubscriber({
@QueueBinding(topic = "events", key = "MyEvent", queue = "my-events"),
@QueueBinding(topic = "events", key = "AnotherEvent", queue = "other")
})
publicclassMySubscriber {
@ZmqHandlerpublicvoidonMyEvent(Stringmessage) {
// Do something with messageSystem.out.println(message);
}
}

Note: More flexible method signatures for @ZmqHandler will be added in a future release.

Listener

This library also includes a basic socket listener abstraction, for times when pub/sub breaks down, and raw 0MQ messaging is required. Instead of forcing the listener to adopt a different programming model, the @ZmqListener annotation can be used.

To use the listener, add the @EnableZmqListener annotation to your Spring Boot application:

@SpringBootApplication@EnableZmqListenerpublicclassMyApplication {
publicstaticvoidmain(String[] args) {
SpringApplication.run(MyApplication.class, args);
}
}

To create a listener, add the @ZmqListener annotation to a class that implements MessageListener:

importio.insource.framework.annotation.QueueBinding;
importio.insource.framework.annotation.ZmqSubscriber;
importio.insource.framework.zeromq.MessageListener;
importorg.zeromq.api.Message;
@ZmqListener(1337)
publicclassMyListenerimplementsMessageListener {
@OverridepublicvoidonMessage(Messagemessage) {
// Do something with messageSystem.out.println(message.popString());
}
}

This will create a PULL socket listening on port 1337 with its own inbox. The inbox can hold up to 1000 messages at a time. Additional messages will cause the sender to block.

Each inbox will be managed by a thread dedicated to dispatching messages to MyListener.

Alternatively, you can register a MessageConverter in your Spring @Configuration to deserialize messages into Java objects:

importio.insource.framework.zeromq.MessageConverter;
importorg.springframework.context.annotation.Bean;
importorg.springframework.context.annotation.Configuration;
importorg.zeromq.api.Message;
@ConfigurationpublicclassMyConfiguration {
@BeanpublicMessageConvertermessageConverter() {
returnnewVerySimpleMessageConverter();
}
}

Then simply add a method to your @ZmqListener with the @ZmqHandler annotation. Make sure it accepts exactly the type returned by your MessageConverter:

importio.insource.framework.annotation.QueueBinding;
importio.insource.framework.annotation.ZmqHandler;
importio.insource.framework.annotation.ZmqSubscriber;
@ZmqListener(1337)
publicclassMyListener {
@ZmqHandlerpublicvoidonMyEvent(Stringmessage) {
// Do something with messageSystem.out.println(message);
}
}

Contributing

To build this project, use the provided gradle build script.

> ./gradlew build