Cannot use JetStream.subscribe with MessageHandler in jnats 2.21.1

[nagesh-sys]

Hi,
I am trying to use the following method signature in my project:

jetStream.subscribe(subject, messageHandler, false, pushOptions);

I have set my dependency to:

<dependency>
  <groupId>io.nats</groupId>
  <artifactId>jnats</artifactId>
  <version>2.21.1</version>
</dependency>

But I get this compilation error:

Compilation failure
/nats/src/main/java/com/test/sample/nats/service/TestNatsJetStreamService.java:[141,41] no suitable method found for subscribe(java.lang.String,io.nats.client.MessageHandler,boolean,io.nats.client.PushSubscribeOptions)
[ERROR]     method io.nats.client.JetStream.subscribe(java.lang.String,io.nats.client.Dispatcher,io.nats.client.MessageHandler,boolean) is not applicable
[ERROR]       (argument mismatch; io.nats.client.MessageHandler cannot be converted to io.nats.client.Dispatcher)
[ERROR]     method io.nats.client.JetStream.subscribe(java.lang.String,io.nats.client.Dispatcher,io.nats.client.MessageHandler,io.nats.client.PullSubscribeOptions) is not applicable
[ERROR]       (argument mismatch; io.nats.client.MessageHandler cannot be converted to io.nats.client.Dispatcher)

if I use the subscribe method with dispatcher it will work below reference

subscription = jetStream.subscribe(subject, dispatcher, messageHandler, false, pushOptions);

Method code block is like below

