Enh external hosts

cluster-api/src/main/java/io/scalecube/cluster/Cluster.java

@@ -3,6 +3,7 @@
 import io.scalecube.cluster.transport.api.Message;
 import io.scalecube.net.Address;
 import java.util.Collection;
+import java.util.List;
 import java.util.Optional;
 import reactor.core.publisher.Mono;
 
@@ -14,7 +15,7 @@ public interface Cluster {
    *
    * @return cluster address
    */
-  Address address();
+  List<Address> addresses();
 
   /**
    * Spreads given message between cluster members using gossiping protocol.

cluster-api/src/main/java/io/scalecube/cluster/ClusterConfig.java

@@ -5,6 +5,8 @@
 import io.scalecube.cluster.membership.MembershipConfig;
 import io.scalecube.cluster.metadata.MetadataCodec;
 import io.scalecube.cluster.transport.api.TransportConfig;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Optional;
 import java.util.StringJoiner;
 import java.util.function.UnaryOperator;
@@ -35,8 +37,7 @@ public final class ClusterConfig implements Cloneable {
 
   private String memberId;
   private String memberAlias;
-  private String externalHost;
-  private Integer externalPort;
+  private List<String> externalHosts;
 
   private TransportConfig transportConfig = TransportConfig.defaultConfig();
   private FailureDetectorConfig failureDetectorConfig = FailureDetectorConfig.defaultConfig();
@@ -136,27 +137,39 @@ public ClusterConfig metadataCodec(MetadataCodec metadataCodec) {
   }
 
   /**
-   * Returns externalHost. {@code externalHost} is a config property for container environments,
-   * it's being set for advertising to scalecube cluster some connectable hostname which maps to
+   * Returns externalHosts. {@code externalHosts} is a config property for container environments,
+   * it's being set for advertising to scalecube cluster some connectable hostnames which maps to
    * scalecube transport's hostname on which scalecube transport is listening.
    *
-   * @return external host
+   * @return external hosts
    */
-  public String externalHost() {
-    return externalHost;
+  public List<String> externalHosts() {
+    return externalHosts;
   }
 
   /**
-   * Setter for externalHost. {@code externalHost} is a config property for container environments,
-   * it's being set for advertising to scalecube cluster some connectable hostname which maps to
-   * scalecube transport's hostname on which scalecube transport is listening.
+   * Setter for externalHosts. {@code externalHosts} is a config property for container
+   * environments, it's being set for advertising to scalecube cluster some connectable hostnames
+   * which maps to scalecube transport's hostname on which scalecube transport is listening.
+   *
+   * @param externalHosts external hosts
+   * @return new {@code ClusterConfig} instance
+   */
+  public ClusterConfig externalHosts(String... externalHosts) {
+    return externalHosts(Arrays.asList(externalHosts));
+  }
+
+  /**
+   * Setter for externalHosts. {@code externalHosts} is a config property for container
+   * environments, it's being set for advertising to scalecube cluster some connectable hostnames
+   * which maps to scalecube transport's hostname on which scalecube transport is listening.
    *
-   * @param externalHost external host
+   * @param externalHosts external hosts
    * @return new {@code ClusterConfig} instance
    */
-  public ClusterConfig externalHost(String externalHost) {
+  public ClusterConfig externalHosts(List<String> externalHosts) {
     ClusterConfig c = clone();
-    c.externalHost = externalHost;
+    c.externalHosts = externalHosts;
     return c;
   }
 
@@ -205,31 +218,6 @@ public ClusterConfig memberAlias(String memberAlias) {
     return c;
   }
 
-  /**
-   * Returns externalPort. {@code externalPort} is a config property for container environments,
-   * it's being set for advertising to scalecube cluster a port which mapped to scalecube
-   * transport's listening port.
-   *
-   * @return external port
-   */
-  public Integer externalPort() {
-    return externalPort;
-  }
-
-  /**
-   * Setter for externalPort. {@code externalPort} is a config property for container environments,
-   * it's being set for advertising to scalecube cluster a port which mapped to scalecube
-   * transport's listening port.
-   *
-   * @param externalPort external port
-   * @return new {@code ClusterConfig} instance
-   */
-  public ClusterConfig externalPort(Integer externalPort) {
-    ClusterConfig c = clone();
-    c.externalPort = externalPort;
-    return c;
-  }
-
   /**
    * Applies {@link TransportConfig} settings.
    *
@@ -316,8 +304,7 @@ public String toString() {
         .add("metadataCodec=" + metadataCodec)
         .add("memberId='" + memberId + "'")
         .add("memberAlias='" + memberAlias + "'")
-        .add("externalHost='" + externalHost + "'")
-        .add("externalPort=" + externalPort)
+        .add("externalHosts=" + externalHosts)
         .add("transportConfig=" + transportConfig)
         .add("failureDetectorConfig=" + failureDetectorConfig)
         .add("gossipConfig=" + gossipConfig)

cluster-api/src/main/java/io/scalecube/cluster/Member.java

@@ -6,6 +6,9 @@
 import java.io.IOException;
 import java.io.ObjectInput;
 import java.io.ObjectOutput;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.Objects;
 import java.util.StringJoiner;
 import java.util.UUID;
@@ -20,24 +23,36 @@ public final class Member implements Externalizable {
 
   private String id;
   private String alias;
-  private Address address;
+  private List<Address> addresses;
   private String namespace;
 
   public Member() {}
 
   /**
    * Constructor.
    *
-   * @param id member id; not null
+   * @param id member id
    * @param alias member alias (optional)
-   * @param address member address; not null
-   * @param namespace namespace; not null
+   * @param addresses member addresses
+   * @param namespace namespace
+   */
+  public Member(String id, String alias, List<Address> addresses, String namespace) {
+    this.id = Objects.requireNonNull(id, "id");
+    this.alias = alias;
+    this.addresses = Objects.requireNonNull(addresses, "addresses");
+    this.namespace = Objects.requireNonNull(namespace, "namespace");
+  }
+
+  /**
+   * Constructor.
+   *
+   * @param id member id
+   * @param alias member alias (optional)
+   * @param address member address
+   * @param namespace namespace
    */
   public Member(String id, String alias, Address address, String namespace) {
-    this.id = Objects.requireNonNull(id, "member id");
-    this.alias = alias; // optional
-    this.address = Objects.requireNonNull(address, "member address");
-    this.namespace = Objects.requireNonNull(namespace, "member namespace");
+    this(id, alias, Collections.singletonList(address), namespace);
   }
 
   /**
@@ -70,14 +85,14 @@ public String namespace() {
   }
 
   /**
-   * Returns cluster member address, an address on which this cluster member listens connections
-   * from other cluster members.
+   * Returns cluster member addresses, those are addresses on which this cluster member listens
+   * connections from other cluster members.
    *
    * @see io.scalecube.cluster.transport.api.TransportConfig#port(int)
    * @return member address
    */
-  public Address address() {
-    return address;
+  public List<Address> addresses() {
+    return addresses;
   }
 
   @Override
@@ -90,13 +105,13 @@ public boolean equals(Object that) {
     }
     Member member = (Member) that;
     return Objects.equals(id, member.id)
-        && Objects.equals(address, member.address)
+        && Objects.equals(addresses, member.addresses)
         && Objects.equals(namespace, member.namespace);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(id, address, namespace);
+    return Objects.hash(id, addresses, namespace);
   }
 
   @Override
@@ -110,7 +125,10 @@ public void writeExternal(ObjectOutput out) throws IOException {
       out.writeUTF(alias);
     }
     // address
-    out.writeUTF(address.toString());
+    out.writeInt(addresses.size());
+    for (Address address : addresses) {
+      out.writeUTF(address.toString());
+    }
     // namespace
     out.writeUTF(namespace);
   }
@@ -124,8 +142,12 @@ public void readExternal(ObjectInput in) throws IOException {
     if (aliasNotNull) {
       alias = in.readUTF();
     }
-    // address
-    address = Address.from(in.readUTF());
+    // addresses
+    final int addressesSize = in.readInt();
+    addresses = new ArrayList<>(addressesSize);
+    for (int i = 0; i < addressesSize; i++) {
+      addresses.add(Address.from(in.readUTF()));
+    }
     // namespace
     this.namespace = in.readUTF();
   }
@@ -143,9 +165,9 @@ private static String stringifyId(String id) {
   public String toString() {
     StringJoiner stringJoiner = new StringJoiner(":");
     if (alias == null) {
-      return stringJoiner.add(namespace).add(stringifyId(id) + "@" + address).toString();
+      return stringJoiner.add(namespace).add(stringifyId(id)).toString();
     } else {
-      return stringJoiner.add(namespace).add(alias).add(stringifyId(id) + "@" + address).toString();
+      return stringJoiner.add(namespace).add(alias).add(stringifyId(id)).toString();
     }
   }
 }

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

@@ -1,10 +1,12 @@
 package io.scalecube.cluster.utils;
 
+import io.scalecube.cluster.Member;
 import io.scalecube.cluster.transport.api.Message;
 import io.scalecube.net.Address;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.StringJoiner;
 import java.util.concurrent.ConcurrentHashMap;
@@ -213,15 +215,44 @@ public InboundSettings inboundSettings(Address destination) {
     return inboundSettings.getOrDefault(destination, defaultInboundSettings);
   }
 
+  /**
+   * Returns network inbound settings applied to the given destination.
+   *
+   * @param member target member
+   * @return network inbound settings
+   */
+  public InboundSettings inboundSettings(Member member) {
+    final List<Address> destinations = member.addresses();
+
+    if (destinations.isEmpty()) {
+      return defaultInboundSettings;
+    }
+
+    for (Address destination : destinations) {
+      InboundSettings inboundSettings = this.inboundSettings.get(destination);
+
+      if (inboundSettings != null) {
+        return inboundSettings;
+      }
+    }
+
+    return defaultInboundSettings;
+  }
+
   /**
    * Setter for network emulator inbound settings for specific destination.
    *
    * @param shallPass shallPass inbound flag
    */
-  public void inboundSettings(Address destination, boolean shallPass) {
+  public void inboundSettings(Member member, boolean shallPass) {
+    final List<Address> destinations = member.addresses();
     InboundSettings settings = new InboundSettings(shallPass);
-    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);
+        });
   }
 
   /**

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

@@ -1,5 +1,6 @@
 package io.scalecube.cluster.utils;
 
+import io.scalecube.cluster.Member;
 import io.scalecube.cluster.transport.api.Message;
 import io.scalecube.cluster.transport.api.Transport;
 import io.scalecube.net.Address;
@@ -49,7 +50,7 @@ public boolean isStopped() {
   public Mono<Void> send(Address address, Message message) {
     return Mono.defer(
         () ->
-            Mono.just(enhanceWithSender(message))
+            Mono.just(Message.with(message).build())
                 .flatMap(msg -> networkEmulator.tryFailOutbound(msg, address))
                 .flatMap(msg -> networkEmulator.tryDelayOutbound(msg, address))
                 .flatMap(msg -> transport.send(address, msg)));
@@ -59,7 +60,7 @@ public Mono<Void> send(Address address, Message message) {
   public Mono<Message> requestResponse(Address address, Message request) {
     return Mono.defer(
         () ->
-            Mono.just(enhanceWithSender(request))
+            Mono.just(Message.with(request).build())
                 .flatMap(msg -> networkEmulator.tryFailOutbound(msg, address))
                 .flatMap(msg -> networkEmulator.tryDelayOutbound(msg, address))
                 .flatMap(
@@ -69,7 +70,9 @@ public Mono<Message> requestResponse(Address address, Message request) {
                             .flatMap(
                                 message -> {
                                   boolean shallPass =
-                                      networkEmulator.inboundSettings(message.sender()).shallPass();
+                                      networkEmulator
+                                          .inboundSettings((Member) message.sender())
+                                          .shallPass();
                                   return shallPass ? Mono.just(message) : Mono.never();
                                 })));
   }
@@ -78,11 +81,7 @@ public Mono<Message> requestResponse(Address address, Message request) {
   public Flux<Message> listen() {
     return transport
         .listen()
-        .filter(message -> networkEmulator.inboundSettings(message.sender()).shallPass())
+        .filter(message -> networkEmulator.inboundSettings((Member) message.sender()).shallPass())
         .onBackpressureBuffer();
   }
-
-  private Message enhanceWithSender(Message message) {
-    return Message.with(message).sender(transport.address()).build();
-  }
 }