Skip to content

Commit fcb126a

Browse files
scottfscottf
authored
Merge pull request #1239 from nats-io/main-2-11
Main for server v2.11
2 parents eecb254 + 7dfd719 commit fcb126a

29 files changed

+1156
-195
lines changed

‎README.md

-2
Original file line numberDiff line numberDiff line change
@@ -979,8 +979,6 @@ You can however set the deliver policy which will be used to start the subscript
979979
| JsConsumerCreate290NotAvailable | CON-90301 | Name field not valid when v2.9.0 consumer create api is not available. |
980980
| JsConsumerNameDurableMismatch | CON-90302 | Name must match durable if both are supplied. |
981981
| JsMultipleFilterSubjects210NotAvailable | CON-90303 | Multiple filter subjects not available until server version 2.10.0. |
982-
| JsAllowDirectRequired | CON-90304 | Stream must have allow direct set. |
983-
| JsDirectBatchGet211NotAvailable | CON-90305 | Batch direct get not available until server version 2.11.0. |
984982
| OsObjectNotFound | OS-90201 | The object was not found. |
985983
| OsObjectIsDeleted | OS-90202 | The object is deleted. |
986984
| OsObjectAlreadyExists | OS-90203 | An object with that name already exists. |

‎build.gradle

+1-1
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ plugins {
1313
id 'signing'
1414
}
1515

16-
def jarVersion = "2.20.7"
16+
def jarVersion = "2.21.0"
1717

1818
def isRelease = System.getenv("BUILD_EVENT") == "release"
1919
def brn = System.getenv("BRANCH_REF_NAME")

‎dependencies.md

+1-4
Original file line numberDiff line numberDiff line change
@@ -14,22 +14,19 @@ This file lists the dependencies used in this repository.
1414
| Dependency | License |
1515
|---------------------------------------|-------------|
1616
| org.bouncycastle:bcprov-lts8on:2.73.7 | MIT License |
17+
1718
#### Test Dependencies
1819

1920
| Dependency | License |
2021
|-------------------------------------------------|-----------------------------------------|
2122
| io.nats:jnats-server-runner:1.2.5 | Apache 2.0 License |
2223
| org.apiguardian:apiguardian-api:1.1.0 | Apache 2.0 License |
2324
| org.junit.jupiter:junit-jupiter:5.9.0 | Eclipse Public License v2.0 |
24-
| org.junit:junit-bom:5.9.0 | Eclipse Public License v2.0 |
2525
| org.junit.jupiter:junit-jupiter:5.9.0 | Eclipse Public License v2.0 |
2626
| org.junit.jupiter:junit-jupiter-api:5.9.0 | Eclipse Public License v2.0 |
2727
| org.junit.jupiter:junit-jupiter-api:5.9.0 | Eclipse Public License v2.0 |
2828
| org.junit.jupiter:junit-jupiter-engine:5.9.0 | Eclipse Public License v2.0 |
2929
| org.junit.jupiter:junit-jupiter-params:5.9.0 | Eclipse Public License v2.0 |
30-
| org.junit.platform:junit-platform-commons:1.9.0 | Eclipse Public License v2.0 |
31-
| org.junit.platform:junit-platform-engine:1.9.0 | Eclipse Public License v2.0 |
32-
| org.opentest4j:opentest4j:1.2.0 | Apache 2.0 License |
3330

3431
#### Build / Coverage Dependencies
3532

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

+71-25
Original file line numberDiff line numberDiff line change
@@ -13,16 +13,17 @@
1313

1414
package io.nats.client;
1515

16-
import io.nats.client.api.ConsumerConfiguration;
1716
import io.nats.client.support.JsonParseException;
1817
import io.nats.client.support.JsonParser;
1918
import io.nats.client.support.JsonSerializable;
2019
import io.nats.client.support.JsonValue;
2120

2221
import static io.nats.client.support.ApiConstants.*;
2322
import static io.nats.client.support.JsonUtils.*;
23+
import static io.nats.client.support.JsonValueUtils.readBoolean;
2424
import static io.nats.client.support.JsonValueUtils.readInteger;
2525
import static io.nats.client.support.JsonValueUtils.readLong;
26+
import static io.nats.client.support.JsonValueUtils.*;
2627

