Merge pull request #330 from scalecube/create-websocket-trasport

Create websocket trasport
scalecube · Jun 11, 2020 · 56040e0 · 56040e0
2 parents 3a7760e + f6b7425
commit 56040e0
Show file tree

Hide file tree

Showing 35 changed files with 1,299 additions and 216 deletions.
diff --git a/cluster-api/pom.xml b/cluster-api/pom.xml
@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.scalecube</groupId>
     <artifactId>scalecube-cluster-parent</artifactId>
-    <version>2.5.1-SNAPSHOT</version>
+    <version>2.6.0-SNAPSHOT</version>
   </parent>
 
   <artifactId>scalecube-cluster-api</artifactId>

diff --git a/cluster-testlib/pom.xml b/cluster-testlib/pom.xml
@@ -3,7 +3,7 @@
   <parent>
     <artifactId>scalecube-cluster-parent</artifactId>
     <groupId>io.scalecube</groupId>
-    <version>2.5.1-SNAPSHOT</version>
+    <version>2.6.0-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

diff --git a/cluster-testlib/src/main/java/io/scalecube/cluster/utils/NetworkEmulatorTransport.java b/cluster-testlib/src/main/java/io/scalecube/cluster/utils/NetworkEmulatorTransport.java
@@ -30,6 +30,11 @@ public Address address() {
     return transport.address();
   }
 
