Ability to supply a name for an ordered consumer.

* Original Ordered Push Consumers can supply a name.
* Key Value and Object Store watchers can supply a name
* Simplified ordered consumers can supply a prefix in OrderedConsumerConfiguration

Author: scottf

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

@@ -28,7 +28,7 @@ public interface KeyValueManagement {
      * @param config the key value configuration
      * @return bucket info
      * @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
      * @throws IllegalArgumentException the server is not JetStream enabled
      */
@@ -39,7 +39,7 @@ public interface KeyValueManagement {
      * @param config the key value configuration
      * @return bucket info
      * @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
      * @throws IllegalArgumentException the server is not JetStream enabled
      */
@@ -49,7 +49,7 @@ public interface KeyValueManagement {
      * Get the list of bucket names.
      * @return list of bucket names
      * @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
      */
     List<String> getBucketNames() throws IOException, JetStreamApiException;
@@ -59,7 +59,7 @@ public interface KeyValueManagement {
      * @deprecated Use {@link #getStatus(String)} instead.
      * @param bucketName the bucket name to use
      * @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
      * @return the bucket status object
      */
@@ -70,7 +70,7 @@ public interface KeyValueManagement {
      * Gets the status for an existing bucket.
      * @param bucketName the bucket name to use
      * @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
      * @return the bucket status object
      */
@@ -80,7 +80,7 @@ public interface KeyValueManagement {
      * Get the statuses for all buckets
      * @return list of statuses
      * @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
      */
     List<KeyValueStatus> getStatuses() throws IOException, JetStreamApiException;
@@ -89,7 +89,7 @@ public interface KeyValueManagement {
      * Deletes an existing bucket. Will throw a JetStreamApiException if the delete fails.
      * @param bucketName the stream name to use.
      * @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
      */
     void delete(String bucketName) throws IOException, JetStreamApiException;

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

@@ -36,11 +36,12 @@ public class OrderedConsumerConfiguration implements JsonSerializable {
     private ZonedDateTime startTime;
     private ReplayPolicy replayPolicy;
     private Boolean headersOnly;
+    private String consumerNamePrefix;
 
     /**
      * OrderedConsumerConfiguration creation works like a builder.
      * The builder supports chaining and will create a default set of options if
-     * no methods are calls, including setting the filter subject to ">"
+     * no methods are calls, including setting the filter subject to >
      */
     public OrderedConsumerConfiguration() {
         startSequence = ConsumerConfiguration.LONG_UNSET;
@@ -67,7 +68,7 @@ public OrderedConsumerConfiguration(JsonValue v) throws JsonParseException {
 
     /**
      * Returns a JSON representation of this ordered consumer configuration.
-     * @return json ordered consumer configuration json string
+     * @return JSON ordered consumer configuration JSON string
      */
     public String toJson() {
         StringBuilder sb = beginJson();
@@ -89,7 +90,7 @@ public String toJson() {
     /**
      * Sets the filter subject of the OrderedConsumerConfiguration.
      * @param filterSubject the filter subject
-     * @return Builder
+     * @return The Builder
      */
     public OrderedConsumerConfiguration filterSubject(String filterSubject) {
         return filterSubjects(Collections.singletonList(filterSubject));
@@ -98,7 +99,7 @@ public OrderedConsumerConfiguration filterSubject(String filterSubject) {
     /**
      * Sets the filter subjects of the OrderedConsumerConfiguration.
      * @param filterSubject the filter subject
-     * @return Builder
+     * @return The Builder
      */
     public OrderedConsumerConfiguration filterSubjects(String... filterSubject) {
         return filterSubjects(Arrays.asList(filterSubject));
@@ -107,14 +108,14 @@ public OrderedConsumerConfiguration filterSubjects(String... filterSubject) {
     /**
      * Sets the filter subject of the OrderedConsumerConfiguration.
      * @param filterSubjects one or more filter subjects
-     * @return Builder
+     * @return The Builder
      */
     public OrderedConsumerConfiguration filterSubjects(List<String> filterSubjects) {
         this.filterSubjects.clear();
         for (String fs : filterSubjects) {
-            String fsean = emptyAsNull(fs);
-            if (fsean != null) {
-                this.filterSubjects.add(fsean);
+            String fsEan = emptyAsNull(fs);
+            if (fsEan != null) {
+                this.filterSubjects.add(fsEan);
             }
         }
         if (this.filterSubjects.isEmpty()) {
@@ -126,7 +127,7 @@ public OrderedConsumerConfiguration filterSubjects(List<String> filterSubjects)
     /**
      * Sets the delivery policy of the OrderedConsumerConfiguration.
      * @param deliverPolicy the delivery policy.
-     * @return Builder
+     * @return The Builder
      */
     public OrderedConsumerConfiguration deliverPolicy(DeliverPolicy deliverPolicy) {
         this.deliverPolicy = deliverPolicy;
@@ -136,7 +137,7 @@ public OrderedConsumerConfiguration deliverPolicy(DeliverPolicy deliverPolicy) {
     /**
      * Sets the start sequence of the OrderedConsumerConfiguration.
      * @param startSequence the start sequence
-     * @return Builder
+     * @return The Builder
      */
     public OrderedConsumerConfiguration startSequence(long startSequence) {
         this.startSequence = startSequence < 1 ? ConsumerConfiguration.LONG_UNSET : startSequence;
@@ -146,7 +147,7 @@ public OrderedConsumerConfiguration startSequence(long startSequence) {
     /**
      * Sets the start time of the OrderedConsumerConfiguration.
      * @param startTime the start time
-     * @return Builder
+     * @return The Builder
      */
     public OrderedConsumerConfiguration startTime(ZonedDateTime startTime) {
         this.startTime = startTime;
@@ -156,7 +157,7 @@ public OrderedConsumerConfiguration startTime(ZonedDateTime startTime) {
     /**
      * Sets the replay policy of the OrderedConsumerConfiguration.
      * @param replayPolicy the replay policy.
-     * @return Builder
+     * @return The Builder
      */
     public OrderedConsumerConfiguration replayPolicy(ReplayPolicy replayPolicy) {
         this.replayPolicy = replayPolicy;
@@ -167,13 +168,23 @@ public OrderedConsumerConfiguration replayPolicy(ReplayPolicy replayPolicy) {
      * set the headers only flag saying to deliver only the headers of
      * messages in the stream and not the bodies
      * @param headersOnly the flag
-     * @return Builder
+     * @return The Builder
      */
     public OrderedConsumerConfiguration headersOnly(Boolean headersOnly) {
         this.headersOnly = headersOnly != null && headersOnly ? true : null;
         return this;
     }
 
+    /**
+     * Sets the consumer name prefix for consumers created by this configuration.
+     * @param consumerNamePrefix the prefix or null to clear.
+     * @return The Builder
+     */
+    public OrderedConsumerConfiguration consumerNamePrefix(String consumerNamePrefix) {
+        this.consumerNamePrefix = emptyAsNull(consumerNamePrefix);
+        return this;
+    }
+
     public String getFilterSubject() {
         return filterSubjects == null || filterSubjects.size() != 1 ? null : filterSubjects.get(0);
     }
@@ -205,4 +216,8 @@ public ReplayPolicy getReplayPolicy() {
     public Boolean getHeadersOnly() {
         return headersOnly;
     }
+
+    public String getConsumerNamePrefix() {
+        return consumerNamePrefix;
+    }
 }

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

@@ -19,8 +19,7 @@
 public interface Watcher<T> {
 
     /**
-     * Called when an object has been updated
-     *
+     * Called when an object has been updated.
      * @param t The watched object
      */
     void watch(T t);
@@ -30,4 +29,13 @@ public interface Watcher<T> {
      * or if there is data, the first time the watch exhausts all existing data.
      */
     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.
+     * @return the name, or null if not needed, which is the default interface implementation.
+     */
+    default String consumerName() {
+        return null;
+    }
 }

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

@@ -37,8 +37,8 @@ public class NatsConsumerContext implements ConsumerContext, SimplifiedSubscript
     private final ReentrantLock stateLock;
     private final NatsStreamContext streamCtx;
     private final boolean ordered;
-    private final ConsumerConfiguration originalOrderedCc;
-    private final String subscribeSubject;
+    private final ConsumerConfiguration orderedConsumerConfigTemplate;
+    private final String orderedConsumerNamePrefix;
     private final PullSubscribeOptions unorderedBindPso;
 
     private final AtomicReference<ConsumerInfo> cachedConsumerInfo;
@@ -47,7 +47,7 @@ public class NatsConsumerContext implements ConsumerContext, SimplifiedSubscript
     private final AtomicReference<Dispatcher> defaultDispatcher;
     private final AtomicReference<NatsMessageConsumerBase> lastConsumer;
 
-    NatsConsumerContext(NatsStreamContext sc, ConsumerInfo unorderedConsumerInfo, OrderedConsumerConfiguration orderedCc) {
+    NatsConsumerContext(NatsStreamContext sc, ConsumerInfo unorderedConsumerInfo, OrderedConsumerConfiguration occ) {
         stateLock = new ReentrantLock();
         streamCtx = sc;
         cachedConsumerInfo = new AtomicReference<>();
@@ -57,23 +57,23 @@ public class NatsConsumerContext implements ConsumerContext, SimplifiedSubscript
         lastConsumer = new AtomicReference<>();
         if (unorderedConsumerInfo != null) {
             ordered = false;
-            originalOrderedCc = null;
-            subscribeSubject = null;
+            orderedConsumerNamePrefix = null;
+            orderedConsumerConfigTemplate = null;
             cachedConsumerInfo.set(unorderedConsumerInfo);
             consumerName.set(unorderedConsumerInfo.getName());
             unorderedBindPso = PullSubscribeOptions.fastBind(sc.streamName, unorderedConsumerInfo.getName());
         }
         else {
             ordered = true;
-            originalOrderedCc = ConsumerConfiguration.builder()
-                .filterSubjects(orderedCc.getFilterSubjects())
-                .deliverPolicy(orderedCc.getDeliverPolicy())
-                .startSequence(orderedCc.getStartSequence())
-                .startTime(orderedCc.getStartTime())
-                .replayPolicy(orderedCc.getReplayPolicy())
-                .headersOnly(orderedCc.getHeadersOnly())
+            orderedConsumerNamePrefix = occ.getConsumerNamePrefix();
+            orderedConsumerConfigTemplate = ConsumerConfiguration.builder()
+                .filterSubjects(occ.getFilterSubjects())
+                .deliverPolicy(occ.getDeliverPolicy())
+                .startSequence(occ.getStartSequence())
+                .startTime(occ.getStartTime())
+                .replayPolicy(occ.getReplayPolicy())
+                .headersOnly(occ.getHeadersOnly())
                 .build();
-            subscribeSubject = Validator.validateSubject(originalOrderedCc.getFilterSubject(), false);
             unorderedBindPso = null;
         }
     }
@@ -86,6 +86,7 @@ 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;
@@ -94,8 +95,9 @@ public NatsJetStreamPullSubscription subscribe(MessageHandler messageHandler, Di
             if (lastCon != null) {
                 highestSeq.set(Math.max(highestSeq.get(), lastCon.pmm.lastStreamSeq));
             }
+            consumerName.set(orderedConsumerNamePrefix == null ? null : orderedConsumerNamePrefix + NUID.nextGlobalSequence());
             ConsumerConfiguration cc = streamCtx.js.consumerConfigurationForOrdered(
-                originalOrderedCc, highestSeq.get(), null, null, optionalInactiveThreshold);
+                orderedConsumerConfigTemplate, highestSeq.get(), null, consumerName.get(), optionalInactiveThreshold);
             pso = new OrderedPullSubscribeOptionsBuilder(streamCtx.streamName, cc).build();
         }
         else {
@@ -104,7 +106,7 @@ public NatsJetStreamPullSubscription subscribe(MessageHandler messageHandler, Di
 
         if (messageHandler == null) {
             return (NatsJetStreamPullSubscription) streamCtx.js.createSubscription(
-                subscribeSubject, null, pso, null, null, null, false, optionalPmm);
+                null, null, pso, null, null, null, false, optionalPmm);
         }
 
         Dispatcher d = userDispatcher;
@@ -116,7 +118,7 @@ public NatsJetStreamPullSubscription subscribe(MessageHandler messageHandler, Di
             }
         }
         return (NatsJetStreamPullSubscription) streamCtx.js.createSubscription(
-            subscribeSubject, null, pso, null, (NatsDispatcher) d, messageHandler, false, optionalPmm);
+            null, null, pso, null, (NatsDispatcher) d, messageHandler, false, optionalPmm);
     }
 
     private void checkState() throws IOException {
@@ -197,7 +199,7 @@ public Message next(long maxWaitMillis) throws IOException, InterruptedException
 
             try {
                 long inactiveThreshold = maxWaitMillis * 110 / 100; // 10% longer than the wait
-                nmcb = new NatsMessageConsumerBase(cachedConsumerInfo.get());
+                nmcb = new NatsMessageConsumerBase(consumerName.get(), cachedConsumerInfo.get());
                 nmcb.initSub(subscribe(null, null, null, inactiveThreshold));
                 nmcb.sub._pull(PullRequestOptions.builder(1)
                     .expiresIn(maxWaitMillis - EXPIRE_ADJUSTMENT)
@@ -258,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, cachedConsumerInfo.get(), fetchConsumeOptions));
+            return (FetchConsumer)trackConsume(new NatsFetchConsumer(this, consumerName.get(), cachedConsumerInfo.get(), fetchConsumeOptions));
         }
         finally {
             stateLock.unlock();
@@ -282,7 +284,7 @@ public IterableConsumer iterate(ConsumeOptions consumeOptions) throws IOExceptio
             stateLock.lock();
             checkState();
             Validator.required(consumeOptions, "Consume Options");
-            return (IterableConsumer) trackConsume(new NatsIterableConsumer(this, cachedConsumerInfo.get(), consumeOptions));
+            return (IterableConsumer) trackConsume(new NatsIterableConsumer(this, consumerName.get(), cachedConsumerInfo.get(), consumeOptions));
         }
         finally {
             stateLock.unlock();
@@ -323,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, cachedConsumerInfo.get(), consumeOptions, userDispatcher, handler));
+            return trackConsume(new NatsMessageConsumer(this, consumerName.get(), cachedConsumerInfo.get(), consumeOptions, userDispatcher, handler));
         }
         finally {
             stateLock.unlock();

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

@@ -29,10 +29,11 @@ class NatsFetchConsumer extends NatsMessageConsumerBase implements FetchConsumer
     private long startNanos;
 
     NatsFetchConsumer(SimplifiedSubscriptionMaker subscriptionMaker,
+                      String consumerName,
                       ConsumerInfo cachedConsumerInfo,
                       FetchConsumeOptions fetchConsumeOptions) throws IOException, JetStreamApiException
     {
-        super(cachedConsumerInfo);
+        super(consumerName, 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, ConsumerInfo cachedConsumerInfo, ConsumeOptions opts) throws IOException, JetStreamApiException {
-        super(subscriptionMaker, cachedConsumerInfo, opts, null, null);
+    NatsIterableConsumer(SimplifiedSubscriptionMaker subscriptionMaker, String consumerName, ConsumerInfo cachedConsumerInfo, ConsumeOptions opts) throws IOException, JetStreamApiException {
+        super(subscriptionMaker, consumerName, 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);
+        finishInit(kv, readSubjects, deliverPolicy, headersOnly, fromRevision, handler, watcher.consumerName());
     }
 }

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

@@ -27,12 +27,13 @@ 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(cachedConsumerInfo);
+        super(consumerName, cachedConsumerInfo);
 
         this.subscriptionMaker = subscriptionMaker;
         this.consumeOpts = consumeOpts;