2728
/**
2829
* Base Consume Options are provided to customize the way the consume and
@@ -40,13 +41,15 @@ public class BaseConsumeOptions implements JsonSerializable {
4041
protected final int messages;
4142
protected final long bytes;
4243
protected final long expiresIn;
43-
protected final long idleHeartbeat;
4444
protected final int thresholdPercent;
45-
protected final boolean noWait;
45+
protected final long idleHeartbeat;
46+
protected final String group;
47+
protected final long minPending;
48+
protected final long minAckPending;
4649
protected final boolean raiseStatusWarnings;
4750

48-
@SuppressWarnings("rawtypes") // Don't need the type of the builder to get its vars
49-
protected BaseConsumeOptions(Builder b) {
51+
protected BaseConsumeOptions(Builder<?, ?> b) {
52+
// Message / bytes is part of base and is calculated
5053
bytes = b.bytes;
5154
if (bytes > 0) {
5255
messages = b.messages < 0 ? DEFAULT_MESSAGE_COUNT_WHEN_BYTES : b.messages;
@@ -55,23 +58,17 @@ protected BaseConsumeOptions(Builder b) {
5558
messages = b.messages < 0 ? DEFAULT_MESSAGE_COUNT : b.messages;
5659
}
5760

58-
// validation handled in builder
61+
// Validation for expiresIn, if any extra, is handled in subclass builder
62+
expiresIn = b.expiresIn;
5963
thresholdPercent = b.thresholdPercent;
60-
noWait = b.noWait;
61-
raiseStatusWarnings = b.raiseStatusWarnings;
6264

63-
// if it's not noWait, it must have an expiresIn
64-
// we can't check this in the builder because we can't guarantee order
65-
// so we always default to LONG_UNSET in the builder and check it here.
66-
if (b.expiresIn == ConsumerConfiguration.LONG_UNSET && !noWait) {
67-
expiresIn = DEFAULT_EXPIRES_IN_MILLIS;
68-
}
69-
else {
70-
expiresIn = b.expiresIn;
71-
}
72-
73-
// calculated
65+
// 3. idleHeartbeat is part of base and is calculated.
7466
idleHeartbeat = Math.min(MAX_HEARTBEAT_MILLIS, expiresIn * MAX_IDLE_HEARTBEAT_PERCENT / 100);
67+
68+
this.group = b.group;
69+
this.minPending = b.minPending;
70+
this.minAckPending = b.minAckPending;
71+
raiseStatusWarnings = b.raiseStatusWarnings;
7572
}
7673

7774
@Override
@@ -82,11 +79,16 @@ public String toJson() {
8279
addField(sb, EXPIRES_IN, expiresIn);
8380
addField(sb, IDLE_HEARTBEAT, idleHeartbeat);
8481
addField(sb, THRESHOLD_PERCENT, thresholdPercent);
82+
addField(sb, GROUP, group);
83+
addField(sb, MIN_PENDING, minPending);
84+
addField(sb, MIN_ACK_PENDING, minAckPending);
8585
addFldWhenTrue(sb, RAISE_STATUS_WARNINGS, raiseStatusWarnings);
86-
addFldWhenTrue(sb, NO_WAIT, noWait);
86+
subclassSpecificToJson(sb);
8787
return endJson(sb).toString();
8888
}
8989

90+
protected void subclassSpecificToJson(StringBuilder sb) {}
91+
9092
public long getExpiresInMillis() {
9193
return expiresIn;
9294
}
@@ -99,21 +101,31 @@ public int getThresholdPercent() {
99101
return thresholdPercent;
100102
}
101103

102-
public boolean isNoWait() {
103-
return noWait;
104-
}
105-
106104
public boolean raiseStatusWarnings() {
107105
return raiseStatusWarnings;
108106
}
109107

108+
public String getGroup() {
109+
return group;
110+
}
111+
112+
public long getMinPending() {
113+
return minPending;
114+
}
115+
116+
public long getMinAckPending() {
117+
return minAckPending;
118+
}
119+
110120
protected static abstract class Builder<B, CO> {
111121
protected int messages = -1;
112122
protected long bytes = 0;
113123
protected int thresholdPercent = DEFAULT_THRESHOLD_PERCENT;
114124
protected long expiresIn = DEFAULT_EXPIRES_IN_MILLIS;
115-
protected boolean noWait = false;
116125
protected boolean raiseStatusWarnings = false;
126+
protected String group;
127+
protected long minPending = -1;
128+
protected long minAckPending = -1;
117129

118130
protected abstract B getThis();
119131

@@ -137,6 +149,10 @@ public B jsonValue(JsonValue jsonValue) {
137149
bytes(readLong(jsonValue, BYTES, -1));
138150
expiresIn(readLong(jsonValue, EXPIRES_IN, MIN_EXPIRES_MILLS));
139151
thresholdPercent(readInteger(jsonValue, THRESHOLD_PERCENT, -1));
152+
raiseStatusWarnings(readBoolean(jsonValue, RAISE_STATUS_WARNINGS, false));
153+
group(readStringEmptyAsNull(jsonValue, GROUP));
154+
minPending(readLong(jsonValue, MIN_PENDING, -1));
155+
minAckPending(readLong(jsonValue, MIN_ACK_PENDING, -1));
140156
return getThis();
141157
}
142158

@@ -210,6 +226,36 @@ public B raiseStatusWarnings(boolean raiseStatusWarnings) {
210226
return getThis();
211227
}
212228

229+
/**
230+
* Sets the group
231+
* @param group the priority group for this pull
232+
* @return Builder
233+
*/
234+
public B group(String group) {
235+
this.group = group;
236+
return getThis();
237+
}
238+
239+
/**
240+
* When specified, the consumer will only receive messages when the consumer has at least this many pending messages.
241+
* @param minPending the min pending
242+
* @return the builder
243+
*/
244+
public B minPending(long minPending) {
245+
this.minPending = minPending < 1 ? -1 : minPending;
246+
return getThis();
247+
}
248+
249+
/**
250+
* When specified, the consumer will only receive messages when the consumer has at least this many ack pending messages.
251+
* @param minAckPending the min ack pending
252+
* @return the builder
253+
*/
254+
public B minAckPending(long minAckPending) {
255+
this.minAckPending = minAckPending < 1 ? -1 : minAckPending;
256+
return getThis();
257+
}
258+
213259
/**
214260
* Build the options.
215261
* @return the built options

‎src/main/java/io/nats/client/FetchConsumeOptions.java

+15
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818

1919
import static io.nats.client.support.ApiConstants.EXPIRES_IN;
2020
import static io.nats.client.support.ApiConstants.NO_WAIT;
21+
import static io.nats.client.support.JsonUtils.addFldWhenTrue;
2122
import static io.nats.client.support.JsonValueUtils.readBoolean;
2223
import static io.nats.client.support.JsonValueUtils.readLong;
2324

@@ -27,8 +28,16 @@
2728
public class FetchConsumeOptions extends BaseConsumeOptions {
2829
public static FetchConsumeOptions DEFAULT_FETCH_OPTIONS = FetchConsumeOptions.builder().build();
2930

31+
private final boolean noWait;
32+
3033
private FetchConsumeOptions(Builder b) {
3134
super(b);
35+
this.noWait = b.noWait;
36+
}
37+
38+
@Override
39+
protected void subclassSpecificToJson(StringBuilder sb) {
40+
addFldWhenTrue(sb, NO_WAIT, noWait);
3241
}
3342

3443
/**
@@ -47,13 +56,19 @@ public long getMaxBytes() {
4756
return bytes;
4857
}
4958

59+
public boolean isNoWait() {
60+
return noWait;
61+
}
62+
5063
public static Builder builder() {
5164
return new Builder();
5265
}
5366

5467
public static class Builder
5568
extends BaseConsumeOptions.Builder<Builder, FetchConsumeOptions> {
5669

70+
protected boolean noWait = false;
71+
5772
protected Builder getThis() { return this; }
5873

5974
@Override

‎src/main/java/io/nats/client/JetStreamManagement.java

+38
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,17 @@ public interface JetStreamManagement {
260260
*/
261261
MessageInfo getMessage(String streamName, long seq) throws IOException, JetStreamApiException;
262262

