Merge pull request #343 from scalecube/add-is-secured-transport-property

Added isSecured configuration setting in TransportConfig; cleaned jav…

Author: artem-v

cluster-api/src/main/java/io/scalecube/cluster/fdetector/FailureDetectorConfig.java

@@ -64,7 +64,7 @@ public static FailureDetectorConfig defaultLocalConfig() {
   }
 
   /**
-   * Sets pingInterval.
+   * Setter for {@code pingInterval}.
    *
    * @param pingInterval ping interval
    * @return new {@code FailureDetectorConfig}
@@ -80,7 +80,7 @@ public int pingInterval() {
   }
 
   /**
-   * Sets ping timeout.
+   * Setter for {@code pingTimeout}.
    *
    * @param pingTimeout ping timeout
    * @return new {@code FailureDetectorConfig}
@@ -96,7 +96,7 @@ public int pingTimeout() {
   }
 
   /**
-   * Sets number of members for requesting a ping.
+   * Setter for number of members for requesting a ping.
    *
    * @param pingReqMembers number of members for requesting a ping
    * @return new {@code FailureDetectorConfig}

cluster-api/src/main/java/io/scalecube/cluster/gossip/GossipConfig.java

@@ -60,7 +60,7 @@ public static GossipConfig defaultLocalConfig() {
   }
 
   /**
-   * Sets gossipFanout.
+   * Setter for {@code gossipFanout}.
    *
    * @param gossipFanout gossip fanout
    * @return new {@code GossipConfig}
@@ -76,7 +76,7 @@ public int gossipFanout() {
   }
 
   /**
-   * Sets gossipInterval.
+   * Setter for {@code gossipInterval}.
    *
    * @param gossipInterval gossip interval
    * @return new {@code GossipConfig}
@@ -92,7 +92,7 @@ public long gossipInterval() {
   }
 
   /**
-   * Sets gossipRepeatMult.
+   * Setter for {@code gossipRepeatMult}.
    *
    * @param gossipRepeatMult gossip repeat multiplier
    * @return new {@code GossipConfig}
@@ -108,7 +108,7 @@ public int gossipRepeatMult() {
   }
 
   /**
-   * Sets gossipSegmentationThreshold.
+   * Setter for {@code gossipSegmentationThreshold}.
    *
    * @param gossipSegmentationThreshold gossip segmentation threshold
    * @return new {@code GossipConfig}

cluster-api/src/main/java/io/scalecube/cluster/membership/MembershipConfig.java

@@ -73,7 +73,7 @@ public List<Address> seedMembers() {
   }
 
   /**
-   * Sets a seedMembers.
+   * Setter for {@code seedMembers}.
    *
    * @param seedMembers seed members
    * @return new {@code MembershipConfig} instance
@@ -83,7 +83,7 @@ public MembershipConfig seedMembers(Address... seedMembers) {
   }
 
   /**
-   * Sets a seedMembers.
+   * Setter for {@code seedMembers}.
    *
    * @param seedMembers seed members
    * @return new {@code MembershipConfig} instance
@@ -99,7 +99,7 @@ public int syncInterval() {
   }
 
   /**
-   * Sets a syncInterval.
+   * Setter for {@code syncInterval}.
    *
    * @param syncInterval sync interval
    * @return new {@code MembershipConfig} instance
@@ -115,7 +115,7 @@ public int syncTimeout() {
   }
 
   /**
-   * Sets a syncTimeout.
+   * Setter for {@code syncTimeout}.
    *
    * @param syncTimeout sync timeout
    * @return new {@code MembershipConfig} instance
@@ -131,7 +131,7 @@ public int suspicionMult() {
   }
 
   /**
-   * Sets a suspicionMult.
+   * Setter for {@code suspicionMult}.
    *
    * @param suspicionMult suspicion multiplier
    * @return new {@code MembershipConfig} instance
@@ -147,7 +147,7 @@ public String namespace() {
   }
 
   /**
-   * Sets a namespace.
+   * Setter for {@code namespace}.
    *
    * @param namespace namespace
    * @return new {@code MembershipConfig} instance
@@ -163,7 +163,7 @@ public int removedMembersHistorySize() {
   }
 
   /**
-   * Sets a removedMembersHistorySize.
+   * Setter for {@code removedMembersHistorySize}.
    *
    * @param removedMembersHistorySize history size for remove members
    * @return new {@code MembershipConfig} instance

cluster-testlib/src/main/java/io/scalecube/cluster/utils/NetworkEmulator.java

@@ -61,7 +61,7 @@ public OutboundSettings outboundSettings(Address destination) {
   }
 
   /**
-   * Sets given network emulator outbound settings for specific destination.
+   * Setter for network emulator outbound settings for specific destination.
    *
    * @param destination address of target endpoint
    * @param lossPercent loss in percents
@@ -74,7 +74,7 @@ public void outboundSettings(Address destination, int lossPercent, int meanDelay
   }
 
   /**
-   * Sets default network emulator outbound settings.
+   * Setter for network emulator outbound settings.
    *
    * @param lossPercent loss in percents
    * @param meanDelay mean delay
@@ -214,7 +214,7 @@ public InboundSettings inboundSettings(Address destination) {
   }
 
   /**
-   * Sets given network emulator inbound settings for specific destination.
+   * Setter for network emulator inbound settings for specific destination.
    *
    * @param shallPass shallPass inbound flag
    */
@@ -225,7 +225,7 @@ public void inboundSettings(Address destination, boolean shallPass) {
   }
 
   /**
-   * Sets default network emulator inbound settings.
+   * Setter for network emulator inbound settings.
    *
    * @param shallPass shallPass inbound flag
    */

transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/TransportConfig.java