+  @Override
+  public Mono<Transport> start() {
+    return transport.start();
+  }
+
   @Override
   public Mono<Void> stop() {
     return transport.stop();

diff --git a/cluster/pom.xml b/cluster/pom.xml
@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.scalecube</groupId>
     <artifactId>scalecube-cluster-parent</artifactId>
-    <version>2.5.1-SNAPSHOT</version>
+    <version>2.6.0-SNAPSHOT</version>
   </parent>
 
   <artifactId>scalecube-cluster</artifactId>

diff --git a/cluster/src/main/java/io/scalecube/cluster/ClusterImpl.java b/cluster/src/main/java/io/scalecube/cluster/ClusterImpl.java
@@ -18,6 +18,7 @@
 import io.scalecube.cluster.transport.api.Message;
 import io.scalecube.cluster.transport.api.Transport;
 import io.scalecube.cluster.transport.api.TransportConfig;
+import io.scalecube.cluster.transport.api.TransportFactory;
 import io.scalecube.net.Address;
 import io.scalecube.transport.netty.TransportImpl;
 import io.scalecube.utils.ServiceLoaderUtil;
@@ -30,6 +31,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import java.util.function.UnaryOperator;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -158,6 +160,19 @@ public ClusterImpl transport(UnaryOperator<TransportConfig> options) {
     return cluster;
   }
 
+  /**
+   * Returns a new cluster's instance which will apply the given options.
+   *
+   * @param supplier transport factory supplier
+   * @return new {@code ClusterImpl} instance
+   */
+  public ClusterImpl transportFactory(Supplier<TransportFactory> supplier) {
+    Objects.requireNonNull(supplier);
+    ClusterImpl cluster = new ClusterImpl(this);
+    cluster.config = config.transport(opts -> opts.transportFactory(supplier.get()));
+    return cluster;
+  }
+
   /**
    * Returns a new cluster's instance which will apply the given options.
    *
@@ -566,6 +581,11 @@ public Address address() {
       return transport.address();
     }
 
+    @Override
+    public Mono<Transport> start() {
+      return transport.start();
+    }
+
     @Override
     public Mono<Void> stop() {
       return transport.stop();

diff --git a/cluster/src/test/java/io/scalecube/cluster/ClusterNamespacesTest.java b/cluster/src/test/java/io/scalecube/cluster/ClusterNamespacesTest.java
@@ -7,6 +7,7 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.params.provider.Arguments.of;
 
+import io.scalecube.transport.netty.websocket.WebsocketTransportFactory;
 import java.util.stream.Stream;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -22,7 +23,11 @@ public void testInvalidNamespaceFormat(String namespace) {
     Exception actualException =
         assertThrows(
             IllegalArgumentException.class,
-            () -> new ClusterImpl().membership(opts -> opts.namespace(namespace)).startAwait());
+            () ->
+                new ClusterImpl()
+                    .transportFactory(WebsocketTransportFactory::new)
+                    .membership(opts -> opts.namespace(namespace))
+                    .startAwait());
     Assertions.assertAll(
         () ->
             assertEquals(
@@ -50,16 +55,22 @@ public static Stream<Arguments> testInvalidNamespaceFormat() {
 
   @Test
   public void testSeparateEmptyNamespaces() {
-    Cluster root = new ClusterImpl().membership(opts -> opts.namespace("root")).startAwait();
+    Cluster root =
+        new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
+            .membership(opts -> opts.namespace("root"))
+            .startAwait();
 
     Cluster root1 =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("root1"))
             .membership(opts -> opts.seedMembers(root.address()))
             .startAwait();
 
     Cluster root2 =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("root2"))
             .membership(opts -> opts.seedMembers(root.address()))
             .startAwait();
@@ -71,28 +82,36 @@ public void testSeparateEmptyNamespaces() {
 
   @Test
   public void testSeparateNonEmptyNamespaces() {
-    Cluster root = new ClusterImpl().membership(opts -> opts.namespace("root")).startAwait();
+    Cluster root =
+        new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
+            .membership(opts -> opts.namespace("root"))
+            .startAwait();
 
     Cluster bob =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("root"))
             .membership(opts -> opts.seedMembers(root.address()))
             .startAwait();
 
     Cluster carol =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("root"))
             .membership(opts -> opts.seedMembers(root.address(), bob.address()))
             .startAwait();
 
     Cluster root2 =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("root2"))
             .membership(opts -> opts.seedMembers(root.address()))
             .startAwait();
 
     Cluster dan =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("root2"))
             .membership(
                 opts ->
@@ -102,6 +121,7 @@ public void testSeparateNonEmptyNamespaces() {
 
     Cluster eve =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("root2"))
             .membership(
                 opts ->
@@ -125,29 +145,36 @@ public void testSeparateNonEmptyNamespaces() {
   @Test
   public void testSimpleNamespacesHierarchy() {
     Cluster rootDevelop =
-        new ClusterImpl().membership(opts -> opts.namespace("develop")).startAwait();
+        new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
+            .membership(opts -> opts.namespace("develop"))
+            .startAwait();
 
     Cluster bob =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("develop/develop"))
             .membership(opts -> opts.seedMembers(rootDevelop.address()))
             .startAwait();
 
     Cluster carol =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("develop/develop"))
             .membership(opts -> opts.seedMembers(rootDevelop.address(), bob.address()))
             .startAwait();
 
     Cluster dan =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("develop/develop-2"))
             .membership(
                 opts -> opts.seedMembers(rootDevelop.address(), bob.address(), carol.address()))
             .startAwait();
 
     Cluster eve =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("develop/develop-2"))
             .membership(
                 opts ->
@@ -168,24 +195,35 @@ public void testSimpleNamespacesHierarchy() {
 
   @Test
   public void testIsolatedParentNamespaces() {
-    Cluster parent1 = new ClusterImpl().membership(opts -> opts.namespace("a/1")).startAwait();
+    Cluster parent1 =
+        new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
+            .membership(opts -> opts.namespace("a/1"))
+            .startAwait();
 
     Cluster bob =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("a/1/c"))
             .membership(opts -> opts.seedMembers(parent1.address()))
             .startAwait();
 
     Cluster carol =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("a/1/c"))
             .membership(opts -> opts.seedMembers(parent1.address(), bob.address()))
             .startAwait();
 
-    Cluster parent2 = new ClusterImpl().membership(opts -> opts.namespace("a/111")).startAwait();
+    Cluster parent2 =
+        new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
+            .membership(opts -> opts.namespace("a/111"))
+            .startAwait();
 
     Cluster dan =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("a/111/c"))
             .membership(
                 opts ->
@@ -195,6 +233,7 @@ public void testIsolatedParentNamespaces() {
 
     Cluster eve =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .membership(opts -> opts.namespace("a/111/c"))
             .membership(
                 opts ->

diff --git a/codec-parent/codec-jackson-smile/pom.xml b/codec-parent/codec-jackson-smile/pom.xml
@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.scalecube</groupId>
     <artifactId>scalecube-codec-parent</artifactId>
-    <version>2.5.1-SNAPSHOT</version>
+    <version>2.6.0-SNAPSHOT</version>
   </parent>
 
   <artifactId>scalecube-codec-jackson-smile</artifactId>

diff --git a/codec-parent/codec-jackson/pom.xml b/codec-parent/codec-jackson/pom.xml
@@ -3,7 +3,7 @@
   <parent>
     <artifactId>scalecube-codec-parent</artifactId>
     <groupId>io.scalecube</groupId>
-    <version>2.5.1-SNAPSHOT</version>
+    <version>2.6.0-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

diff --git a/codec-parent/pom.xml b/codec-parent/pom.xml
@@ -4,7 +4,7 @@
   <parent>
     <artifactId>scalecube-cluster-parent</artifactId>
     <groupId>io.scalecube</groupId>
-    <version>2.5.1-SNAPSHOT</version>
+    <version>2.6.0-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
   <packaging>pom</packaging>

diff --git a/examples/pom.xml b/examples/pom.xml
@@ -4,7 +4,7 @@
   <parent>
     <groupId>io.scalecube</groupId>
     <artifactId>scalecube-cluster-parent</artifactId>
-    <version>2.5.1-SNAPSHOT</version>
+    <version>2.6.0-SNAPSHOT</version>
   </parent>
 
   <artifactId>scalecube-cluster-examples</artifactId>
@@ -16,6 +16,11 @@
       <artifactId>scalecube-cluster</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>scalecube-transport-netty</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>io.scalecube</groupId>
       <artifactId>scalecube-cluster-testlib</artifactId>

diff --git a/...ClusterCustomMetadataEncodingExample.java → ...amples/CustomMetadataEncodingExample.java b/...ClusterCustomMetadataEncodingExample.java → ...amples/CustomMetadataEncodingExample.java
@@ -3,21 +3,26 @@
 import io.scalecube.cluster.Cluster;
 import io.scalecube.cluster.ClusterImpl;
 import io.scalecube.cluster.metadata.MetadataCodec;
+import io.scalecube.transport.netty.websocket.WebsocketTransportFactory;
 import java.nio.ByteBuffer;
 import java.util.concurrent.TimeUnit;
 
-public class ClusterCustomMetadataEncodingExample {
+public class CustomMetadataEncodingExample {
 
   /** Main method. */
   public static void main(String[] args) throws Exception {
     // Start seed cluster member Alice
     Cluster alice =
-        new ClusterImpl().config(opts -> opts.metadataCodec(new LongMetadataCodec())).startAwait();
+        new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
+            .config(opts -> opts.metadataCodec(new LongMetadataCodec()))
+            .startAwait();
     System.out.println(
         "[" + alice.member().id() + "] Alice's metadata: " + alice.metadata().orElse(null));
 
     Cluster joe =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .config(opts -> opts.metadataCodec(new LongMetadataCodec()).metadata(123L))
             .membership(opts -> opts.seedMembers(alice.address()))
             .startAwait();
@@ -26,6 +31,7 @@ public static void main(String[] args) throws Exception {
 
     Cluster bob =
         new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
             .config(opts -> opts.metadataCodec(new LongMetadataCodec()).metadata(456L))
             .membership(opts -> opts.seedMembers(alice.address()))
             .startAwait();