Added example

Author: artem-v

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.
    *

examples/src/main/java/io/scalecube/examples/WebsocketMessagingExample.java

@@ -0,0 +1,84 @@
+package io.scalecube.examples;
+
+import io.scalecube.cluster.Cluster;
+import io.scalecube.cluster.ClusterImpl;
+import io.scalecube.cluster.ClusterMessageHandler;
+import io.scalecube.cluster.transport.api.Message;
+import io.scalecube.transport.netty.websocket.WebsocketTransportFactory;
+import reactor.core.publisher.Flux;
+
+/**
+ * Exactly the same example as {@link MessagingExample} but cluster transport is websocket.
+ *
+ * @see io.scalecube.transport.netty.websocket.WebsocketTransportFactory
+ * @author arvy
+ */
+public class WebsocketMessagingExample {
+
+  /** Main method. */
+  public static void main(String[] args) throws Exception {
+    // Start cluster node Alice to listen and respond for incoming greeting messages
+    Cluster alice =
+        new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
+            .handler(
+                cluster -> {
+                  return new ClusterMessageHandler() {
+                    @Override
+                    public void onMessage(Message msg) {
+                      System.out.println("Alice received: " + msg.data());
+                      cluster
+                          .send(msg.sender(), Message.fromData("Greetings from Alice"))
+                          .subscribe(null, Throwable::printStackTrace);
+                    }
+                  };
+                })
+            .startAwait();
+
+    // Join cluster node Bob to cluster with Alice, listen and respond for incoming greeting
+    // messages
+    Cluster bob =
+        new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
+            .membership(opts -> opts.seedMembers(alice.address()))
+            .handler(
+                cluster -> {
+                  return new ClusterMessageHandler() {
+                    @Override
+                    public void onMessage(Message msg) {
+                      System.out.println("Bob received: " + msg.data());
+                      cluster
+                          .send(msg.sender(), Message.fromData("Greetings from Bob"))
+                          .subscribe(null, Throwable::printStackTrace);
+                    }
+                  };
+                })
+            .startAwait();
+
+    // Join cluster node Carol to cluster with Alice and Bob
+    Cluster carol =
+        new ClusterImpl()
+            .transportFactory(WebsocketTransportFactory::new)
+            .membership(opts -> opts.seedMembers(alice.address(), bob.address()))
+            .handler(
+                cluster -> {
+                  return new ClusterMessageHandler() {
+                    @Override
+                    public void onMessage(Message msg) {
+                      System.out.println("Carol received: " + msg.data());
+                    }
+                  };
+                })
+            .startAwait();
+
+    // Send from Carol greeting message to all other cluster members (which is Alice and Bob)
+    Message greetingMsg = Message.fromData("Greetings from Carol");
+
+    Flux.fromIterable(carol.otherMembers())
+        .flatMap(member -> carol.send(member, greetingMsg))
+        .subscribe(null, Throwable::printStackTrace);
+
+    // Avoid exit main thread immediately ]:->
+    Thread.sleep(1000);
+  }
+}

examples/src/main/java/io/scalecube/examples/WebsocketTransportExample.java

@@ -12,15 +12,15 @@
   <Loggers>
     <Logger name="io.netty" level="info"/>
     <Logger name="reactor.netty" level="info"/>
-    <Logger name="io.scalecube.cluster.transport" level="debug"/>
-    <Logger name="io.scalecube.cluster.fdetector" level="debug"/>
-    <Logger name="io.scalecube.cluster.gossip" level="debug"/>
-    <Logger name="io.scalecube.cluster.membership" level="debug"/>
-    <Logger name="io.scalecube.cluster.metadata" level="debug"/>
-    <Logger name="io.scalecube.cluster" level="debug"/>
-    <Logger name="io.scalecube.services" level="debug"/>
+    <Logger name="io.scalecube.cluster.transport" level="info"/>
+    <Logger name="io.scalecube.cluster.fdetector" level="info"/>
+    <Logger name="io.scalecube.cluster.gossip" level="info"/>
+    <Logger name="io.scalecube.cluster.membership" level="info"/>
+    <Logger name="io.scalecube.cluster.metadata" level="info"/>
+    <Logger name="io.scalecube.cluster" level="info"/>
+    <Logger name="io.scalecube.services" level="info"/>
 
-    <Root level="debug">
+    <Root level="info">
       <AppenderRef ref="console"/>
     </Root>
   </Loggers>

examples/src/main/resources/log4j2.xml

@@ -12,6 +12,7 @@
 import io.scalecube.cluster.transport.api.TransportConfig;
 import io.scalecube.cluster.transport.api.TransportFactory;
 import io.scalecube.net.Address;
+import io.scalecube.transport.netty.tcp.TcpTransportFactory;
 import java.net.InetAddress;
 import java.util.Map;
 import java.util.Objects;
@@ -137,7 +138,7 @@ public static Mono<Transport> bind(TransportConfig config) {
     TransportFactory transportFactory =
         Optional.ofNullable(config.transportFactory())
             .or(() -> Optional.ofNullable(TransportFactory.INSTANCE))
-            .orElseThrow(() -> new IllegalStateException("No TransportFactory is defined"));
+            .orElse(new TcpTransportFactory());
     return transportFactory.createTransport(config).start();
   }