@@ -15,6 +15,7 @@ public final class TransportConfig implements Cloneable {
   public static final int DEFAULT_LOCAL_CONNECT_TIMEOUT = 1_000;
 
   private int port = 0;
+  private boolean isSecured = false; // is client secured
   private int connectTimeout = DEFAULT_CONNECT_TIMEOUT;
   private MessageCodec messageCodec = MessageCodec.INSTANCE;
   private int maxFrameLength = 2 * 1024 * 1024; // 2 MB
@@ -58,7 +59,7 @@ public int port() {
   }
 
   /**
-   * Sets a port.
+   * Setter for {@code port}.
    *
    * @param port port
    * @return new {@code TransportConfig} instance
@@ -69,12 +70,28 @@ public TransportConfig port(int port) {
     return t;
   }
 
+  public boolean isSecured() {
+    return isSecured;
+  }
+
+  /**
+   * Setter to denote whether client part of the transport is secured.
+   *
+   * @param isSecured isSecured
+   * @return new {@code TransportConfig} instance
+   */
+  public TransportConfig secured(boolean isSecured) {
+    TransportConfig t = clone();
+    t.isSecured = isSecured;
+    return t;
+  }
+
   public int connectTimeout() {
     return connectTimeout;
   }
 
   /**
-   * Sets a connectTimeout.
+   * Setter for {@code connectTimeout}.
    *
    * @param connectTimeout connect timeout
    * @return new {@code TransportConfig} instance
@@ -90,7 +107,7 @@ public MessageCodec messageCodec() {
   }
 
   /**
-   * Sets a messageCodec.
+   * Setter for {@code messageCodec}.
    *
    * @param messageCodec message codec
    * @return new {@code TransportConfig} instance
@@ -106,7 +123,7 @@ public int maxFrameLength() {
   }
 
   /**
-   * Sets a maxFrameLength.
+   * Setter for {@code maxFrameLength}.
    *
    * @param maxFrameLength max frame length
    * @return new {@code TransportConfig} instance
@@ -122,7 +139,7 @@ public TransportFactory transportFactory() {
   }
 
   /**
-   * Sets a transportFactory.
+   * Setter for {@code transportFactory}.
    *
    * @param transportFactory transport factory
    * @return new {@code TransportConfig} instance
@@ -146,6 +163,7 @@ public TransportConfig clone() {
   public String toString() {
     return new StringJoiner(", ", TransportConfig.class.getSimpleName() + "[", "]")
         .add("port=" + port)
+        .add("isSecured=" + isSecured)
         .add("connectTimeout=" + connectTimeout)
         .add("messageCodec=" + messageCodec)
         .add("maxFrameLength=" + maxFrameLength)

transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/tcp/TcpSender.java

@@ -41,17 +41,19 @@ public Mono<Void> send(Message message) {
   }
 
   private TcpClient newTcpClient(SenderContext context, Address address) {
-    return TcpClient.create(ConnectionProvider.newConnection())
-        .runOn(context.loopResources())
-        .host(address.host())
-        .port(address.port())
-        .option(ChannelOption.TCP_NODELAY, true)
-        .option(ChannelOption.SO_KEEPALIVE, true)
-        .option(ChannelOption.SO_REUSEADDR, true)
-        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.connectTimeout())
-        .bootstrap(
-            b ->
-                BootstrapHandlers.updateConfiguration(
-                    b, "outbound", new TcpChannelInitializer(config.maxFrameLength())));
+    TcpClient tcpClient =
+        TcpClient.create(ConnectionProvider.newConnection())
+            .runOn(context.loopResources())
+            .host(address.host())
+            .port(address.port())
+            .option(ChannelOption.TCP_NODELAY, true)
+            .option(ChannelOption.SO_KEEPALIVE, true)
+            .option(ChannelOption.SO_REUSEADDR, true)
+            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.connectTimeout())
+            .bootstrap(
+                b ->
+                    BootstrapHandlers.updateConfiguration(
+                        b, "outbound", new TcpChannelInitializer(config.maxFrameLength())));
+    return config.isSecured() ? tcpClient.secure() : tcpClient;
   }
 }

transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/websocket/WebsocketSender.java

@@ -11,6 +11,7 @@
 import reactor.netty.Connection;
 import reactor.netty.http.client.HttpClient;
 import reactor.netty.http.client.WebsocketClientSpec;
+import reactor.netty.tcp.TcpClient;
 
 final class WebsocketSender implements Sender {
 
@@ -47,15 +48,18 @@ public Mono<Void> send(Message message) {
   private HttpClient.WebsocketSender newWebsocketSender(SenderContext context, Address address) {
     return HttpClient.newConnection()
         .tcpConfiguration(
-            tcpClient ->
-                tcpClient
-                    .runOn(context.loopResources())
-                    .host(address.host())
-                    .port(address.port())
-                    .option(ChannelOption.TCP_NODELAY, true)
-                    .option(ChannelOption.SO_KEEPALIVE, true)
-                    .option(ChannelOption.SO_REUSEADDR, true)
-                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.connectTimeout()))
+            tcpClient -> {
+              TcpClient tcpClient1 =
+                  tcpClient
+                      .runOn(context.loopResources())
+                      .host(address.host())
+                      .port(address.port())
+                      .option(ChannelOption.TCP_NODELAY, true)
+                      .option(ChannelOption.SO_KEEPALIVE, true)
+                      .option(ChannelOption.SO_REUSEADDR, true)
+                      .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.connectTimeout());
+              return config.isSecured() ? tcpClient1.secure() : tcpClient1;
+            })
         .websocket(
             WebsocketClientSpec.builder().maxFramePayloadLength(config.maxFrameLength()).build());
   }