Skip to content

Commit fbe942b

Browse files
prashanna-frshprashanna-frsh
authored
Merge branch 'main' into task/add-stream-commands
2 parents ac85afa + 9613118 commit fbe942b

File tree

21 files changed

+831
-207
lines changed

21 files changed

+831
-207
lines changed

‎.github/json_matrices/supported-languages-versions.json‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,8 +11,8 @@
1111
},
1212
{
1313
"language": "node",
14-
"versions": ["16.x", "20.x", "22.x", "23.x"],
15-
"always-run-versions": ["20.x", "23.x"]
14+
"versions": ["20.x", "22.x", "24.x"],
15+
"always-run-versions": ["20.x", "24.x"]
1616
},
1717
{
1818
"language": "go",

‎CHANGELOG.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,10 @@
11
## 2.3
22

3+
#### Breaking Changes
4+
5+
* Node: Remove support for EOL Node.js 16.x and 18.x ([#5276](https://github.com/valkey-io/valkey-glide/issues/5276))
6+
* Node: Drop support for Node.js 16.x and 18.x. Minimum supported version is now Node.js 20.x.
7+
38
#### Changes
49

510
* JAVA: Add EVAL_RO, EVALSHA_RO, and SCRIPT DEBUG commands ([#5125](https://github.com/valkey-io/valkey-glide/pull/5125))

‎examples/node/README.MD‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
1-
### Install node 16 (or newer)
1+
### Install node 20 (or newer)
22
```
3-
curl -s https://deb.nodesource.com/setup_16.x | sudo bash
3+
curl -s https://deb.nodesource.com/setup_20.x | sudo bash
44
apt-get install nodejs npm
5-
npm i -g npm@8
5+
npm i -g npm@10
66
```
77

88
## Install GLIDE package and build the example

‎glide-core/src/iam/mod.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -131,7 +131,7 @@ async fn get_signing_identity(
131131
creds.access_key_id(),
132132
creds.secret_access_key(),
133133
creds.session_token().map(|s| s.to_string()),
134-
None,
134+
creds.expiry(),
135135
service_name,
136136
))
137137
}

‎java/client/src/main/java/glide/api/BaseClient.java‎

Lines changed: 40 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -519,7 +519,9 @@ protected static CommandManager buildCommandManager(ConnectionManager connection
519519
// We'll update this once the connection provides the native handle
520520
GlideCoreClient core =
521521
new GlideCoreClient(
522-
connectionManager.getNativeClientHandle(), connectionManager.getMaxInflightRequests());
522+
connectionManager.getNativeClientHandle(),
523+
connectionManager.getMaxInflightRequests(),
524+
connectionManager.getRequestTimeoutMs());
523525
// Register for PubSub push delivery
524526
try {
525527
GlideCoreClient.registerClient(connectionManager.getNativeClientHandle(), null);
@@ -2703,13 +2705,14 @@ public CompletableFuture<Map<GlideString, Double>> zpopmin(@NonNull GlideString
27032705
@Override
27042706
public CompletableFuture<Object[]> bzpopmin(@NonNull String[] keys, double timeout) {
27052707
String[] arguments = ArrayUtils.add(keys, Double.toString(timeout));
2706-
return commandManager.submitNewCommand(BZPopMin, arguments, this::handleArrayOrNullResponse);
2708+
return commandManager.submitBlockingCommand(
2709+
BZPopMin, arguments, this::handleArrayOrNullResponse);
27072710
}
27082711

27092712
@Override
27102713
public CompletableFuture<Object[]> bzpopmin(@NonNull GlideString[] keys, double timeout) {
27112714
GlideString[] arguments = ArrayUtils.add(keys, gs(Double.toString(timeout)));
2712-
return commandManager.submitNewCommand(
2715+
return commandManager.submitBlockingCommand(
27132716
BZPopMin, arguments, this::handleArrayOrNullResponseBinary);
27142717
}
27152718

@@ -2741,13 +2744,14 @@ public CompletableFuture<Map<GlideString, Double>> zpopmax(@NonNull GlideString
27412744
@Override
27422745
public CompletableFuture<Object[]> bzpopmax(@NonNull String[] keys, double timeout) {
27432746
String[] arguments = ArrayUtils.add(keys, Double.toString(timeout));
2744-
return commandManager.submitNewCommand(BZPopMax, arguments, this::handleArrayOrNullResponse);
2747+
return commandManager.submitBlockingCommand(
2748+
BZPopMax, arguments, this::handleArrayOrNullResponse);
27452749
}
27462750

27472751
@Override
27482752
public CompletableFuture<Object[]> bzpopmax(@NonNull GlideString[] keys, double timeout) {
27492753
GlideString[] arguments = ArrayUtils.add(keys, gs(Double.toString(timeout)));
2750-
return commandManager.submitNewCommand(
2754+
return commandManager.submitBlockingCommand(
27512755
BZPopMax, arguments, this::handleArrayOrNullResponseBinary);
27522756
}
27532757

@@ -3374,13 +3378,20 @@ public CompletableFuture<Map<GlideString, Map<GlideString, GlideString[][]>>> xr
33743378
public CompletableFuture<Map<String, Map<String, String[][]>>> xread(
33753379
@NonNull Map<String, String> keysAndIds, @NonNull StreamReadOptions options) {
33763380
String[] arguments = options.toArgs(keysAndIds);
3381+
if (options.isBlocking()) {
3382+
return commandManager.submitBlockingCommand(XRead, arguments, this::handleXReadResponse);
3383+
}
33773384
return commandManager.submitNewCommand(XRead, arguments, this::handleXReadResponse);
33783385
}
33793386

33803387
@Override
33813388
public CompletableFuture<Map<GlideString, Map<GlideString, GlideString[][]>>> xreadBinary(
33823389
@NonNull Map<GlideString, GlideString> keysAndIds, @NonNull StreamReadOptions options) {
33833390
GlideString[] arguments = options.toArgsBinary(keysAndIds);
3391+
if (options.isBlocking()) {
3392+
return commandManager.submitBlockingCommand(
3393+
XRead, arguments, this::handleXReadResponseBinary);
3394+
}
33843395
return commandManager.submitNewCommand(XRead, arguments, this::handleXReadResponseBinary);
33853396
}
33863397

@@ -3650,6 +3661,9 @@ public CompletableFuture<Map<String, Map<String, String[][]>>> xreadgroup(
36503661
@NonNull String consumer,
36513662
@NonNull StreamReadGroupOptions options) {
36523663
String[] arguments = options.toArgs(group, consumer, keysAndIds);
3664+
if (options.isBlocking()) {
3665+
return commandManager.submitBlockingCommand(XReadGroup, arguments, this::handleXReadResponse);
3666+
}
36533667
return commandManager.submitNewCommand(XReadGroup, arguments, this::handleXReadResponse);
36543668
}
36553669

@@ -3660,6 +3674,10 @@ public CompletableFuture<Map<GlideString, Map<GlideString, GlideString[][]>>> xr
36603674
@NonNull GlideString consumer,
36613675
@NonNull StreamReadGroupOptions options) {
36623676
GlideString[] arguments = options.toArgsBinary(group, consumer, keysAndIds);
3677+
if (options.isBlocking()) {
3678+
return commandManager.submitBlockingCommand(
3679+
XReadGroup, arguments, this::handleXReadResponseBinary);
3680+
}
36633681
return commandManager.submitNewCommand(XReadGroup, arguments, this::handleXReadResponseBinary);
36643682
}
36653683

@@ -4142,14 +4160,14 @@ public CompletableFuture<Long> linsert(
41424160
@Override
41434161
public CompletableFuture<String[]> blpop(@NonNull String[] keys, double timeout) {
41444162
String[] arguments = ArrayUtils.add(keys, Double.toString(timeout));
4145-
return commandManager.submitNewCommand(
4163+
return commandManager.submitBlockingCommand(
41464164
BLPop, arguments, response -> castArray(handleArrayOrNullResponse(response), String.class));
41474165
}
41484166

41494167
@Override
41504168
public CompletableFuture<GlideString[]> blpop(@NonNull GlideString[] keys, double timeout) {
41514169
GlideString[] arguments = ArrayUtils.add(keys, gs(Double.toString(timeout)));
4152-
return commandManager.submitNewCommand(
4170+
return commandManager.submitBlockingCommand(
41534171
BLPop,
41544172
arguments,
41554173
response -> castArray(handleArrayOrNullResponseBinary(response), GlideString.class));
@@ -4158,14 +4176,14 @@ public CompletableFuture<GlideString[]> blpop(@NonNull GlideString[] keys, doubl
41584176
@Override
41594177
public CompletableFuture<String[]> brpop(@NonNull String[] keys, double timeout) {
41604178
String[] arguments = ArrayUtils.add(keys, Double.toString(timeout));
4161-
return commandManager.submitNewCommand(
4179+
return commandManager.submitBlockingCommand(
41624180
BRPop, arguments, response -> castArray(handleArrayOrNullResponse(response), String.class));
41634181
}
41644182

41654183
@Override
41664184
public CompletableFuture<GlideString[]> brpop(@NonNull GlideString[] keys, double timeout) {
41674185
GlideString[] arguments = ArrayUtils.add(keys, gs(Double.toString(timeout)));
4168-
return commandManager.submitNewCommand(
4186+
return commandManager.submitBlockingCommand(
41694187
BRPop,
41704188
arguments,
41714189
response -> castArray(handleArrayOrNullResponseBinary(response), GlideString.class));
@@ -4324,7 +4342,7 @@ public CompletableFuture<Map<String, Object>> bzmpop(
43244342
new String[] {Double.toString(timeout), Integer.toString(keys.length)},
43254343
keys,
43264344
new String[] {modifier.toString()});
4327-
return commandManager.submitNewCommand(
4345+
return commandManager.submitBlockingCommand(
43284346
BZMPop,
43294347
arguments,
43304348
response -> convertKeyValueArrayToMap(handleArrayOrNullResponse(response), Double.class));
@@ -4338,7 +4356,7 @@ public CompletableFuture<Map<GlideString, Object>> bzmpop(
43384356
new GlideString[] {gs(Double.toString(timeout)), gs(Integer.toString(keys.length))},
43394357
keys,
43404358
new GlideString[] {gs(modifier.toString())});
4341-
return commandManager.submitNewCommand(
4359+
return commandManager.submitBlockingCommand(
43424360
BZMPop,
43434361
arguments,
43444362
response ->
@@ -4354,7 +4372,7 @@ public CompletableFuture<Map<String, Object>> bzmpop(
43544372
new String[] {Double.toString(timeout), Integer.toString(keys.length)},
43554373
keys,
43564374
new String[] {modifier.toString(), COUNT_VALKEY_API, Long.toString(count)});
4357-
return commandManager.submitNewCommand(
4375+
return commandManager.submitBlockingCommand(
43584376
BZMPop,
43594377
arguments,
43604378
response -> convertKeyValueArrayToMap(handleArrayOrNullResponse(response), Double.class));
@@ -4370,7 +4388,7 @@ public CompletableFuture<Map<GlideString, Object>> bzmpop(
43704388
new GlideString[] {
43714389
gs(modifier.toString()), gs(COUNT_VALKEY_API), gs(Long.toString(count))
43724390
});
4373-
return commandManager.submitNewCommand(
4391+
return commandManager.submitBlockingCommand(
43744392
BZMPop,
43754393
arguments,
43764394
response ->
@@ -4768,7 +4786,7 @@ public CompletableFuture<Map<String, String[]>> blmpop(
47684786
new String[] {Double.toString(timeout), Long.toString(keys.length)},
47694787
keys,
47704788
new String[] {direction.toString(), COUNT_FOR_LIST_VALKEY_API, Long.toString(count)});
4771-
return commandManager.submitNewCommand(
4789+
return commandManager.submitBlockingCommand(
47724790
BLMPop,
47734791
arguments,
47744792
response -> castMapOfArrays(handleMapOrNullResponse(response), String.class));
@@ -4784,7 +4802,7 @@ public CompletableFuture<Map<GlideString, GlideString[]>> blmpop(
47844802
new GlideString[] {
47854803
gs(direction.toString()), gs(COUNT_FOR_LIST_VALKEY_API), gs(Long.toString(count))
47864804
});
4787-
return commandManager.submitNewCommand(
4805+
return commandManager.submitBlockingCommand(
47884806
BLMPop,
47894807
arguments,
47904808
response ->
@@ -4800,7 +4818,7 @@ public CompletableFuture<Map<String, String[]>> blmpop(
48004818
new String[] {Double.toString(timeout), Long.toString(keys.length)},
48014819
keys,
48024820
new String[] {direction.toString()});
4803-
return commandManager.submitNewCommand(
4821+
return commandManager.submitBlockingCommand(
48044822
BLMPop,
48054823
arguments,
48064824
response -> castMapOfArrays(handleMapOrNullResponse(response), String.class));
@@ -4814,7 +4832,7 @@ public CompletableFuture<Map<GlideString, GlideString[]>> blmpop(
48144832
new GlideString[] {gs(Double.toString(timeout)), gs(Long.toString(keys.length))},
48154833
keys,
48164834
new GlideString[] {gs(direction.toString())});
4817-
return commandManager.submitNewCommand(
4835+
return commandManager.submitBlockingCommand(
48184836
BLMPop,
48194837
arguments,
48204838
response ->
@@ -4868,7 +4886,8 @@ public CompletableFuture<String> blmove(
48684886
new String[] {
48694887
source, destination, wherefrom.toString(), whereto.toString(), Double.toString(timeout)
48704888
};
4871-
return commandManager.submitNewCommand(BLMove, arguments, this::handleStringOrNullResponse);
4889+
return commandManager.submitBlockingCommand(
4890+
BLMove, arguments, this::handleStringOrNullResponse);
48724891
}
48734892

48744893
@Override
@@ -4886,7 +4905,7 @@ public CompletableFuture<GlideString> blmove(
48864905
gs(whereto.toString()),
48874906
gs(Double.toString(timeout))
48884907
};
4889-
return commandManager.submitNewCommand(
4908+
return commandManager.submitBlockingCommand(
48904909
BLMove, arguments, this::handleGlideStringOrNullResponse);
48914910
}
48924911

@@ -6055,15 +6074,15 @@ public CompletableFuture<Object[]> hscan(
60556074

60566075
@Override
60576076
public CompletableFuture<Long> wait(long numreplicas, long timeout) {
6058-
return commandManager.submitNewCommand(
6077+
return commandManager.submitBlockingCommand(
60596078
Wait,
60606079
new String[] {Long.toString(numreplicas), Long.toString(timeout)},
60616080
this::handleLongResponse);
60626081
}
60636082

60646083
@Override
60656084
public CompletableFuture<Long[]> waitaof(long numlocal, long numreplicas, long timeout) {
6066-
return commandManager.submitNewCommand(
6085+
return commandManager.submitBlockingCommand(
60676086
WaitAof,
60686087
new String[] {Long.toString(numlocal), Long.toString(numreplicas), Long.toString(timeout)},
60696088
response -> castArray(handleArrayResponse(response), Long.class));

‎java/client/src/main/java/glide/api/GlideClient.java‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77
import static command_request.CommandRequestOuterClass.RequestType.ConfigResetStat;
88
import static command_request.CommandRequestOuterClass.RequestType.ConfigRewrite;
99
import static command_request.CommandRequestOuterClass.RequestType.ConfigSet;
10-
import static command_request.CommandRequestOuterClass.RequestType.CustomCommand;
1110
import static command_request.CommandRequestOuterClass.RequestType.DBSize;
1211
import static command_request.CommandRequestOuterClass.RequestType.Echo;
1312
import static command_request.CommandRequestOuterClass.RequestType.FlushAll;
@@ -136,13 +135,12 @@ public static CompletableFuture<GlideClient> createClient(
136135

137136
@Override
138137
public CompletableFuture<Object> customCommand(@NonNull String[] args) {
139-
return commandManager.submitNewCommand(CustomCommand, args, this::handleObjectOrNullResponse);
138+
return commandManager.submitCustomCommand(args, this::handleObjectOrNullResponse);
140139
}
141140

142141
@Override
143142
public CompletableFuture<Object> customCommand(@NonNull GlideString[] args) {
144-
return commandManager.submitNewCommand(
145-
CustomCommand, args, this::handleBinaryObjectOrNullResponse);
143+
return commandManager.submitCustomCommand(args, this::handleBinaryObjectOrNullResponse);
146144
}
147145

148146
@Deprecated

‎java/client/src/main/java/glide/api/GlideClusterClient.java‎

Lines changed: 10 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77
import static command_request.CommandRequestOuterClass.RequestType.ConfigResetStat;
88
import static command_request.CommandRequestOuterClass.RequestType.ConfigRewrite;
99
import static command_request.CommandRequestOuterClass.RequestType.ConfigSet;
10-
import static command_request.CommandRequestOuterClass.RequestType.CustomCommand;
1110
import static command_request.CommandRequestOuterClass.RequestType.DBSize;
1211
import static command_request.CommandRequestOuterClass.RequestType.Echo;
1312
import static command_request.CommandRequestOuterClass.RequestType.FCall;
@@ -167,31 +166,29 @@ public static CompletableFuture<GlideClusterClient> createClient(
167166
@Override
168167
public CompletableFuture<ClusterValue<Object>> customCommand(@NonNull String[] args) {
169168
// TODO if a command returns a map as a single value, ClusterValue misleads user
170-
return commandManager.submitNewCommand(
171-
CustomCommand, args, response -> ClusterValue.of(handleObjectOrNullResponse(response)));
169+
return commandManager.submitCustomCommand(
170+
args, response -> ClusterValue.of(handleObjectOrNullResponse(response)));
172171
}
173172

174173
@Override
175174
public CompletableFuture<ClusterValue<Object>> customCommand(@NonNull GlideString[] args) {
176175
// TODO if a command returns a map as a single value, ClusterValue misleads user
177-
return commandManager.submitNewCommand(
178-
CustomCommand,
179-
args,
180-
response -> ClusterValue.of(handleBinaryObjectOrNullResponse(response)));
176+
return commandManager.submitCustomCommand(
177+
args, response -> ClusterValue.of(handleBinaryObjectOrNullResponse(response)));
181178
}
182179

183180
@Override
184181
public CompletableFuture<ClusterValue<Object>> customCommand(
185182
@NonNull String[] args, @NonNull Route route) {
186-
return commandManager.submitNewCommand(
187-
CustomCommand, args, route, response -> handleCustomCommandResponse(route, response));
183+
return commandManager.submitCustomCommand(
184+
args, route, response -> handleCustomCommandResponse(route, response));
188185
}
189186

190187
@Override
191188
public CompletableFuture<ClusterValue<Object>> customCommand(
192189
@NonNull GlideString[] args, @NonNull Route route) {
193-
return commandManager.submitNewCommand(
194-
CustomCommand, args, route, response -> handleCustomCommandBinaryResponse(route, response));
190+
return commandManager.submitCustomCommand(
191+
args, route, response -> handleCustomCommandBinaryResponse(route, response));
195192
}
196193

197194
@SuppressWarnings("unchecked")
@@ -1350,7 +1347,7 @@ public CompletableFuture<ClusterValue<GlideString[]>> keys(
13501347
@Override
13511348
public CompletableFuture<Long> wait(long numreplicas, long timeout) {
13521349
String[] arguments = new String[] {Long.toString(numreplicas), Long.toString(timeout)};
1353-
return commandManager.submitNewCommand(
1350+
return commandManager.submitBlockingCommand(
13541351
Wait, arguments, SimpleSingleNodeRoute.RANDOM, this::handleLongResponse);
13551352
}
13561353

@@ -1359,7 +1356,7 @@ public CompletableFuture<Long[]> waitaof(
13591356
long numlocal, long numreplicas, long timeout, @NonNull Route route) {
13601357
String[] arguments =
13611358
new String[] {Long.toString(numlocal), Long.toString(numreplicas), Long.toString(timeout)};
1362-
return commandManager.submitNewCommand(
1359+
return commandManager.submitBlockingCommand(
13631360
WaitAof,
13641361
arguments,
13651362
route,

‎java/client/src/main/java/glide/api/models/commands/stream/StreamReadOptions.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,20 @@ public class StreamReadOptions {
3636
*/
3737
protected Long count;
3838

39+
/**
40+
* Returns true if this options object specifies a BLOCK timeout, making the command a blocking
41+
* command.
42+
*
43+
* <p>Note: We check {@code block != null} rather than {@code block != null && block != 0} because
44+
* {@code BLOCK 0} means "block indefinitely" in Valkey/Redis, which is still a blocking command
45+
* that should skip Java-side timeout enforcement.
46+
*
47+
* @return true if BLOCK option is set (including BLOCK 0 for indefinite blocking)
48+
*/
49+
public boolean isBlocking() {
50+
return this.block != null;
51+
}
52+
3953
/**
4054
* Converts options and the key-to-id input for {@link StreamBaseCommands#xread(Map,
4155
* StreamReadOptions)} into a String[].

0 commit comments

Comments
 (0)