Skip to content

Commit 06a7c31

Browse files
committed
Merge remote-tracking branch 'origin/refactor-reactive-mget-method' into refactor-reactive-mget-method
2 parents 9d7dddb + bcb2885 commit 06a7c31

File tree

1 file changed

+26
-0
lines changed

1 file changed

+26
-0
lines changed

src/main/java/io/lettuce/core/cluster/RedisAdvancedClusterReactiveCommandsImpl.java

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -264,6 +264,7 @@ public Flux<KeyValue<K, V>> mget(Iterable<K> keys) {
264264
return super.mget(keyList);
265265
}
266266

267+
<<<<<<< HEAD
267268
List<Publisher<KeyValue<K, V>>> publishers = partitioned.values().stream().map(super::mget)
268269
.collect(Collectors.toList());
269270

@@ -283,6 +284,31 @@ public Flux<KeyValue<K, V>> mget(Iterable<K> keys) {
283284

284285
return Arrays.asList(values);
285286
}).flatMapMany(Flux::fromIterable);
287+
=======
288+
List<Publisher<KeyValue<K, V>>> publishers = partitioned.values().stream()
289+
.map(super::mget)
290+
.collect(Collectors.toList());
291+
292+
return Flux.mergeSequential(publishers)
293+
.collectList()
294+
.map(results -> {
295+
KeyValue<K, V>[] values = new KeyValue[keyList.size()];
296+
int offset = 0;
297+
298+
for (List<K> partitionKeys : partitioned.values()) {
299+
for (int i = 0; i < keyList.size(); i++) {
300+
int index = partitionKeys.indexOf(keyList.get(i));
301+
if (index != -1) {
302+
values[i] = results.get(offset + index);
303+
}
304+
}
305+
offset += partitionKeys.size();
306+
}
307+
308+
return Arrays.asList(values);
309+
})
310+
.flatMapMany(Flux::fromIterable);
311+
>>>>>>> origin/refactor-reactive-mget-method
286312
}
287313

288314
@Override

0 commit comments

Comments
 (0)