Move MRM content parsing outside of “withBinaryClusterReactive”
This commit is contained in:
parent
0e552bd602
commit
e25291c74c
|
@ -443,7 +443,7 @@ public class MessagesCache extends RedisClusterPubSubAdapter<String, String> imp
|
||||||
final Mono<MessageProtos.Envelope> mrmMessageMono = Mono.from(redisCluster.withBinaryClusterReactive(
|
final Mono<MessageProtos.Envelope> mrmMessageMono = Mono.from(redisCluster.withBinaryClusterReactive(
|
||||||
conn -> conn.reactive().hmget(key, "data".getBytes(StandardCharsets.UTF_8), sharedMrmViewKey)
|
conn -> conn.reactive().hmget(key, "data".getBytes(StandardCharsets.UTF_8), sharedMrmViewKey)
|
||||||
.collectList()
|
.collectList()
|
||||||
.publishOn(messageDeliveryScheduler)
|
.publishOn(messageDeliveryScheduler)))
|
||||||
.handle((mrmDataAndView, sink) -> {
|
.handle((mrmDataAndView, sink) -> {
|
||||||
try {
|
try {
|
||||||
assert mrmDataAndView.size() == 2;
|
assert mrmDataAndView.size() == 2;
|
||||||
|
@ -461,7 +461,7 @@ public class MessagesCache extends RedisClusterPubSubAdapter<String, String> imp
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
sink.error(e);
|
sink.error(e);
|
||||||
}
|
}
|
||||||
})));
|
});
|
||||||
|
|
||||||
experiment.compareFutureResult(mrmMessage.toBuilder().clearSharedMrmKey().build(),
|
experiment.compareFutureResult(mrmMessage.toBuilder().clearSharedMrmKey().build(),
|
||||||
mrmMessageMono.toFuture());
|
mrmMessageMono.toFuture());
|
||||||
|
|
Loading…
Reference in New Issue