WIP

scalecube · Sep 1, 2023 · 0f16819 · 0f16819
1 parent ad9db19
commit 0f16819
Show file tree

Hide file tree

Showing 2 changed files with 176 additions and 211 deletions.
diff --git a/cluster/src/main/java/io/scalecube/cluster/TransportWrapper.java b/cluster/src/main/java/io/scalecube/cluster/TransportWrapper.java
@@ -39,7 +39,8 @@ public Mono<Message> requestResponse(Member member, Message request) {
                     int increment = currentIndex.getAndIncrement();
 
                     if (increment == addresses.size()) {
-                      currentIndex.set(increment = 0);
+                      increment = 0;
+                      currentIndex.set(1);
                     }
 
                     final Address address = addresses.get(increment);
@@ -60,13 +61,23 @@ public Mono<Void> send(Member member, Message request) {
     return Mono.defer(
         () -> {
           final List<Address> addresses = member.addresses();
-          final AtomicInteger currentIndex = new AtomicInteger();
+          final int numRetries = addresses.size() - 1;
+          final Integer index = addressIndexByMember.getOrDefault(member, 0);
+          final AtomicInteger currentIndex = new AtomicInteger(index);
+
           return Mono.defer(
                   () -> {
-                    final int index = currentIndex.getAndIncrement();
-                    return transport.send(addresses.get(index), request);
+                    int increment = currentIndex.getAndIncrement();
+
+                    if (increment == addresses.size()) {
+                      increment = 0;
+                      currentIndex.set(1);
+                    }
+
+                    final Address address = addresses.get(increment);
+                    return transport.send(address, request);
                   })
-              .retry(addresses.size() - 1);
+              .retry(numRetries);
         });
   }
 }