Ordered Consumers automatically are prefixed if a consumer configuration name is supplied.

Author: scottf

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

@@ -20,10 +20,10 @@
 
 import static io.nats.client.support.ApiConstants.*;
 import static io.nats.client.support.JsonUtils.*;
+import static io.nats.client.support.JsonValueUtils.*;
 import static io.nats.client.support.JsonValueUtils.readBoolean;
 import static io.nats.client.support.JsonValueUtils.readInteger;
 import static io.nats.client.support.JsonValueUtils.readLong;
-import static io.nats.client.support.JsonValueUtils.*;
 
 /**
  * Base Consume Options are provided to customize the way the consume and
@@ -208,7 +208,7 @@ public B thresholdPercent(int thresholdPercent) {
 
         /**
          * Raise status warning turns on sending status messages to the error listener.
-         * The default of to not raise status warning
+         * The default is to not raise status warnings
          * @return the builder
          */
         public B raiseStatusWarnings() {
@@ -218,7 +218,8 @@ public B raiseStatusWarnings() {
 
         /**
          * Turn on or off raise status warning turns. When on, status messages are sent to the error listener.
-         * The default of to not raise status warning
+         * The default is to not raise status warnings
+         * @param raiseStatusWarnings flag indicating whether to raise status messages
          * @return the builder
          */
         public B raiseStatusWarnings(boolean raiseStatusWarnings) {

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

@@ -34,8 +34,8 @@ public abstract class SubscribeOptions {
     protected final boolean ordered;
     protected final long messageAlarmTime;
     protected final ConsumerConfiguration consumerConfig;
-    protected final long pendingMessageLimit; // Only applicable for non dispatched (sync) push consumers.
-    protected final long pendingByteLimit; // Only applicable for non dispatched (sync) push consumers.
+    protected final long pendingMessageLimit; // Only applicable for non-dispatched (sync) push consumers.
+    protected final long pendingByteLimit; // Only applicable for non-dispatched (sync) push consumers.
     protected final String name;
 
     protected SubscribeOptions(Builder<?, ?> builder, boolean isPull,
@@ -210,15 +210,15 @@ public ConsumerConfiguration getConsumerConfiguration() {
     }
 
     /**
-     * Gets the pending message limit. Only applicable for non dispatched (sync) push consumers.
+     * Gets the pending message limit. Only applicable for non-dispatched (sync) push consumers.
      * @return the message limit
      */
     public long getPendingMessageLimit() {
         return pendingMessageLimit;
     }
 
     /**
-     * Gets the pending byte limit. Only applicable for non dispatched (sync) push consumers.
+     * Gets the pending byte limit. Only applicable for non-dispatched (sync) push consumers.
      * @return the byte limit
      */
     public long getPendingByteLimit() {
@@ -256,7 +256,7 @@ protected static abstract class Builder<B, SO> {
         protected abstract B getThis();
 
         /**
-         * Specify the stream to attach to. If not supplied the stream will be looked up by subject.
+         * Specify the stream to attach to. If not supplied, the stream will look up by the stream by subject.
          * Null or empty clears the field.
          * @param stream the name of the stream
          * @return the builder
@@ -269,8 +269,8 @@ public B stream(String stream) {
         /**
          * Specify binding to an existing consumer via name.
          * The client validates regular (non-fast)
-         * binds to ensure that provided consumer configuration
-         * is consistent with the server version and that
+         * binds to ensure that the provided consumer configuration
+         * is consistent with the server version and that the
          * consumer type (push versus pull) matches the subscription type.
          * @return the builder
          * @param bind whether to bind or not

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

@@ -37,8 +37,7 @@ public class NatsConsumerContext implements ConsumerContext, SimplifiedSubscript
     private final ReentrantLock stateLock;
     private final NatsStreamContext streamCtx;
     private final boolean ordered;
-    private final ConsumerConfiguration orderedConsumerConfigTemplate;
-    private final String orderedConsumerNamePrefix;
+    private final ConsumerConfiguration initialOrderedConsumerConfig;
     private final PullSubscribeOptions unorderedBindPso;
 
     private final AtomicReference<ConsumerInfo> cachedConsumerInfo;
@@ -57,16 +56,15 @@ public class NatsConsumerContext implements ConsumerContext, SimplifiedSubscript
         lastConsumer = new AtomicReference<>();
         if (unorderedConsumerInfo != null) {
             ordered = false;
-            orderedConsumerNamePrefix = null;
-            orderedConsumerConfigTemplate = null;
+            initialOrderedConsumerConfig = null;
             cachedConsumerInfo.set(unorderedConsumerInfo);
             consumerName.set(unorderedConsumerInfo.getName());
             unorderedBindPso = PullSubscribeOptions.fastBind(sc.streamName, unorderedConsumerInfo.getName());
         }
         else {
             ordered = true;
-            orderedConsumerNamePrefix = occ.getConsumerNamePrefix();
-            orderedConsumerConfigTemplate = ConsumerConfiguration.builder()
+            initialOrderedConsumerConfig = ConsumerConfiguration.builder()
+                .name(occ.getConsumerNamePrefix())
                 .filterSubjects(occ.getFilterSubjects())
                 .deliverPolicy(occ.getDeliverPolicy())
                 .startSequence(occ.getStartSequence())
@@ -94,30 +92,32 @@ 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(
-                orderedConsumerConfigTemplate, highestSeq.get(), null, consumerName.get(), optionalInactiveThreshold);
+            ConsumerConfiguration cc = streamCtx.js.consumerConfigurationForOrdered(initialOrderedConsumerConfig, highestSeq.get(), null, optionalInactiveThreshold).build();
             pso = new OrderedPullSubscribeOptionsBuilder(streamCtx.streamName, cc).build();
         }
         else {
             pso = unorderedBindPso;
         }
 
+        NatsJetStreamPullSubscription sub;
         if (messageHandler == null) {
-            return (NatsJetStreamPullSubscription) streamCtx.js.createSubscription(
+            sub = (NatsJetStreamPullSubscription) streamCtx.js.createSubscription(
                 null, null, pso, null, null, null, false, optionalPmm);
         }
-
-        Dispatcher d = userDispatcher;
-        if (d == null) {
-            d = defaultDispatcher.get();
+        else {
+            Dispatcher d = userDispatcher;
             if (d == null) {
-                d = streamCtx.js.conn.createDispatcher();
-                defaultDispatcher.set(d);
+                d = defaultDispatcher.get();
+                if (d == null) {
+                    d = streamCtx.js.conn.createDispatcher();
+                    defaultDispatcher.set(d);
+                }
             }
+            sub = (NatsJetStreamPullSubscription) streamCtx.js.createSubscription(
+                null, null, pso, null, (NatsDispatcher) d, messageHandler, false, optionalPmm);
         }
-        return (NatsJetStreamPullSubscription) streamCtx.js.createSubscription(
-            null, null, pso, null, (NatsDispatcher) d, messageHandler, false, optionalPmm);
+        consumerName.set(sub.getConsumerName());
+        return sub;
     }
 
     private void checkState() throws IOException {

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

@@ -25,6 +25,7 @@
 import static io.nats.client.PushSubscribeOptions.DEFAULT_PUSH_OPTS;
 import static io.nats.client.impl.MessageManager.ManageResult;
 import static io.nats.client.support.NatsJetStreamClientError.*;
+import static io.nats.client.support.NatsJetStreamUtil.generateConsumerName;
 import static io.nats.client.support.NatsRequestCompletableFuture.CancelAction;
 import static io.nats.client.support.Validator.*;
 
@@ -255,6 +256,7 @@ JetStreamSubscription createSubscription(String userSubscribeSubject,
         SubscribeOptions so;
         String stream;
         ConsumerConfiguration userCC;
+        boolean ordered;
         String settledDeliverGroup = null; // push might set this
 
         if (isPullMode) {
@@ -290,7 +292,7 @@ JetStreamSubscription createSubscription(String userSubscribeSubject,
             }
         }
 
-        // 1B. Flow Control / heartbeat not always valid
+        // 1B. Flow Control / heartbeat is not always valid
         if (userCC.getIdleHeartbeat() != null && userCC.getIdleHeartbeat().toMillis() > 0) {
             if (isPullMode) {
                 throw JsSubFcHbNotValidPull.instance();
@@ -300,7 +302,7 @@ JetStreamSubscription createSubscription(String userSubscribeSubject,
             }
         }
 
-        // 2. figure out user provided subjects and prepare the settledFilterSubjects
+        // 2. figure out user-provided subjects and prepare the settledFilterSubjects
         userSubscribeSubject = emptyAsNull(userSubscribeSubject);
         List<String> settledFilterSubjects = new ArrayList<>();
         if (userCC.getFilterSubjects() == null) { // empty filterSubjects gives null
@@ -312,13 +314,13 @@ JetStreamSubscription createSubscription(String userSubscribeSubject,
         else {
             // userCC.filterSubjects not empty, validate them
             settledFilterSubjects.addAll(userCC.getFilterSubjects());
-            // If userSubscribeSubject is provided it must be one of the filter subjects.
+            // If userSubscribeSubject is provided, it must be one of the filter subjects.
             if (userSubscribeSubject != null && !settledFilterSubjects.contains(userSubscribeSubject)) {
                 throw JsSubSubjectDoesNotMatchFilter.instance();
             }
         }
 
-        // 3. Did they tell me what stream? No? look it up.
+        // 3. Did they tell me what stream? No? Look it up.
         final String settledStream;
         if (stream == null) {
             if (settledFilterSubjects.isEmpty()) {
@@ -338,18 +340,20 @@ JetStreamSubscription createSubscription(String userSubscribeSubject,
         if (consumerName == null) {
             consumerName = userCC.getName();
         }
+
         String inboxDeliver = userCC.getDeliverSubject();
 
-        // 4. Does this consumer already exist? FastBind bypasses the lookup;
-        //    the dev better know what they are doing...
-        if (!so.isFastBind() && consumerName != null) {
+        // 4. Does this consumer already exist?
+        //    * FastBind bypasses the lookup; the dev better know what they are doing...
+        //    * ordered also bypasses the lookup b/c we know it's always a new name
+        if (!so.isFastBind() && !so.isOrdered() && consumerName != null) {
             ConsumerInfo serverInfo = lookupConsumerInfo(settledStream, consumerName);
 
             if (serverInfo != null) { // the consumer for that durable already exists
                 serverCC = serverInfo.getConsumerConfiguration();
 
                 // check to see if the user sent a different version than the server has
-                // because modifications are not allowed during create subscription
+                // because modifications are not allowed during "create" subscription
                 ConsumerConfigurationComparer userCCC = new ConsumerConfigurationComparer(userCC);
                 List<String> changes = userCCC.getChanges(serverCC);
                 if (!changes.isEmpty()) {
@@ -367,15 +371,15 @@ else if (nullOrEmpty(serverCC.getDeliverSubject())) {
                 }
 
                 if (serverCC.getDeliverGroup() == null) {
-                    // lookedUp was null, means existing consumer is not a queue consumer
+                    // lookedUp was null, means the existing consumer is not a queue consumer
                     if (settledDeliverGroup == null) {
-                        // ok fine, no queue requested and the existing consumer is also not a queue consumer
+                        // ok fine, no queue was requested, and the existing consumer is also not a queue consumer
                         // we must check if the consumer is in use though
                         if (serverInfo.isPushBound()) {
                             throw JsSubConsumerAlreadyBound.instance();
                         }
                     }
-                    else { // else they requested a queue but this durable was not configured as queue
+                    else { // else they requested a queue, but this durable was not configured as a queue.
                         throw JsSubExistingConsumerNotQueue.instance();
                     }
                 }
@@ -399,7 +403,7 @@ else if (!listsAreEquivalent(settledFilterSubjects, serverCC.getFilterSubjects()
                     throw JsSubSubjectDoesNotMatchFilter.instance();
                 }
 
-                inboxDeliver = serverCC.getDeliverSubject(); // use the deliver subject as the inbox. It may be null, that's ok, we'll fix that later
+                inboxDeliver = serverCC.getDeliverSubject(); // Use the deliverSubject as the inbox. It may be null, that's ok, we'll fix that later
             }
             else if (so.isBind()) {
                 throw JsSubConsumerNotFoundRequiredInBind.instance();
@@ -418,7 +422,7 @@ else if (inboxDeliver == null) {
             settledInboxDeliver = inboxDeliver;
         }
 
-        // 6. If consumer does not exist, create and settle on the config. Name will have to wait
+        // 6. If the consumer does not exist, create and settle on the config. Name will have to wait
         //    If the consumer exists, I know what the settled info is
         final ConsumerConfiguration settledCC;
         final String settledConsumerName;
@@ -429,20 +433,32 @@ else if (inboxDeliver == null) {
         else {
             ConsumerConfiguration.Builder ccBuilder = ConsumerConfiguration.builder(userCC);
 
-            // Pull mode doesn't maintain a deliver subject. It's actually an error if we send it.
+            // Pull mode doesn't maintain a deliverSubject. It's actually an error if we send it.
             if (!isPullMode) {
                 ccBuilder.deliverSubject(settledInboxDeliver);
             }
 
-            // userCC.filterSubjects might have originally been empty
+            // userCC.filterSubjects might have originally been empty,
             // but there might have been a userSubscribeSubject,
             // so this makes sure it's resolved either way
             ccBuilder.filterSubjects(settledFilterSubjects);
 
             ccBuilder.deliverGroup(settledDeliverGroup);
 
+            if (so.isOrdered()) {
+                // we have to handle the fact that ordered consumers must always have a unique name
+                // if the user supplied a name, well call generateConsumerName with the original name as a prefix
+                if (consumerName != null) {
+                    consumerName = generateConsumerName(userCC.getName());
+                    ccBuilder.name(consumerName);
+                }
+                settledConsumerName = consumerName;
+            }
+            else {
+                settledConsumerName = null; // the server will give us a name if the user's was null
+            }
+
             settledCC = ccBuilder.build();
-            settledConsumerName = null; // the server will give us a name
         }
 
         // 7. create the subscription. lambda needs final or effectively final vars
@@ -481,9 +497,13 @@ else if (inboxDeliver == null) {
         }
 
         // 8. The consumer might need to be created, do it here
-        if (settledConsumerName == null) {
+        if (settledConsumerName == null || so.isOrdered()) {
+            // the _create method sets the consumer name for us
             _createConsumerUnsubscribeOnException(settledStream, settledCC, sub);
         }
+        else {
+            sub.setConsumerName(settledConsumerName);
+        }
 
         return sub;
     }

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

@@ -16,7 +16,6 @@
 import io.nats.client.JetStreamApiException;
 import io.nats.client.JetStreamOptions;
 import io.nats.client.Message;
-import io.nats.client.NUID;
 import io.nats.client.api.*;
 import io.nats.client.support.NatsJetStreamConstants;
 
@@ -28,6 +27,7 @@
 import static io.nats.client.support.NatsConstants.GREATER_THAN;
 import static io.nats.client.support.NatsJetStreamClientError.JsConsumerCreate290NotAvailable;
 import static io.nats.client.support.NatsJetStreamClientError.JsMultipleFilterSubjects210NotAvailable;
+import static io.nats.client.support.NatsJetStreamUtil.generateConsumerName;
 import static io.nats.client.support.NatsRequestCompletableFuture.CancelAction;
 
 class NatsJetStreamImpl implements NatsJetStreamConstants {
@@ -112,7 +112,7 @@ ConsumerInfo _createConsumer(String streamName, ConsumerConfiguration config, Co
                 // if both consumerName and durable are null, generate a name
                 consumerName = durable == null ? generateConsumerName() : durable;
             }
-            String fs = config.getFilterSubject(); // we've already determined not multiple so this gives us 1 or null
+            String fs = config.getFilterSubject(); // we've already determined there are not more than 1 filter subjects, so this gives us one or null
             if (fs == null || fs.equals(GREATER_THAN)) {
                 subj = String.format(JSAPI_CONSUMER_CREATE_V290, streamName, consumerName);
             }
@@ -185,43 +185,40 @@ List<String> _getStreamNames(String subjectFilter) throws IOException, JetStream
     // ----------------------------------------------------------------------------------------------------
     // General Utils
     // ----------------------------------------------------------------------------------------------------
-    String generateConsumerName() {
-        return NUID.nextGlobal();
-    }
-
-    ConsumerConfiguration consumerConfigurationForOrdered(
-        ConsumerConfiguration originalCc,
+    ConsumerConfiguration.Builder consumerConfigurationForOrdered(
+        ConsumerConfiguration initial,
         long lastStreamSeq,
         String newDeliverSubject,
-        String consumerName,
         Long inactiveThreshold)
     {
-        ConsumerConfiguration.Builder builder = ConsumerConfiguration.builder(originalCc).deliverSubject(newDeliverSubject);
+        ConsumerConfiguration.Builder builder = ConsumerConfiguration.builder(initial);
+
+        // push will always give one, pull will always give null
+        if (newDeliverSubject != null) {
+            builder.deliverSubject(newDeliverSubject);
+        }
 
+        // if the last stream seq is > 0, this means this config is for an ordered restart at a sequence
         if (lastStreamSeq > 0) {
             builder
                 .deliverPolicy(DeliverPolicy.ByStartSequence)
                 .startSequence(Math.max(1, lastStreamSeq + 1))
                 .startTime(null); // clear start time in case it was originally set
         }
 
-        if (consumerName != null && consumerCreate290Available) {
-            builder.name(consumerName);
-        }
-
         if (inactiveThreshold != null) {
             builder.inactiveThreshold(inactiveThreshold);
         }
 
-        return builder.build();
+        return builder;
     }
 
     ConsumerInfo lookupConsumerInfo(String streamName, String consumerName) throws IOException, JetStreamApiException {
         try {
             return _getConsumerInfo(streamName, consumerName);
         }
         catch (JetStreamApiException e) {
-            // the right side of this condition...  ( starting here \/ ) is for backward compatibility with server versions that did not provide api error codes
+            // The right side of this condition (after the ||) is for backward compatibility with server versions that did not provide api error codes
             if (e.getApiErrorCode() == JS_CONSUMER_NOT_FOUND_ERR || (e.getErrorCode() == 404 && e.getErrorDescription().contains("consumer"))) {
                 return null;
             }