WIP

scalecube · Sep 1, 2023 · ad9db19 · ad9db19
1 parent 93cf26a
commit ad9db19
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 3 deletions.
diff --git a/cluster/src/main/java/io/scalecube/cluster/TransportWrapper.java b/cluster/src/main/java/io/scalecube/cluster/TransportWrapper.java
@@ -4,13 +4,17 @@
 import io.scalecube.cluster.transport.api.Transport;
 import io.scalecube.net.Address;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import reactor.core.publisher.Mono;
 
 public class TransportWrapper {
 
   private final Transport transport;
 
+  private final Map<Member, Integer> addressIndexByMember = new ConcurrentHashMap<>();
+
   public TransportWrapper(Transport transport) {
     this.transport = transport;
   }
@@ -26,13 +30,22 @@ public Mono<Message> requestResponse(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 Address address = addresses.get(currentIndex.getAndIncrement());
+                    int increment = currentIndex.getAndIncrement();
+
+                    if (increment == addresses.size()) {
+                      currentIndex.set(increment = 0);
+                    }
+
+                    final Address address = addresses.get(increment);
                     return transport.requestResponse(address, request);
                   })
-              .retry(addresses.size() - 1);
+              .retry(numRetries);
         });
   }
 

diff --git a/cluster/src/test/java/io/scalecube/cluster/TransportWrapperTest.java b/cluster/src/test/java/io/scalecube/cluster/TransportWrapperTest.java
@@ -50,6 +50,20 @@ public TransportWrapperTest() {
   @Nested
   class RequestResponseTests {
 
+    @Test
+    void requestResponseShouldWorkIndex2() {
+      final List<Address> addresses = Collections.singletonList(Address.from("test:0"));
+      final Member member = new Member("test", null, addresses, "namespace");
+
+      when(transport.requestResponse(addresses.get(0), request)).thenReturn(Mono.just(response));
+
+      StepVerifier.create(transportWrapper.requestResponse(member, request))
+        .assertNext(message -> Assertions.assertSame(response, message, "response"))
+        .thenCancel()
+        .verify();
+
+    }
+
     @Test
     void requestResponseShouldWork() {
       final List<Address> addresses = Collections.singletonList(Address.from("test:0"));