Main for server v2.11

.github/workflows/build-pr.yml

@@ -31,6 +31,7 @@ jobs:
           cd $GITHUB_WORKSPACE
           git clone https://github.com/nats-io/nats-server.git
           cd nats-server
+          git checkout consumer-groups
           go get
           go build main.go
           mkdir -p ~/.local/bin

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

@@ -14,16 +14,14 @@
 package io.nats.client;
 
 import io.nats.client.api.ConsumerConfiguration;
-import io.nats.client.support.JsonParseException;
-import io.nats.client.support.JsonParser;
-import io.nats.client.support.JsonSerializable;
-import io.nats.client.support.JsonValue;
+import io.nats.client.support.*;
 
 import static io.nats.client.support.ApiConstants.*;
 import static io.nats.client.support.JsonUtils.*;
 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
@@ -44,17 +42,23 @@ public class BaseConsumeOptions implements JsonSerializable {
     protected final long idleHeartbeat;
     protected final int thresholdPercent;
     protected final boolean noWait;
+    protected final String group;
+    protected final long minPending;
+    protected final long minAckPending;
 
-    @SuppressWarnings("rawtypes") // Don't need the type of the builder to get its vars
-    protected BaseConsumeOptions(Builder b) {
+    protected BaseConsumeOptions(Builder<?, ?> b) {
         bytes = b.bytes;
         if (bytes > 0) {
-            messages = b.messages == -1 ? DEFAULT_MESSAGE_COUNT_WHEN_BYTES : b.messages;
+            messages = b.messages < 0 ? DEFAULT_MESSAGE_COUNT_WHEN_BYTES : b.messages;
         }
         else {
-            messages = b.messages == -1 ? DEFAULT_MESSAGE_COUNT : b.messages;
+            messages = b.messages < 0 ? DEFAULT_MESSAGE_COUNT : b.messages;
         }
 
+        this.group = b.group;
+        this.minPending = b.minPending;
+        this.minAckPending = b.minAckPending;
+
         // validation handled in builder
         thresholdPercent = b.thresholdPercent;
         noWait = b.noWait;
@@ -82,6 +86,9 @@ public String toJson() {
         addField(sb, IDLE_HEARTBEAT, idleHeartbeat);
         addField(sb, THRESHOLD_PERCENT, thresholdPercent);
         addFldWhenTrue(sb, NO_WAIT, noWait);
+        JsonUtils.addField(sb, GROUP, group);
+        JsonUtils.addField(sb, MIN_PENDING, minPending);
+        JsonUtils.addField(sb, MIN_ACK_PENDING, minAckPending);
         return endJson(sb).toString();
     }
 
@@ -101,12 +108,27 @@ public boolean isNoWait() {
         return noWait;
     }
 
+    public String getGroup() {
+        return group;
+    }
+
+    public long getMinPending() {
+        return minPending;
+    }
+
+    public long getMinAckPending() {
+        return minAckPending;
+    }
+
     protected static abstract class Builder<B, CO> {
         protected int messages = -1;
         protected long bytes = 0;
         protected int thresholdPercent = DEFAULT_THRESHOLD_PERCENT;
         protected long expiresIn = DEFAULT_EXPIRES_IN_MILLIS;
         protected boolean noWait = false;
+        protected String group;
+        protected long minPending = -1;
+        protected long minAckPending = -1;
 
         protected abstract B getThis();
 
@@ -137,6 +159,9 @@ public B jsonValue(JsonValue jsonValue) {
             if (readBoolean(jsonValue, NO_WAIT, false)) {
                 noWait();
             }
+            group(readStringEmptyAsNull(jsonValue, GROUP));
+            minPending(readLong(jsonValue, MIN_PENDING, -1));
+            minAckPending(readLong(jsonValue, MIN_ACK_PENDING, -1));
             return getThis();
         }
 
@@ -190,6 +215,36 @@ public B thresholdPercent(int thresholdPercent) {
             return getThis();
         }
 
+        /**
+         * Sets the group
+         * @param group the priority group for this pull
+         * @return Builder
+         */
+        public B group(String group) {
+            this.group = group;
+            return getThis();
+        }
+
+        /**
+         * When specified, the consumer will only receive messages when the consumer has at least this many pending messages.
+         * @param minPending the min pending
+         * @return the builder
+         */
+        public B minPending(long minPending) {
+            this.minPending = minPending < 1 ? -1 : minPending;
+            return getThis();
+        }
+
+        /**
+         * When specified, the consumer will only receive messages when the consumer has at least this many ack pending messages.
+         * @param minAckPending the min ack pending
+         * @return the builder
+         */
+        public B minAckPending(long minAckPending) {
+            this.minAckPending = minAckPending < 1 ? -1 : minAckPending;
+            return getThis();
+        }
+
         /**
          * Build the options.
          * @return the built options

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

@@ -31,13 +31,19 @@ public class PullRequestOptions implements JsonSerializable {
     private final boolean noWait;
     private final Duration expiresIn;
     private final Duration idleHeartbeat;
+    private final String group;
+    private final long minPending;
+    private final long minAckPending;
 
     public PullRequestOptions(Builder b) {
         this.batchSize = b.batchSize;
         this.maxBytes = b.maxBytes;
         this.noWait = b.noWait;
         this.expiresIn = b.expiresIn;
         this.idleHeartbeat = b.idleHeartbeat;
+        this.group = b.group;
+        this.minPending = b.minPending < 0 ? -1 : b.minPending;
+        this.minAckPending = b.minAckPending < 0 ? -1 : b.minAckPending;
     }
 
     @Override
@@ -48,6 +54,10 @@ public String toJson() {
         JsonUtils.addFldWhenTrue(sb, NO_WAIT, noWait);
         JsonUtils.addFieldAsNanos(sb, EXPIRES, expiresIn);
         JsonUtils.addFieldAsNanos(sb, IDLE_HEARTBEAT, idleHeartbeat);
+
+        JsonUtils.addField(sb, GROUP, group);
+        JsonUtils.addField(sb, MIN_PENDING, minPending);
+        JsonUtils.addField(sb, MIN_ACK_PENDING, minAckPending);
         return JsonUtils.endJson(sb).toString();
     }
 
@@ -91,6 +101,18 @@ public Duration getIdleHeartbeat() {
         return idleHeartbeat;
     }
 
+    public String getGroup() {
+        return group;
+    }
+
+    public long getMinPending() {
+        return minPending;
+    }
+
+    public long getMinAckPending() {
+        return minAckPending;
+    }
+
     /**
      * Creates a builder for the pull options, with batch size since it's always required
      * @param batchSize the size of the batch. Must be greater than 0
@@ -115,6 +137,9 @@ public static class Builder {
         private boolean noWait;
         private Duration expiresIn;
         private Duration idleHeartbeat;
+        private String group;
+        private long minPending = -1;
+        private long minAckPending = -1;
 
         /**
          * Set the batch size for the pull
@@ -195,6 +220,37 @@ public Builder idleHeartbeat(Duration idleHeartbeat) {
             return this;
         }
 
+        /**
+         * Sets the group
+         * Replaces any other groups set in the builder
+         * @param group the priority group for this pull
+         * @return Builder
+         */
+        public Builder group(String group) {
+            this.group = group;
+            return this;
+        }
+
+        /**
+         * When specified, the pull request will only receive messages when the consumer has at least this many pending messages.
+         * @param minPending the min pending
+         * @return the builder
+         */
+        public Builder minPending(long minPending) {
+            this.minPending = minPending < 1 ? -1 : minPending;
+            return this;
+        }
+
+        /**
+         * When specified, this Pull request will only receive messages when the consumer has at least this many ack pending messages.
+         * @param minAckPending the min ack pending
+         * @return the builder
+         */
+        public Builder minAckPending(long minAckPending) {
+            this.minAckPending = minAckPending < 1 ? -1 : minAckPending;
+            return this;
+        }
+
         /**
          * Build the PullRequestOptions.
          * <p>Validates that the batch size is greater than 0</p>