263+
/**
264+
* Get MessageInfo for the message matching the {@link MessageGetRequest}.
265+
* @param streamName the name of the stream.
266+
* @param messageGetRequest the {@link MessageGetRequest} to get a message
267+
* @return The MessageInfo
268+
* @throws IOException covers various communication issues with the NATS
269+
* server such as timeout or interruption
270+
* @throws JetStreamApiException the request had an error related to the data
271+
*/
272+
MessageInfo getMessage(String streamName, MessageGetRequest messageGetRequest) throws IOException, JetStreamApiException;
273+
263274
/**
264275
* Get MessageInfo for the last message of the subject.
265276
* @param streamName the name of the stream.
@@ -282,6 +293,33 @@ public interface JetStreamManagement {
282293
*/
283294
MessageInfo getFirstMessage(String streamName, String subject) throws IOException, JetStreamApiException;
284295

296+
/**
297+
* Get MessageInfo for the first message created at or after the start time.
298+
* <p>
299+
* This API 1) is currently EXPERIMENTAL and is subject to change. 2) Works on Server 2.11 or later
300+
* @param streamName the name of the stream.
301+
* @param startTime the start time to get the first message for.
302+
* @return The MessageInfo
303+
* @throws IOException covers various communication issues with the NATS
304+
* server such as timeout or interruption
305+
* @throws JetStreamApiException the request had an error related to the data
306+
*/
307+
MessageInfo getFirstMessage(String streamName, ZonedDateTime startTime) throws IOException, JetStreamApiException;
308+
309+
/**
310+
* Get MessageInfo for the first message created at or after the start time matching the subject.
311+
* <p>
312+
* This API 1) is currently EXPERIMENTAL and is subject to change. 2) Works on Server 2.11 or later
313+
* @param streamName the name of the stream.
314+
* @param startTime the start time to get the first message for.
315+
* @param subject the subject to get the first message for.
316+
* @return The MessageInfo
317+
* @throws IOException covers various communication issues with the NATS
318+
* server such as timeout or interruption
319+
* @throws JetStreamApiException the request had an error related to the data
320+
*/
321+
MessageInfo getFirstMessage(String streamName, ZonedDateTime startTime, String subject) throws IOException, JetStreamApiException;
322+
285323
/**
286324
* Get MessageInfo for the message of the message sequence
287325
* is equal to or greater the requested sequence for the subject.

‎src/main/java/io/nats/client/MessageInfoHandler.java

+1-2
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,6 @@ public interface MessageInfoHandler {
2323
* Called to deliver a {@link MessageInfo} to the handler.
2424
*
2525
* @param messageInfo the received {@link MessageInfo}
26-
* @throws InterruptedException if the thread for this handler is interrupted
2726
*/
28-
void onMessageInfo(MessageInfo messageInfo) throws InterruptedException;
27+
void onMessageInfo(MessageInfo messageInfo);
2928
}

0 commit comments

Comments
 (0)