Skip to content

Instantly share code, notes, and snippets.

@ledoyen
Last active October 4, 2017 08:32
Show Gist options
  • Save ledoyen/1fdbdb10214f2d76a4fbbe57eef55668 to your computer and use it in GitHub Desktop.
Save ledoyen/1fdbdb10214f2d76a4fbbe57eef55668 to your computer and use it in GitHub Desktop.
Reactive AMQP in Java

Reactive AMQP in Java

Using this Maven configuration

<dependencies>
	<dependency>
		<groupId>io.projectreactor</groupId>
		<artifactId>reactor-core</artifactId>
		<version>3.0.6.RELEASE</version>
	</dependency>

	<dependency>
		<groupId>io.scalac</groupId>
		<artifactId>reactive-rabbit_2.10</artifactId>
		<version>1.0.3</version>
	</dependency>
</dependencies>

Scala reactive-rabbit library can be used in Java. Here is an example with Reactor :

import io.scalac.amqp.Connection;
import io.scalac.amqp.Delivery;
import reactor.core.publisher.Flux;

...
Function<Delivery, Message> deliveryToMessage = ...
Connection connection = io.scalac.amqp.Connection$.MODULE$.apply();
Flux<Message> messageFlux = Flux
  .from(connection.consume("lol-queue", connection.consume$default$2()))
	.map(deliveryToMessage);

messageFlux.subscribe(System.out::println);

As both reactive-rabbit and Reactor implement the Reactive Streams contract, interoperability is trivial.

The glue part left to code are:

  • use of Delivery to build a more Java-like object (deliveryToMessage function)
  • configure the client
    • using Typesafe Config mechanisms
    • mapping some Java-like object to io.scalac.amqp.ConnectionSettings.ConnectionSettings
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment