Prefix only and comprehensive unit testing.

Author: scottf

src/main/java/io/nats/client/BaseConsumerContext.java

@@ -31,8 +31,8 @@ public interface BaseConsumerContext {
      * Read the next message with max wait set to {@value BaseConsumeOptions#DEFAULT_EXPIRES_IN_MILLIS} ms
      * @return the next message or null if the max wait expires
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
-     * @throws InterruptedException if one is thrown, in order to propagate it up
+     *         server, such as timeout or interruption
+     * @throws InterruptedException if one is thrown, to propagate it up
      * @throws JetStreamStatusCheckedException an exception representing a status that requires attention,
      *         such as the consumer was deleted on the server in the middle of use.
      * @throws JetStreamApiException the request had an error related to the data
@@ -44,8 +44,8 @@ public interface BaseConsumerContext {
      * @param maxWait duration of max wait. Cannot be less than {@value BaseConsumeOptions#MIN_EXPIRES_MILLS} milliseconds.
      * @return the next message or null if the max wait expires
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
-     * @throws InterruptedException if one is thrown, in order to propagate it up
+     *         server, such as timeout or interruption
+     * @throws InterruptedException if one is thrown, to propagate it up
      * @throws JetStreamStatusCheckedException an exception representing a status that requires attention,
      *         such as the consumer was deleted on the server in the middle of use.
      * @throws JetStreamApiException the request had an error related to the data
@@ -57,8 +57,8 @@ public interface BaseConsumerContext {
      * @param maxWaitMillis the max wait value in milliseconds. Cannot be less than {@value BaseConsumeOptions#MIN_EXPIRES_MILLS} milliseconds.
      * @return the next message or null if the max wait expires
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
-     * @throws InterruptedException if one is thrown, in order to propagate it up
+     *         server, such as timeout or interruption
+     * @throws InterruptedException if one is thrown, to propagate it up
      * @throws JetStreamStatusCheckedException an exception representing a status that requires attention,
      *         such as the consumer was deleted on the server in the middle of use.
      * @throws JetStreamApiException the request had an error related to the data
@@ -67,10 +67,10 @@ public interface BaseConsumerContext {
 
     /**
      * Start a one use Fetch Consumer using all defaults other than the number of messages. See {@link FetchConsumer}
-     * @param maxMessages the maximum number of message to consume
+     * @param maxMessages the maximum number of messages to consume
      * @return the FetchConsumer instance
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     FetchConsumer fetchMessages(int maxMessages) throws IOException, JetStreamApiException;
@@ -80,38 +80,38 @@ public interface BaseConsumerContext {
      * @param maxBytes the maximum number of bytes to consume
      * @return the FetchConsumer instance
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     FetchConsumer fetchBytes(int maxBytes) throws IOException, JetStreamApiException;
 
     /**
-     * Start a one use Fetch Consumer with complete custom consume options. See {@link FetchConsumer}
+     * Start a one-use Fetch Consumer with complete custom consume options. See {@link FetchConsumer}
      * @param fetchConsumeOptions the custom fetch consume options. See {@link FetchConsumeOptions}
      * @return the FetchConsumer instance
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     FetchConsumer fetch(FetchConsumeOptions fetchConsumeOptions) throws IOException, JetStreamApiException;
 
     /**
      * Start a long-running IterableConsumer with default ConsumeOptions. See {@link IterableConsumer} and {@link ConsumeOptions}
-     * IterableConsumer require the developer call nextMessage.
+     * IterableConsumer require the developer calls nextMessage.
      * @return the IterableConsumer instance
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     IterableConsumer iterate() throws IOException, JetStreamApiException;
 
     /**
      * Start a long-running IterableConsumer with custom ConsumeOptions. See {@link IterableConsumer} and {@link ConsumeOptions}
-     * IterableConsumer requires the developer call nextMessage.
+     * IterableConsumer requires the developer calls nextMessage.
      * @param consumeOptions the custom consume options
      * @return the IterableConsumer instance
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     IterableConsumer iterate(ConsumeOptions consumeOptions) throws IOException, JetStreamApiException;
@@ -121,7 +121,7 @@ public interface BaseConsumerContext {
      * @param handler the MessageHandler used for receiving messages.
      * @return the MessageConsumer instance
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     MessageConsumer consume(MessageHandler handler) throws IOException, JetStreamApiException;
@@ -130,10 +130,10 @@ public interface BaseConsumerContext {
      * Start a long-running MessageConsumer with default ConsumeOptions. See {@link MessageConsumer} and  {@link ConsumeOptions}
      *
      * @param dispatcher The dispatcher to handle this subscription
-     * @param handler    the MessageHandler used for receiving messages.
+     * @param handler the MessageHandler used for receiving messages.
      * @return the MessageConsumer instance
-     * @throws IOException           covers various communication issues with the NATS
-     *                               server such as timeout or interruption
+     * @throws IOException covers various communication issues with the NATS
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     MessageConsumer consume(Dispatcher dispatcher, MessageHandler handler) throws IOException, JetStreamApiException;
@@ -142,10 +142,10 @@ public interface BaseConsumerContext {
      * Start a long-running MessageConsumer with custom ConsumeOptions. See {@link MessageConsumer} and  {@link ConsumeOptions}
      *
      * @param consumeOptions the custom consume options
-     * @param handler        the MessageHandler used for receiving messages.
+     * @param handler the MessageHandler used for receiving messages.
      * @return the MessageConsumer instance
-     * @throws IOException           covers various communication issues with the NATS
-     *                               server such as timeout or interruption
+     * @throws IOException covers various communication issues with the NATS
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     MessageConsumer consume(ConsumeOptions consumeOptions, MessageHandler handler) throws IOException, JetStreamApiException;
@@ -154,11 +154,11 @@ public interface BaseConsumerContext {
      * Start a long-running MessageConsumer with custom ConsumeOptions. See {@link MessageConsumer} and  {@link ConsumeOptions}
      *
      * @param consumeOptions the custom consume options
-     * @param dispatcher     The dispatcher to handle this subscription
-     * @param handler        the MessageHandler used for receiving messages.
+     * @param dispatcher the dispatcher to handle this subscription
+     * @param handler the MessageHandler used for receiving messages.
      * @return the MessageConsumer instance
-     * @throws IOException           covers various communication issues with the NATS
-     *                               server such as timeout or interruption
+     * @throws IOException covers various communication issues with the NATS
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     MessageConsumer consume(ConsumeOptions consumeOptions, Dispatcher dispatcher, MessageHandler handler) throws IOException, JetStreamApiException;

src/main/java/io/nats/client/MessageConsumer.java

@@ -24,7 +24,8 @@
 public interface MessageConsumer extends AutoCloseable {
     /**
      * Gets the consumer name associated with the subscription.
-     * Some simplified consumer types do not support this, so it might be null.
+     * For simplified consumers, this value can be null unless
+     * the consumer info was manually read via {@link #getConsumerInfo()}.
      * @return the consumer name
      */
     String getConsumerName();
@@ -33,21 +34,22 @@ public interface MessageConsumer extends AutoCloseable {
      * Gets information about the consumer behind this subscription.
      * @return consumer information
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     ConsumerInfo getConsumerInfo() throws IOException, JetStreamApiException;
 
     /**
      * Gets information about the consumer behind this subscription.
-     * This returns the last read version of Consumer Info, which could technically be out of date.
+     * This returns the last read version of Consumer Info,
+     * which could be null or out of date.
      * @return consumer information
      */
     ConsumerInfo getCachedConsumerInfo();
 
     /**
-     * Use {@link close()} to unsubscribe. Stop will not unsubcribe or clean up resources.
-     * The consumer will finish all pull request already in progress, but will not start any new ones.
+     * Use {@link #close()} to unsubscribe. Stop will not unsubcribe or clean up resources.
+     * The consumer will finish all pull requests already in progress, but will not start any new ones.
      */
     void stop();
 

src/main/java/io/nats/client/api/Watcher.java

@@ -31,11 +31,12 @@ public interface Watcher<T> {
     void endOfData();
 
     /**
-     * The watcher can supply a consumer name to be used when creating the internal watch consumer,
-     * improving the ability to monitor the consumer.
+     * The watcher can supply a prefix to use on the consumer name
+     * that is generated when creating the internal watch consumer.
+     * This can be useful for monitoring the consumer.
      * @return the name, or null if not needed, which is the default interface implementation.
      */
-    default String consumerName() {
+    default String getConsumerNamePrefix() {
         return null;
     }
 }

src/main/java/io/nats/client/impl/NatsConsumerContext.java

@@ -86,7 +86,6 @@ static class OrderedPullSubscribeOptionsBuilder extends PullSubscribeOptions.Bui
         }
     }
 
-    int x = 0;
     @Override
     public NatsJetStreamPullSubscription subscribe(MessageHandler messageHandler, Dispatcher userDispatcher, PullMessageManager optionalPmm, Long optionalInactiveThreshold) throws IOException, JetStreamApiException {
         PullSubscribeOptions pso;
@@ -199,12 +198,13 @@ public Message next(long maxWaitMillis) throws IOException, InterruptedException
 
             try {
                 long inactiveThreshold = maxWaitMillis * 110 / 100; // 10% longer than the wait
-                nmcb = new NatsMessageConsumerBase(consumerName.get(), cachedConsumerInfo.get());
+                nmcb = new NatsMessageConsumerBase(cachedConsumerInfo.get());
                 nmcb.initSub(subscribe(null, null, null, inactiveThreshold));
+                nmcb.setConsumerName(consumerName.get()); // the call to subscribe sets this
+                trackConsume(nmcb); // this has to be done after the nmcb is fully set up
                 nmcb.sub._pull(PullRequestOptions.builder(1)
                     .expiresIn(maxWaitMillis - EXPIRE_ADJUSTMENT)
                     .build(), false, null);
-                trackConsume(nmcb);
             }
             catch (Exception e) {
                 if (nmcb != null) {
@@ -220,7 +220,7 @@ public Message next(long maxWaitMillis) throws IOException, InterruptedException
             stateLock.unlock();
         }
 
-        // intentionally outside of lock
+        // intentionally outside the lock
         try {
             return nmcb.sub.nextMessage(maxWaitMillis);
         }
@@ -260,7 +260,7 @@ public FetchConsumer fetch(FetchConsumeOptions fetchConsumeOptions) throws IOExc
             stateLock.lock();
             checkState();
             Validator.required(fetchConsumeOptions, "Fetch Consume Options");
-            return (FetchConsumer)trackConsume(new NatsFetchConsumer(this, consumerName.get(), cachedConsumerInfo.get(), fetchConsumeOptions));
+            return (FetchConsumer)trackConsume(new NatsFetchConsumer(this, cachedConsumerInfo.get(), fetchConsumeOptions));
         }
         finally {
             stateLock.unlock();
@@ -284,7 +284,7 @@ public IterableConsumer iterate(ConsumeOptions consumeOptions) throws IOExceptio
             stateLock.lock();
             checkState();
             Validator.required(consumeOptions, "Consume Options");
-            return (IterableConsumer) trackConsume(new NatsIterableConsumer(this, consumerName.get(), cachedConsumerInfo.get(), consumeOptions));
+            return (IterableConsumer) trackConsume(new NatsIterableConsumer(this, cachedConsumerInfo.get(), consumeOptions));
         }
         finally {
             stateLock.unlock();
@@ -325,7 +325,7 @@ public MessageConsumer consume(ConsumeOptions consumeOptions, Dispatcher userDis
             checkState();
             Validator.required(handler, "Message Handler");
             Validator.required(consumeOptions, "Consume Options");
-            return trackConsume(new NatsMessageConsumer(this, consumerName.get(), cachedConsumerInfo.get(), consumeOptions, userDispatcher, handler));
+            return trackConsume(new NatsMessageConsumer(this, cachedConsumerInfo.get(), consumeOptions, userDispatcher, handler));
         }
         finally {
             stateLock.unlock();

src/main/java/io/nats/client/impl/NatsFetchConsumer.java

@@ -29,11 +29,10 @@ class NatsFetchConsumer extends NatsMessageConsumerBase implements FetchConsumer
     private long startNanos;
 
     NatsFetchConsumer(SimplifiedSubscriptionMaker subscriptionMaker,
-                      String consumerName,
                       ConsumerInfo cachedConsumerInfo,
                       FetchConsumeOptions fetchConsumeOptions) throws IOException, JetStreamApiException
     {
-        super(consumerName, cachedConsumerInfo);
+        super(cachedConsumerInfo);
 
         boolean isNoWait = fetchConsumeOptions.isNoWait();
         long expiresInMillis = fetchConsumeOptions.getExpiresInMillis();

src/main/java/io/nats/client/impl/NatsIterableConsumer.java

@@ -21,8 +21,8 @@
 
 class NatsIterableConsumer extends NatsMessageConsumer implements IterableConsumer {
 
-    NatsIterableConsumer(SimplifiedSubscriptionMaker subscriptionMaker, String consumerName, ConsumerInfo cachedConsumerInfo, ConsumeOptions opts) throws IOException, JetStreamApiException {
-        super(subscriptionMaker, consumerName, cachedConsumerInfo, opts, null, null);
+    NatsIterableConsumer(SimplifiedSubscriptionMaker subscriptionMaker, ConsumerInfo cachedConsumerInfo, ConsumeOptions opts) throws IOException, JetStreamApiException {
+        super(subscriptionMaker, cachedConsumerInfo, opts, null, null);
     }
 
     /**

src/main/java/io/nats/client/impl/NatsKeyValueWatchSubscription.java

@@ -70,6 +70,6 @@ public void onMessage(Message m) throws InterruptedException {
             readSubjects.add(kv.readSubject(keyPattern.trim()));
         }
 
-        finishInit(kv, readSubjects, deliverPolicy, headersOnly, fromRevision, handler, watcher.consumerName());
+        finishInit(kv, readSubjects, deliverPolicy, headersOnly, fromRevision, handler, watcher.getConsumerNamePrefix());
     }
 }

src/main/java/io/nats/client/impl/NatsMessageConsumer.java

@@ -27,13 +27,12 @@ class NatsMessageConsumer extends NatsMessageConsumerBase implements PullManager
     protected final MessageHandler userMessageHandler;
 
     NatsMessageConsumer(SimplifiedSubscriptionMaker subscriptionMaker,
-                        String consumerName,
                         ConsumerInfo cachedConsumerInfo,
                         ConsumeOptions consumeOpts,
                         Dispatcher userDispatcher,
                         final MessageHandler userMessageHandler) throws IOException, JetStreamApiException
     {
-        super(consumerName, cachedConsumerInfo);
+        super(cachedConsumerInfo);
 
         this.subscriptionMaker = subscriptionMaker;
         this.consumeOpts = consumeOpts;

src/main/java/io/nats/client/impl/NatsMessageConsumerBase.java

@@ -28,13 +28,19 @@ class NatsMessageConsumerBase implements MessageConsumer {
     protected ConsumerInfo cachedConsumerInfo;
     protected String consumerName;
 
-    NatsMessageConsumerBase(String consumerName, ConsumerInfo cachedConsumerInfo) {
-        this.consumerName = consumerName;
+    NatsMessageConsumerBase(ConsumerInfo cachedConsumerInfo) {
         this.cachedConsumerInfo = cachedConsumerInfo;
+        if (cachedConsumerInfo != null) {
+            this.consumerName = cachedConsumerInfo.getName();
+        }
         this.stopped = new AtomicBoolean(false);
         this.finished = new AtomicBoolean(false);
     }
 
+    void setConsumerName(String consumerName) {
+        this.consumerName = consumerName;
+    }
+
     void initSub(NatsJetStreamPullSubscription sub) {
         this.sub = sub;
         pmm = (PullMessageManager)sub.manager;
@@ -59,6 +65,9 @@ public boolean isFinished() {
      */
     @Override
     public String getConsumerName() {
+        if (consumerName == null) {
+            consumerName = cachedConsumerInfo.getName();
+        }
         return consumerName;
     }
 
@@ -67,9 +76,9 @@ public String getConsumerName() {
      */
     @Override
     public ConsumerInfo getConsumerInfo() throws IOException, JetStreamApiException {
-        // don't look up consumer info if it was never set - this check is for ordered consumer
-        if (cachedConsumerInfo != null) {
+        if (cachedConsumerInfo == null) {
             cachedConsumerInfo = sub.getConsumerInfo();
+            consumerName = cachedConsumerInfo.getName();
         }
         return cachedConsumerInfo;
     }

src/main/java/io/nats/client/impl/NatsObjectStoreWatchSubscription.java

@@ -59,6 +59,6 @@ public void onMessage(Message m) throws InterruptedException {
                 }
             };
 
-        finishInit(os, Collections.singletonList(os.rawAllMetaSubject()), deliverPolicy, headersOnly, ULONG_UNSET, handler, watcher.consumerName());
+        finishInit(os, Collections.singletonList(os.rawAllMetaSubject()), deliverPolicy, headersOnly, ULONG_UNSET, handler, watcher.getConsumerNamePrefix());
     }
 }