Tag: rabittmq

RabittMQ batch message processing

There are times when we want to fire set of jobs and be notified when all of them complete. This can be easily accomplished with the latest version of Hoplin.io RabbitMQ client.

A use case for using batch messages would be partitioning a document and processing each partition via separate client.

As always we need our Publisher and Subscriber we will start with publisher first.

We start by creating a new client and then enqueuing number of jobs to process, upon completion we display the time it took to complete all jobs. Client will attempt to use Direct-Reply queue if available.

   public static void main(final String... args) throws IOException, InterruptedException
    {
        final BatchClient client = new DefaultBatchClient(options(), bind());

        client.startNew(context ->
        {
            for(int i = 0; i < 1000; ++i)
            {
                context.enque(() -> new LogDetail("Msg >> " + System.nanoTime(), "info"));
                context.enque(() -> new LogDetail("Msg >> " + System.nanoTime(), "warn"));
            }
        })
        .whenComplete((context, throwable)->
        {
                log.info("Batch completed in : {}", context.duration());
        });

        Thread.currentThread().join();
    }

    private static Binding bind()
    {
        return BindingBuilder
                .bind("batch.documents")
                .to(new DirectExchange("exchange.batch"))
                .build()
                ;
    }

The subscriber is little bit more involved, this is the part of API that needs to be simplified.


/**
 * Batch Job receiver
 */
public class ReceiveBatchJob extends BaseExample
{
    private static final Logger log = LoggerFactory.getLogger(ReceiveBatchJob.class);

    private static final String EXCHANGE = "exchange.batch";

    private static RabbitMQClient mqClient;

    public static void main(final String... args) throws InterruptedException
    {
        final ExchangeClient client = DirectExchangeClient.create(options(), EXCHANGE);
        mqClient = client.getMqClient();

        client.subscribe("test", LogDetail.class, ReceiveBatchJob::handle);
        Thread.currentThread().join();
    }

    private static void handle(final LogDetail msg, final MessageContext context)
    {
        final AMQP.BasicProperties properties = context.getProperties();
        final String replyTo = properties.getReplyTo();
        final String correlationId = properties.getCorrelationId();
        final Map<String, Object> headers = properties.getHeaders();
        final Object batchId = headers.get("x-batch-id");
        headers.put("x-batch-correlationId", correlationId);

        log.info("Incoming context        >  {}", context);
        log.info("Incoming replyTo        >  {}", replyTo);
        log.info("Incoming msg            >  {}", msg);
        log.info("Incoming correlationId  >  {}", correlationId);
        log.info("Incoming batchId        >  {}", batchId);

        final LogDetail reply = new LogDetail("Reply Message", "WARN");
        mqClient.basicPublish("", replyTo, reply, headers);
    }
}

There at two important properties that need to be looked at x-batch-id and x-batch-correlationId, currently they need to be copied directly form message and then reply need to be published via the underlying client.

Not really best API design at the moment as it mixes concerns and exposed the underlying RattitMQ client but the initial release will try to address this and simplify usage.

January 24, 2019

RabittMQ RPC Request/Response example

RabittMQ RPC Request/Response example using hoplin.io library

Following example creates RPC client and then setups Async response handler, which follows by the request to get processed.

Hoplin client supports both Direct-Reply and Queue per Request/Response patterns.

RpcClient<LogDetailRequest, LogDetailResponse> client = DefaultRpcClient.create(options(), bind());

// rpc response
client.respondAsync((request)->
{
	final LogDetailResponse response = new LogDetailResponse("Response message", "info");
	return response;
});


// rpc request
final LogDetailResponse response = client.request(new LogDetailRequest("Request message", "info"));
log.info("RPC response : {} ", response);

This is the binding that is used to create our client.

  private static Binding bind()
    {
        return BindingBuilder
                .bind("rpc.request.log")
                .to(new FanoutExchange("rpc.logs"));
    }

November 28, 2018

hoplin.io A lightweight RabbitMQ client for Java (built on top of rabittmq java client)

A lightweight RabbitMQ client for Java (built on top of rabittmq java client)

Documentation and project available at GitHub repo
https://github.com/gregbugaj/hoplin.io

To make working with RabbitMQ as simple as possible with minimum dependencies.

Minimal dependencies, simple configuration and API.

Subscriber client
Publisher client
Async RPC Client

Creating simple RabbitMQ client can be done in couple different ways.

The simplest way with minimal configuration

ExchangeClient client = ExchangeClient.topic(RabbitMQOptions.from("host=localhost"))

This creates new Exchange client bound to a Topic exchange.

We can also specify which queue and which routing key we want to handle.

final RabbitMQOptions options = RabbitMQOptions.from("host=localhost");
final ExchangeClient client = ExchangeClient.topic(options, "my.exchange", "log.critical", "log.critical.*")

For complete control we can use the Exchange to Queue Binding builder.

ExchangeClient clientFromBinding(final String exchange, final String queue, final String routingKey)
    {
        final Binding binding = BindingBuilder
                .bind(queue)
                .to(new TopicExchange(exchange))
                .withAutoAck(true)
                .withPrefetchCount(1)
                .withPublisherConfirms(true)
                .with(routingKey)
                .build();

        return ExchangeClient.topic(options(), binding);
    }

This is the most flexible method as it allows us to control all the aspect of how messages are handled.

Publishing and receiving messages is simple as well. Both methods provide number of overloaded methods to provide different levels of flexibility.

// Publish message
client.publish(new LogDetail("Msg : " + System.nanoTime()));

// Consume message
 client.subscribe(LogDetail.class, msg-> log.info("Message received [{}]", msg));

Here is example that includes both the Publisher and Subscriber

public class SamePublisherConsumerExample extends BaseExample
{
    private static final Logger log = LoggerFactory.getLogger(SamePublisherConsumerExample.class);

    private static final String EXCHANGE = "topic_logs";

    public static void main(final String... args) throws InterruptedException
    {
        log.info("Starting producer/consumer for exchange : {}", EXCHANGE);
        final ExchangeClient client = ExchangeClient.topic(options(), EXCHANGE);
        client.subscribe(LogDetail.class, SamePublisherConsumerExample::handle);

        for(int i = 0; i < 5; ++i)
        {
            client.publish(createMessage("info"), "log.info.info");
            client.publish(createMessage("debug"), "log.info.debug");

            Thread.sleep(1000L);
        }
    }

    private static void handle(final LogDetail msg)
    {
        log.info("Incoming msg : {}", msg);
    }

    private static LogDetail createMessage(final String level)
    {
      return new LogDetail("Msg : " + System.nanoTime(), level);
    }

}

October 29, 2018