Added cluster-tests

scalecube · Aug 31, 2023 · b8e42b4 · b8e42b4
1 parent 76d1576
commit b8e42b4
Show file tree

Hide file tree

Showing 18 changed files with 113 additions and 46 deletions.
diff --git a/cluster-testlib/src/main/java/io/scalecube/cluster/utils/NetworkEmulator.java b/cluster-testlib/src/main/java/io/scalecube/cluster/utils/NetworkEmulator.java
@@ -244,10 +244,11 @@ public InboundSettings inboundSettings(List<Address> destinations) {
   public void inboundSettings(List<Address> destinations, boolean shallPass) {
     InboundSettings settings = new InboundSettings(shallPass);
 
-    destinations.forEach(destination -> {
-      inboundSettings.put(destination, settings);
-      LOGGER.debug("[{}] Set inbound settings {} to {}", address, settings, destination);
-    });
+    destinations.forEach(
+        destination -> {
+          inboundSettings.put(destination, settings);
+          LOGGER.debug("[{}] Set inbound settings {} to {}", address, settings, destination);
+        });
   }
 
   /**

diff --git a/cluster-tests/pom.xml b/cluster-tests/pom.xml
@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>io.scalecube</groupId>
+    <artifactId>scalecube-cluster-parent</artifactId>
+    <version>2.6.18-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>cluster-tests</artifactId>
+
+  <dependencies>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>scalecube-cluster-testlib</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>scalecube-transport-netty</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-slf4j-impl</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-core</artifactId>
+    </dependency>
+  </dependencies>
+
+</project>
diff --git a/...o/scalecube/transport/netty/BaseTest.java → ...o/scalecube/transport/netty/BaseTest.java b/...o/scalecube/transport/netty/BaseTest.java → ...o/scalecube/transport/netty/BaseTest.java
@@ -10,7 +10,6 @@
 import io.scalecube.transport.netty.websocket.WebsocketTransportFactory;
 import java.time.Duration;
 import java.util.List;
-
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.TestInfo;
@@ -62,14 +61,14 @@ protected Mono<Void> send(Transport transport, Address to, Message msg) {
    */
   protected Mono<Void> send(Transport transport, List<Address> to, Message msg) {
     return TransportWrapper.send(transport, to, msg)
-      .doOnError(
-        th ->
-          LOGGER.error(
-            "Failed to send {} to {} from transport: {}, cause: {}",
-            msg,
-            to,
-            transport,
-            th.toString()));
+        .doOnError(
+            th ->
+                LOGGER.error(
+                    "Failed to send {} to {} from transport: {}, cause: {}",
+                    msg,
+                    to,
+                    transport,
+                    th.toString()));
   }
 
   /**

diff --git a/.../netty/tcp/TcpTransportSendOrderTest.java → .../netty/tcp/TcpTransportSendOrderTest.java b/.../netty/tcp/TcpTransportSendOrderTest.java → .../netty/tcp/TcpTransportSendOrderTest.java
diff --git a/...transport/netty/tcp/TcpTransportTest.java → ...transport/netty/tcp/TcpTransportTest.java b/...transport/netty/tcp/TcpTransportTest.java → ...transport/netty/tcp/TcpTransportTest.java
diff --git a/...cket/WebsocketTransportSendOrderTest.java → ...cket/WebsocketTransportSendOrderTest.java b/...cket/WebsocketTransportSendOrderTest.java → ...cket/WebsocketTransportSendOrderTest.java
diff --git a/...tty/websocket/WebsocketTransportTest.java → ...tty/websocket/WebsocketTransportTest.java b/...tty/websocket/WebsocketTransportTest.java → ...tty/websocket/WebsocketTransportTest.java
diff --git a/...-netty/src/test/resources/log4j2-test.xml → ...-tests/src/test/resources/log4j2-test.xml b/...-netty/src/test/resources/log4j2-test.xml → ...-tests/src/test/resources/log4j2-test.xml
diff --git a/cluster/src/main/java/io/scalecube/cluster/TransportWrapper.java b/cluster/src/main/java/io/scalecube/cluster/TransportWrapper.java
@@ -19,6 +19,13 @@ public TransportWrapper(Transport transport) {
     this.transport = transport;
   }
 
+  /**
+   * Execute request response call.
+   *
+   * @param member member
+   * @param request request
+   * @return mono result
+   */
   public Mono<Message> requestResponse(Member member, Message request) {
     return connections
         .compute(
@@ -36,6 +43,28 @@ public Mono<Message> requestResponse(Member member, Message request) {
         .map(result -> result.message);
   }
 
+  private Mono<Result> requestResponse(List<Address> addresses, Message request) {
+    final AtomicInteger currentIndex = new AtomicInteger();
+    return Mono.defer(
+            () -> {
+              final int index = currentIndex.getAndIncrement();
+              return transport.requestResponse(addresses.get(index), request);
+            })
+        .retry(addresses.size() - 1)
+        .map(
+            message -> {
+              final int index = currentIndex.get();
+              return new Result(addresses.get(index), message);
+            });
+  }
+
+  /**
+   * Execute send call.
+   *
+   * @param member member
+   * @param request request
+   * @return mono result
+   */
   public Mono<Void> send(Member member, Message request) {
     return connections
         .compute(
@@ -53,21 +82,6 @@ public Mono<Void> send(Member member, Message request) {
         .then();
   }
 
-  private Mono<Result> requestResponse(List<Address> addresses, Message request) {
-    final AtomicInteger currentIndex = new AtomicInteger();
-    return Mono.defer(
-            () -> {
-              final int index = currentIndex.getAndIncrement();
-              return transport.requestResponse(addresses.get(index), request);
-            })
-        .retry(addresses.size() - 1)
-        .map(
-            message -> {
-              final int index = currentIndex.get();
-              return new Result(addresses.get(index), message);
-            });
-  }
-
   private Mono<Result> send(List<Address> addresses, Message request) {
     final AtomicInteger currentIndex = new AtomicInteger();
     return Mono.defer(

diff --git a/cluster/src/main/java/io/scalecube/cluster/fdetector/FailureDetectorImpl.java b/cluster/src/main/java/io/scalecube/cluster/fdetector/FailureDetectorImpl.java
@@ -3,12 +3,12 @@
 import static io.scalecube.reactor.RetryNonSerializedEmitFailureHandler.RETRY_NON_SERIALIZED;
 
 import io.scalecube.cluster.Member;
+import io.scalecube.cluster.TransportWrapper;
 import io.scalecube.cluster.fdetector.PingData.AckType;
 import io.scalecube.cluster.membership.MemberStatus;
 import io.scalecube.cluster.membership.MembershipEvent;
 import io.scalecube.cluster.transport.api.Message;
 import io.scalecube.cluster.transport.api.Transport;
-import io.scalecube.cluster.TransportWrapper;
 import io.scalecube.net.Address;
 import java.time.Duration;
 import java.util.ArrayList;

diff --git a/cluster/src/main/java/io/scalecube/cluster/gossip/GossipProtocolImpl.java b/cluster/src/main/java/io/scalecube/cluster/gossip/GossipProtocolImpl.java
@@ -4,10 +4,10 @@
 
 import io.scalecube.cluster.ClusterMath;
 import io.scalecube.cluster.Member;
+import io.scalecube.cluster.TransportWrapper;
 import io.scalecube.cluster.membership.MembershipEvent;
 import io.scalecube.cluster.transport.api.Message;
 import io.scalecube.cluster.transport.api.Transport;
-import io.scalecube.cluster.TransportWrapper;
 import io.scalecube.net.Address;
 import java.util.ArrayList;
 import java.util.Arrays;

diff --git a/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocolImpl.java b/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocolImpl.java
@@ -8,14 +8,14 @@
 import io.scalecube.cluster.ClusterConfig;
 import io.scalecube.cluster.ClusterMath;
 import io.scalecube.cluster.Member;
+import io.scalecube.cluster.TransportWrapper;
 import io.scalecube.cluster.fdetector.FailureDetector;
 import io.scalecube.cluster.fdetector.FailureDetectorConfig;
 import io.scalecube.cluster.fdetector.FailureDetectorEvent;
 import io.scalecube.cluster.gossip.GossipProtocol;
 import io.scalecube.cluster.metadata.MetadataStore;
 import io.scalecube.cluster.transport.api.Message;
 import io.scalecube.cluster.transport.api.Transport;
-import io.scalecube.cluster.TransportWrapper;
 import io.scalecube.net.Address;
 import java.net.InetAddress;
 import java.nio.ByteBuffer;

diff --git a/cluster/src/main/java/io/scalecube/cluster/metadata/MetadataStoreImpl.java b/cluster/src/main/java/io/scalecube/cluster/metadata/MetadataStoreImpl.java
@@ -2,9 +2,9 @@
 
 import io.scalecube.cluster.ClusterConfig;
 import io.scalecube.cluster.Member;
+import io.scalecube.cluster.TransportWrapper;
 import io.scalecube.cluster.transport.api.Message;
 import io.scalecube.cluster.transport.api.Transport;
-import io.scalecube.cluster.TransportWrapper;
 import io.scalecube.net.Address;
 import java.nio.ByteBuffer;
 import java.time.Duration;

diff --git a/cluster/src/test/java/io/scalecube/cluster/ClusterNamespacesTest.java b/cluster/src/test/java/io/scalecube/cluster/ClusterNamespacesTest.java
@@ -117,7 +117,10 @@ public void testSeparateNonEmptyNamespaces() {
             .membership(
                 opts ->
                     opts.seedMembers(
-                        root.addresses().get(0), root2.addresses().get(0), bob.addresses().get(0), carol.addresses().get(0)))
+                        root.addresses().get(0),
+                        root2.addresses().get(0),
+                        bob.addresses().get(0),
+                        carol.addresses().get(0)))
             .startAwait();
 
     Cluster eve =
@@ -162,15 +165,20 @@ public void testSimpleNamespacesHierarchy() {
         new ClusterImpl()
             .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("develop/develop"))
-            .membership(opts -> opts.seedMembers(rootDevelop.addresses().get(0), bob.addresses().get(0)))
+            .membership(
+                opts -> opts.seedMembers(rootDevelop.addresses().get(0), bob.addresses().get(0)))
             .startAwait();
 
     Cluster dan =
         new ClusterImpl()
             .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("develop/develop-2"))
             .membership(
-                opts -> opts.seedMembers(rootDevelop.addresses().get(0), bob.addresses().get(0), carol.addresses().get(0)))
+                opts ->
+                    opts.seedMembers(
+                        rootDevelop.addresses().get(0),
+                        bob.addresses().get(0),
+                        carol.addresses().get(0)))
             .startAwait();
 
     Cluster eve =
@@ -180,7 +188,10 @@ public void testSimpleNamespacesHierarchy() {
             .membership(
                 opts ->
                     opts.seedMembers(
-                        rootDevelop.addresses().get(0), bob.addresses().get(0), carol.addresses().get(0), dan.addresses().get(0)))
+                        rootDevelop.addresses().get(0),
+                        bob.addresses().get(0),
+                        carol.addresses().get(0),
+                        dan.addresses().get(0)))
             .startAwait();
 
     assertThat(
@@ -213,7 +224,8 @@ public void testIsolatedParentNamespaces() {
         new ClusterImpl()
             .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("a/1/c"))
-            .membership(opts -> opts.seedMembers(parent1.addresses().get(0), bob.addresses().get(0)))
+            .membership(
+                opts -> opts.seedMembers(parent1.addresses().get(0), bob.addresses().get(0)))
             .startAwait();
 
     Cluster parent2 =
@@ -229,7 +241,10 @@ public void testIsolatedParentNamespaces() {
             .membership(
                 opts ->
                     opts.seedMembers(
-                        parent1.addresses().get(0), parent2.addresses().get(0), bob.addresses().get(0), carol.addresses().get(0)))
+                        parent1.addresses().get(0),
+                        parent2.addresses().get(0),
+                        bob.addresses().get(0),
+                        carol.addresses().get(0)))
             .startAwait();
 
     //noinspection unused

diff --git a/cluster/src/test/java/io/scalecube/cluster/ClusterTest.java b/cluster/src/test/java/io/scalecube/cluster/ClusterTest.java
@@ -562,7 +562,7 @@ public void testJoinSeedClusterWithNoExistingSeedMember() {
     List<Address> seeds = new ArrayList<>();
 
     seeds.add(Address.from("localhost:1234")); // Not existent
-    seeds.add(Address.from("localhost:5678"));  // Not existent
+    seeds.add(Address.from("localhost:5678")); // Not existent
     seeds.addAll(seedNode.addresses());
 
     Cluster otherNode =

diff --git a/cluster/src/test/java/io/scalecube/cluster/fdetector/FailureDetectorTest.java b/cluster/src/test/java/io/scalecube/cluster/fdetector/FailureDetectorTest.java
@@ -474,7 +474,8 @@ private Future<List<FailureDetectorEvent>> listenNextEventFor(
     for (final Address member : addresses) {
       final CompletableFuture<FailureDetectorEvent> future = new CompletableFuture<>();
       fd.listen()
-        .filter(event -> event.member().addresses().contains(member)).subscribe(future::complete);
+          .filter(event -> event.member().addresses().contains(member))
+          .subscribe(future::complete);
       resultFuture.add(future);
     }
 

diff --git a/pom.xml b/pom.xml
@@ -1,5 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
   <modelVersion>4.0.0</modelVersion>
 
   <parent>
@@ -55,6 +57,7 @@
     <module>cluster-testlib</module>
     <module>transport-parent</module>
     <module>codec-parent</module>
+    <module>cluster-tests</module>
   </modules>
 
   <dependencies>

diff --git a/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/Message.java b/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/Message.java
@@ -200,9 +200,7 @@ public List<Address> sender() {
       return Collections.emptyList();
     }
 
-    return Arrays.stream(headerValue.split(","))
-        .map(Address::from)
-        .collect(Collectors.toList());
+    return Arrays.stream(headerValue.split(",")).map(Address::from).collect(Collectors.toList());
   }
 
   @Override
@@ -300,8 +298,8 @@ public Builder correlationId(String correlationId) {
      */
     public Builder sender(List<Address> addresses) {
       return header(
-        HEADER_SENDER,
-        addresses.stream().map(Address::toString).collect(Collectors.joining(",")));
+          HEADER_SENDER,
+          addresses.stream().map(Address::toString).collect(Collectors.joining(",")));
     }
 
     public Message build() {