Merge branch 'refs/heads/master' into transportApi2

cluster-api/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/xsd/maven-4.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/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
 
  <parent>
@@ -12,10 +14,6 @@
  <name>ScaleCube/ClusterApi</name>
 
  <dependencies>
- <dependency>
- <groupId>io.scalecube</groupId>
- <artifactId>scalecube-commons</artifactId>
- </dependency>
  <dependency>
  <groupId>${project.groupId}</groupId>
  <artifactId>scalecube-transport-api</artifactId>

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

@@ -1,7 +1,6 @@
 package io.scalecube.cluster;
 
 import io.scalecube.cluster.transport.api.Message;
-import io.scalecube.net.Address;
 import java.util.Collection;
 import java.util.Optional;
 import reactor.core.publisher.Mono;
@@ -10,11 +9,11 @@
 public interface Cluster {
 
  /**
- * Returns {@link Address} of this cluster instance.
+ * Returns address of this cluster instance.
  *
  * @return cluster address
  */
- Address address();
+ String address();
 
  /**
  * Spreads given message between cluster members using gossiping protocol.
@@ -52,15 +51,15 @@ public interface Cluster {
  *
  * @return member by id
  */
- Optional<Member> member(String id);
+ Optional<Member> memberById(String id);
 
  /**
  * Returns cluster member by given address or null if no member with such address exists at joined
  * cluster.
  *
  * @return member by address
  */
- Optional<Member> member(Address address);
+ Optional<Member> memberByAddress(String address);
 
  /**
  * Returns list of all members of the joined cluster. This will include all cluster members

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

@@ -1,7 +1,6 @@
 package io.scalecube.cluster;
 
 import io.scalecube.cluster.membership.MembershipConfig;
-import io.scalecube.net.Address;
 import java.io.Externalizable;
 import java.io.IOException;
 import java.io.ObjectInput;
@@ -20,7 +19,7 @@ public final class Member implements Externalizable {
 
  private String id;
  private String alias;
- private Address address;
+ private String address;
  private String namespace;
 
  public Member() {}
@@ -33,7 +32,7 @@ public Member() {}
  * @param address member address; not null
  * @param namespace namespace; not null
  */
- public Member(String id, String alias, Address address, String namespace) {
+ public Member(String id, String alias, String address, String namespace) {
  this.id = Objects.requireNonNull(id, "member id");
  this.alias = alias; // optional
  this.address = Objects.requireNonNull(address, "member address");
@@ -76,7 +75,7 @@ public String namespace() {
  * @see io.scalecube.cluster.transport.api.TransportConfig#port(int)
  * @return member address
  */
- public Address address() {
+ public String address() {
  return address;
  }
 
@@ -110,7 +109,7 @@ public void writeExternal(ObjectOutput out) throws IOException {
  out.writeUTF(alias);
  }
  // address
- out.writeUTF(address.toString());
+ out.writeUTF(address);
  // namespace
  out.writeUTF(namespace);
  }
@@ -125,7 +124,7 @@ public void readExternal(ObjectInput in) throws IOException {
  alias = in.readUTF();
  }
  // address
- address = Address.from(in.readUTF());
+ address = in.readUTF();
  // namespace
  this.namespace = in.readUTF();
  }

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

@@ -1,6 +1,5 @@
 package io.scalecube.cluster.membership;
 
-import io.scalecube.net.Address;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -24,7 +23,7 @@ public final class MembershipConfig implements Cloneable {
  public static final int DEFAULT_LOCAL_SUSPICION_MULT = 3;
  public static final int DEFAULT_LOCAL_SYNC_INTERVAL = 15_000;
 
- private List<Address> seedMembers = Collections.emptyList();
+ private List<String> seedMembers = Collections.emptyList();
  private int syncInterval = DEFAULT_SYNC_INTERVAL;
  private int syncTimeout = DEFAULT_SYNC_TIMEOUT;
  private int suspicionMult = DEFAULT_SUSPICION_MULT;
@@ -67,7 +66,7 @@ public static MembershipConfig defaultLocalConfig() {
  .syncInterval(DEFAULT_LOCAL_SYNC_INTERVAL);
  }
 
- public List<Address> seedMembers() {
+ public List<String> seedMembers() {
  return seedMembers;
  }
 
@@ -77,7 +76,7 @@ public List<Address> seedMembers() {
  * @param seedMembers seed members
  * @return new {@code MembershipConfig} instance
  */
- public MembershipConfig seedMembers(Address... seedMembers) {
+ public MembershipConfig seedMembers(String... seedMembers) {
  return seedMembers(Arrays.asList(seedMembers));
  }
 
@@ -87,7 +86,7 @@ public MembershipConfig seedMembers(Address... seedMembers) {
  * @param seedMembers seed members
  * @return new {@code MembershipConfig} instance
  */
- public MembershipConfig seedMembers(List<Address> seedMembers) {
+ public MembershipConfig seedMembers(List<String> seedMembers) {
  MembershipConfig m = clone();
  m.seedMembers = Collections.unmodifiableList(new ArrayList<>(seedMembers));
  return m;

cluster-api/src/main/java/io/scalecube/cluster/metadata/MetadataCodec.java

@@ -1,13 +1,16 @@
 package io.scalecube.cluster.metadata;
 
-import io.scalecube.utils.ServiceLoaderUtil;
 import java.nio.ByteBuffer;
+import java.util.ServiceLoader;
+import java.util.stream.StreamSupport;
 
 /** Contains methods for metadata serializing/deserializing logic. */
 public interface MetadataCodec {
 
  MetadataCodec INSTANCE =
- ServiceLoaderUtil.findFirst(MetadataCodec.class).orElseGet(JdkMetadataCodec::new);
+ StreamSupport.stream(ServiceLoader.load(MetadataCodec.class).spliterator(), false)
+ .findFirst()
+ .orElseGet(JdkMetadataCodec::new);
 
  /**
  * Deserializes metadata from buffer.

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

@@ -1,7 +1,6 @@
 package io.scalecube.cluster.utils;
 
 import io.scalecube.cluster.transport.api.Message;
-import io.scalecube.net.Address;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
@@ -30,21 +29,21 @@ public final class NetworkEmulator {
  private volatile OutboundSettings defaultOutboundSettings = new OutboundSettings(0, 0);
  private volatile InboundSettings defaultInboundSettings = new InboundSettings(true);
 
- private final Map<Address, OutboundSettings> outboundSettings = new ConcurrentHashMap<>();
- private final Map<Address, InboundSettings> inboundSettings = new ConcurrentHashMap<>();
+ private final Map<String, OutboundSettings> outboundSettings = new ConcurrentHashMap<>();
+ private final Map<String, InboundSettings> inboundSettings = new ConcurrentHashMap<>();
 
  private final AtomicLong totalMessageSentCount = new AtomicLong();
  private final AtomicLong totalOutboundMessageLostCount = new AtomicLong();
  private final AtomicLong totalInboundMessageLostCount = new AtomicLong();
 
- private final Address address;
+ private final String address;
 
  /**
  * Creates new instance of network emulator.
  *
  * @param address local address
  */
- NetworkEmulator(Address address) {
+ NetworkEmulator(String address) {
  this.address = address;
  }
 
@@ -56,7 +55,7 @@ public final class NetworkEmulator {
  * @param destination address of target endpoint
  * @return network outbound settings
  */
- public OutboundSettings outboundSettings(Address destination) {
+ public OutboundSettings outboundSettings(String destination) {
  return outboundSettings.getOrDefault(destination, defaultOutboundSettings);
  }
 
@@ -67,7 +66,7 @@ public OutboundSettings outboundSettings(Address destination) {
  * @param lossPercent loss in percents
  * @param meanDelay mean delay
  */
- public void outboundSettings(Address destination, int lossPercent, int meanDelay) {
+ public void outboundSettings(String destination, int lossPercent, int meanDelay) {
  OutboundSettings settings = new OutboundSettings(lossPercent, meanDelay);
  outboundSettings.put(destination, settings);
  LOGGER.debug("[{}] Set outbound settings {} to {}", address, settings, destination);
@@ -103,7 +102,7 @@ public void unblockAllOutbound() {
  *
  * @param destinations collection of target endpoints where to apply
  */
- public void blockOutbound(Address... destinations) {
+ public void blockOutbound(String... destinations) {
  blockOutbound(Arrays.asList(destinations));
  }
 
@@ -112,8 +111,8 @@ public void blockOutbound(Address... destinations) {
  *
  * @param destinations collection of target endpoints where to apply
  */
- public void blockOutbound(Collection<Address> destinations) {
- for (Address destination : destinations) {
+ public void blockOutbound(Collection<String> destinations) {
+ for (String destination : destinations) {
  outboundSettings.put(destination, new OutboundSettings(100, 0));
  }
  LOGGER.debug("[{}] Blocked outbound to {}", address, destinations);
@@ -124,7 +123,7 @@ public void blockOutbound(Collection<Address> destinations) {
  *
  * @param destinations collection of target endpoints where to apply
  */
- public void unblockOutbound(Address... destinations) {
+ public void unblockOutbound(String... destinations) {
  unblockOutbound(Arrays.asList(destinations));
  }
 
@@ -133,7 +132,7 @@ public void unblockOutbound(Address... destinations) {
  *
  * @param destinations collection of target endpoints where to apply
  */
- public void unblockOutbound(Collection<Address> destinations) {
+ public void unblockOutbound(Collection<String> destinations) {
  destinations.forEach(outboundSettings::remove);
  LOGGER.debug("[{}] Unblocked outbound {}", address, destinations);
  }
@@ -164,7 +163,7 @@ public long totalOutboundMessageLostCount() {
  * @param address target address
  * @return mono message
  */
- public Mono<Message> tryFailOutbound(Message msg, Address address) {
+ public Mono<Message> tryFailOutbound(Message msg, String address) {
  return Mono.defer(
  () -> {
  totalMessageSentCount.incrementAndGet();
@@ -187,7 +186,7 @@ public Mono<Message> tryFailOutbound(Message msg, Address address) {
  * @param address target address
  * @return mono message
  */
- public Mono<Message> tryDelayOutbound(Message msg, Address address) {
+ public Mono<Message> tryDelayOutbound(Message msg, String address) {
  return Mono.defer(
  () -> {
  totalMessageSentCount.incrementAndGet();
@@ -209,7 +208,7 @@ public Mono<Message> tryDelayOutbound(Message msg, Address address) {
  * @param destination address of target endpoint
  * @return network inbound settings
  */
- public InboundSettings inboundSettings(Address destination) {
+ public InboundSettings inboundSettings(String destination) {
  return inboundSettings.getOrDefault(destination, defaultInboundSettings);
  }
 
@@ -218,7 +217,7 @@ public InboundSettings inboundSettings(Address destination) {
  *
  * @param shallPass shallPass inbound flag
  */
- public void inboundSettings(Address destination, boolean shallPass) {
+ public void inboundSettings(String destination, boolean shallPass) {
  InboundSettings settings = new InboundSettings(shallPass);
  inboundSettings.put(destination, settings);
  LOGGER.debug("[{}] Set inbound settings {} to {}", address, settings, destination);
@@ -253,7 +252,7 @@ public void unblockAllInbound() {
  *
  * @param destinations collection of target endpoints where to apply
  */
- public void blockInbound(Address... destinations) {
+ public void blockInbound(String... destinations) {
  blockInbound(Arrays.asList(destinations));
  }
 
@@ -262,8 +261,8 @@ public void blockInbound(Address... destinations) {
  *
  * @param destinations collection of target endpoints where to apply
  */
- public void blockInbound(Collection<Address> destinations) {
- for (Address destination : destinations) {
+ public void blockInbound(Collection<String> destinations) {
+ for (String destination : destinations) {
  inboundSettings.put(destination, new InboundSettings(false));
  }
  LOGGER.debug("[{}] Blocked inbound from {}", address, destinations);
@@ -274,7 +273,7 @@ public void blockInbound(Collection<Address> destinations) {
  *
  * @param destinations collection of target endpoints where to apply
  */
- public void unblockInbound(Address... destinations) {
+ public void unblockInbound(String... destinations) {
  unblockInbound(Arrays.asList(destinations));
  }
 
@@ -283,7 +282,7 @@ public void unblockInbound(Address... destinations) {
  *
  * @param destinations collection of target endpoints where to apply
  */
- public void unblockInbound(Collection<Address> destinations) {
+ public void unblockInbound(Collection<String> destinations) {
  destinations.forEach(inboundSettings::remove);
  LOGGER.debug("[{}] Unblocked inbound from {}", address, destinations);
  }

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

@@ -2,7 +2,6 @@
 
 import io.scalecube.cluster.transport.api.Message;
 import io.scalecube.cluster.transport.api.Transport;
-import io.scalecube.net.Address;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -26,7 +25,7 @@ public NetworkEmulator networkEmulator() {
  }
 
  @Override
- public Address address() {
+ public String address() {
  return transport.address();
  }
 
@@ -46,7 +45,7 @@ public boolean isStopped() {
  }
 
  @Override
- public Mono<Void> send(Address address, Message message) {
+ public Mono<Void> send(String address, Message message) {
  return Mono.defer(
  () ->
  Mono.just(enhanceWithSender(message))
@@ -56,7 +55,7 @@ public Mono<Void> send(Address address, Message message) {
  }
 
  @Override
- public Mono<Message> requestResponse(Address address, Message request) {
+ public Mono<Message> requestResponse(String address, Message request) {
  return Mono.defer(
  () ->
  Mono.just(enhanceWithSender(request))