public void subscribe() {
	    try {
	        LOG.error("Entered in Subscribe method");
	        String natsChannel = test-channel;
	        final String subject = test-channel;

	        var natsInstance = JetStreamConnection.getInstance();
	        JetStream jetStream = natsInstance.getJetStreamConnection();
	        JetStreamManagement jsm = natsInstance.getJetStreamManagement();

	        // Create Stream if it doesn't exist
	        StreamConfiguration streamConfig = StreamConfiguration.builder()
	                .name("mystream")
	                .subjects(subject)
	                .storageType(StorageType.File)
	                .retentionPolicy(RetentionPolicy.Limits)
	                .build();

	        try {
	            jsm.addStream(streamConfig);
	            LOG.info("JetStream stream created for subject: {}", subject);
	        } catch (JetStreamApiException e) {
	            if (e.getErrorCode() == 10058 || e.getMessage().contains("stream name already in use")) {
	                LOG.info("Stream already exists for subject: {}", subject);
	            } else {
	                LOG.error("Failed to create JetStream stream: {}", e.getMessage(), e);
	            }
	        }
	        var ackWaitTime = getAckWaitTime();
	        var maxInFlight = getMaxInFlight();

	        ConsumerConfiguration consumerConfig = ConsumerConfiguration.builder()
	                .durable(natsChannel)
	                .ackPolicy(AckPolicy.Explicit)
	                .ackWait(Duration.ofSeconds(ackWaitTime))
	                .maxAckPending(maxInFlight)
	                .deliverPolicy(DeliverPolicy.New)
	                .build();

	        PushSubscribeOptions pushOptions = PushSubscribeOptions.builder()
	                .durable(natsChannel)
	                .configuration(consumerConfig)
	                .build();

	        // Message Handler (no Dispatcher)
	        MessageHandler messageHandler = message -> {
	            try {
	                if (message.getData() == null || message.getData().length == 0) {
	                    LOG.warn("Received status or empty message, skipping.");
	                    return;
	                }
	                new JetStreamMessageHandler().onMessage(message);
	                message.ack();
	            } catch (Exception e) {
	                LOG.error("Error handling JetStream message: {}", e.getMessage(), e);
	            }
	        };

	        // Direct push subscription without Dispatcher
	        subscription = jetStream.subscribe(subject, messageHandler, false, pushOptions);
	        LOG.error("Successfully started JetStream push-based message subscription on subject: {}", subject);
	        gatewayInfoLog.info("{}# Successfully started JetStream push-based message subscription on subject: {}", this.getClass().getSimpleName(), subject);

	    } catch (Exception e) {
	        LOG.error("Failed to start JetStream push-based subscription. Error: {}", e.getMessage(), e);
	    }
	}

    ```
    I have run mvn clean install -U and confirmed via mvn dependency:tree that only 2.21.1 is being used.
   Is there something I am missing, or is this method signature not available in 2.21.1?
   I tried in 2.17.7, 2.18.0m, 2.20.5, 2.21.1
   
   NOTE : with dispatcher it is working 
   
   Thanks

[scottf]

I'm working on some example code. In the meantime,

Minor

1. You set

.retentionPolicy(RetentionPolicy.Limits)

but don't set any limits.

2. Your message handler will never get status messages. It will only get actual message from a stream.
3. getData will only be null or empty if your message data is null or empty. Probably null if I remember correctly.
4. You are using push messages. This works and certainly if you have a use case of a small amount of messages, but I would consider using simplification consuming.

Major

5. No subscription api takes a handler without a dispatcher, so I don't know how this even compiles.

subscription = jetStream.subscribe(subject, messageHandler, false, pushOptions);

The correct subscribe takes both a dispatcher and a handler.

[scottf]

Please see the repo for links to docs and examples:
https://github.com/nats-io/nats.java?tab=readme-ov-file#examples-and-other-documentation

[scottf]

public static void main(String[] args) {
    try {
        var stream = "stream1329";
        var subject = "sub1329";
        var consumerName = "con1329";

        try (var connection = Nats.connect()) {
            var jetStream = connection.jetStream();
            var jsManagement = connection.jetStreamManagement();

            // Create Stream
            // - remove any existing stream and make a new one for this test
            try {
                jsManagement.deleteStream(stream);
            }
            catch (Exception ignore) {}
            try {
                jsManagement.addStream(StreamConfiguration.builder()
                    .name(stream)
                    .subjects(subject)
                    .storageType(StorageType.Memory)
                    .retentionPolicy(RetentionPolicy.Limits)
                    .build());
                System.out.println("[Info] JetStream stream created for subject: " + subject);
            }
            catch (JetStreamApiException e) {
                System.err.println("[Error] Failed to create JetStream stream: " + e);
            }

            // publish some messages, deliverPolicy(DeliverPolicy.New) new should not get these
            for (int x = 0; x < 10; x++) {
                jetStream.publish(subject, ("before." + x).getBytes());
            }

            // just giving time to make sure deliver policy takes effect
            // by not creating the consumer for a little bit
            Thread.sleep(500);

            // Create a Message Handler
            MessageHandler messageHandler = message -> {
                try {
                    System.out.println("[Info] Message received: Subject: " + new String(message.getData()));
                    message.ack();
                }
                catch (Exception e) {
                    System.err.println("[Error] Error handling JetStream message: " + e);
                }
            };

            // We need a dispatcher for the subscription
            var dispatcher = connection.createDispatcher();

            // create a standard push consumer
            var pushSubscribeOptions = PushSubscribeOptions.builder()
                .configuration(ConsumerConfiguration.builder()
                    .durable(consumerName)
                    .ackPolicy(AckPolicy.Explicit)
                    .deliverPolicy(DeliverPolicy.New)
                    .filterSubject(subject)
                    .build())
                .build();

            // subscribe, which will also create the consumer
            var subscription = jetStream.subscribe(null, dispatcher, messageHandler, false, pushSubscribeOptions);
            System.out.println("[Info] Successfully started JetStream push-based message subscription on subject: " + subject);

            // make sure the consumer is made
            var consumerInfo = jsManagement.getConsumerInfo(stream, consumerName);
            System.out.println("[Info] " + consumerInfo);

            // publish more messages that deliverPolicy(DeliverPolicy.New) will get
            // since the subscription is in place
            for (int x = 0; x < 10; x++) {
                jetStream.publish(subject, ("after." + x).getBytes());
            }

            Thread.sleep(100); // sleeping long enough to get 10 messages.
        }
        catch (Exception e) {
            System.out.println("[Error] Failed to start JetStream push-based subscription. Error: " + e);
        }
    }
    catch (Exception e) {
        throw new RuntimeException(e);
    }
}