Listener for NATS JetStream

user_3212324 0
Viewed 269

Can some one help how to configure NATS jet stream subscription in spring boot asynchronously example: looking for an equivalent annotation like @kafkalistener for Nats jetstream

I am able to pull the messages using endpoint but however when tried to pull messages using pushSubscription dispatcherhandler is not invoked. Need to know how to make the listener to be active and consume messages immediately once the messages are published to the subject.

Any insights /examples regarding this will be helpful, thanks in advance.

nats-streaming-server nats.io jetstream spring-boot
1 Answers

I don't know what is your JetStream retention policy, neither the way you want to subscribe. But I have sample code for WorkQueuePolicy push subscription, wish this will help you.

public static void subscribe(String streamName, String subjectKey,
                             String queueName, IMessageHandler iMessageHandler) throws IOException,
        InterruptedException, JetStreamApiException {
    long s = System.currentTimeMillis();
    Connection nc = Nats.connect(options);
    long e = System.currentTimeMillis();
    logger.info("Nats Connect in " + (e - s) + " ms");
    JetStream js = nc.jetStream();
    Dispatcher disp = nc.createDispatcher();
    MessageHandler handler = (msg) -> {
        try {
            iMessageHandler.onMessageReceived(msg);
        } catch (Exception exc) {
            msg.nak();
        }
    };
    ConsumerConfiguration cc = ConsumerConfiguration.builder()
            .durable(queueName)
            .deliverGroup(queueName)
            .maxDeliver(3)
            .ackWait(Duration.ofMinutes(2))
            .build();
    PushSubscribeOptions so = PushSubscribeOptions.builder()
            .stream(streamName)
            .configuration(cc)
            .build();
    js.subscribe(subjectKey, disp, handler, false, so);
    System.out.println("NatsUtil: " + durableName + "subscribe");
}

IMessageHandler is my custom interface to handle nats.io received messages.

user_50057560
Related