Get rid of commons (#849)

* Removed RetryNonSerializedEmitFailureHandler
* Removed dependency on commons
* Added `Address` (copied from commons)

Author: artem-v

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/maven-v4_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/maven-v4_0_0.xsd">
   <modelVersion>4.0.0</modelVersion>
 
   <parent>
@@ -57,8 +59,7 @@
   </scm>
 
   <properties>
-    <scalecube-cluster.version>2.6.17</scalecube-cluster.version>
-    <scalecube-commons.version>1.0.24</scalecube-commons.version>
+    <scalecube-cluster.version>2.6.18.rc1</scalecube-cluster.version>
     <scalecube-security.version>1.0.32</scalecube-security.version>
 
     <reactor.version>2020.0.32</reactor.version>
@@ -93,13 +94,6 @@
 
   <dependencyManagement>
     <dependencies>
-      <!-- Scalecube commons -->
-      <dependency>
-        <groupId>io.scalecube</groupId>
-        <artifactId>scalecube-commons</artifactId>
-        <version>${scalecube-commons.version}</version>
-      </dependency>
-
       <!-- Scalecube security tokens -->
       <dependency>
         <groupId>io.scalecube</groupId>

services-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>
 
@@ -11,20 +13,11 @@
 
   <artifactId>scalecube-services-api</artifactId>
 
-  <properties>
-    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-  </properties>
-
   <dependencies>
-    <dependency>
-      <groupId>io.scalecube</groupId>
-      <artifactId>scalecube-commons</artifactId>
-    </dependency>
     <dependency>
       <groupId>io.projectreactor</groupId>
       <artifactId>reactor-core</artifactId>
     </dependency>
-
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>

services-api/src/main/java/io/scalecube/services/Address.java

@@ -0,0 +1,139 @@
+package io.scalecube.services;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class Address {
+
+  public static final Address NULL_ADDRESS = Address.create("nullhost", 0);
+
+  public static final Pattern ADDRESS_FORMAT = Pattern.compile("(?<host>^.*):(?<port>\\d+$)");
+
+  private String host;
+  private int port;
+
+  Address() {}
+
+  private Address(String host, int port) {
+    this.host = host;
+    this.port = port;
+  }
+
+  /**
+   * Parses given host:port string to create Address instance.
+   *
+   * @param hostandport must come in form {@code host:port}
+   */
+  public static Address from(String hostandport) {
+    if (hostandport == null || hostandport.isEmpty()) {
+      throw new IllegalArgumentException("host-and-port string must be present");
+    }
+
+    Matcher matcher = ADDRESS_FORMAT.matcher(hostandport);
+    if (!matcher.find()) {
+      throw new IllegalArgumentException("can't parse host-and-port string from: " + hostandport);
+    }
+
+    String host = matcher.group(1);
+    if (host == null || host.isEmpty()) {
+      throw new IllegalArgumentException("can't parse host from: " + hostandport);
+    }
+
+    int port;
+    try {
+      port = Integer.parseInt(matcher.group(2));
+    } catch (NumberFormatException ex) {
+      throw new IllegalArgumentException("can't parse port from: " + hostandport, ex);
+    }
+
+    return new Address(host, port);
+  }
+
+  /**
+   * Create address from host and port.
+   *
+   * @param host host
+   * @param port port
+   * @return address
+   */
+  public static Address create(String host, int port) {
+    return new Address(host, port);
+  }
+
+  /**
+   * Getting local IP address by the address of local host. <b>NOTE:</b> returned IP address is
+   * expected to be a publicly visible IP address.
+   *
+   * @throws RuntimeException wrapped {@link UnknownHostException}
+   */
+  public static InetAddress getLocalIpAddress() {
+    try {
+      return InetAddress.getLocalHost();
+    } catch (UnknownHostException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Returns host.
+   *
+   * @return host
+   */
+  public String host() {
+    return host;
+  }
+
+  /**
+   * Returns port.
+   *
+   * @return port
+   */
+  public int port() {
+    return port;
+  }
+
+  /**
+   * Returns new address instance with the specified port.
+   *
+   * @param port port
+   * @return address instance
+   */
+  public Address port(int port) {
+    return Address.create(host, port);
+  }
+
+  /**
+   * Returns new address instance with applied port offset.
+   *
+   * @param portOffset portOffset
+   * @return address instance
+   */
+  public Address withPortOffset(int portOffset) {
+    return Address.create(host, port + portOffset);
+  }
+
+  @Override
+  public boolean equals(Object other) {
+    if (this == other) {
+      return true;
+    }
+    if (other == null || getClass() != other.getClass()) {
+      return false;
+    }
+    Address that = (Address) other;
+    return Objects.equals(host, that.host) && Objects.equals(port, that.port);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(host, port);
+  }
+
+  @Override
+  public String toString() {
+    return host + ":" + port;
+  }
+}

services-api/src/main/java/io/scalecube/services/ServiceEndpoint.java

@@ -1,6 +1,5 @@
 package io.scalecube.services;
 
-import io.scalecube.net.Address;
 import java.io.Externalizable;
 import java.io.IOException;
 import java.io.ObjectInput;

services-api/src/main/java/io/scalecube/services/ServiceReference.java

@@ -1,6 +1,5 @@
 package io.scalecube.services;
 
-import io.scalecube.net.Address;
 import io.scalecube.services.api.Qualifier;
 import java.util.Collections;
 import java.util.HashMap;

services-api/src/main/java/io/scalecube/services/discovery/api/ServiceDiscovery.java

@@ -1,6 +1,6 @@
 package io.scalecube.services.discovery.api;
 
-import io.scalecube.net.Address;
+import io.scalecube.services.Address;
 import reactor.core.publisher.Flux;
 
 public interface ServiceDiscovery {

services-api/src/main/java/io/scalecube/services/gateway/Gateway.java

@@ -1,6 +1,6 @@
 package io.scalecube.services.gateway;
 
-import io.scalecube.net.Address;
+import io.scalecube.services.Address;
 import reactor.core.publisher.Mono;
 
 public interface Gateway {

services-api/src/main/java/io/scalecube/services/transport/api/ServerTransport.java

@@ -1,6 +1,6 @@
 package io.scalecube.services.transport.api;
 
-import io.scalecube.net.Address;
+import io.scalecube.services.Address;
 
 public interface ServerTransport {
 

services-discovery/src/main/java/io/scalecube/services/discovery/ScalecubeServiceDiscovery.java

@@ -1,9 +1,9 @@
 package io.scalecube.services.discovery;
 
-import static io.scalecube.reactor.RetryNonSerializedEmitFailureHandler.RETRY_NON_SERIALIZED;
 import static io.scalecube.services.discovery.api.ServiceDiscoveryEvent.newEndpointAdded;
 import static io.scalecube.services.discovery.api.ServiceDiscoveryEvent.newEndpointLeaving;
 import static io.scalecube.services.discovery.api.ServiceDiscoveryEvent.newEndpointRemoved;
+import static reactor.core.publisher.Sinks.EmitFailureHandler.busyLooping;
 
 import io.scalecube.cluster.Cluster;
 import io.scalecube.cluster.ClusterConfig;
@@ -14,11 +14,12 @@
 import io.scalecube.cluster.membership.MembershipConfig;
 import io.scalecube.cluster.membership.MembershipEvent;
 import io.scalecube.cluster.transport.api.TransportConfig;
-import io.scalecube.net.Address;
+import io.scalecube.services.Address;
 import io.scalecube.services.ServiceEndpoint;
 import io.scalecube.services.discovery.api.ServiceDiscovery;
 import io.scalecube.services.discovery.api.ServiceDiscoveryEvent;
 import java.nio.ByteBuffer;
+import java.time.Duration;
 import java.util.StringJoiner;
 import java.util.function.UnaryOperator;
 import org.slf4j.Logger;
@@ -89,7 +90,7 @@ public void onMembershipEvent(MembershipEvent event) {
 
   @Override
   public Address address() {
-    return cluster != null ? cluster.address() : null;
+    return cluster != null ? Address.from(cluster.address()) : null;
   }
 
   @Override
@@ -99,7 +100,7 @@ public Flux<ServiceDiscoveryEvent> listen() {
 
   @Override
   public void shutdown() {
-    sink.emitComplete(RETRY_NON_SERIALIZED);
+    sink.emitComplete(busyLooping(Duration.ofSeconds(3)));
     if (cluster != null) {
       cluster.shutdown();
     }
@@ -117,7 +118,7 @@ private void onMembershipEvent(MembershipEvent membershipEvent) {
     }
 
     LOGGER.debug("Publish discoveryEvent: {}", discoveryEvent);
-    sink.emitNext(discoveryEvent, RETRY_NON_SERIALIZED);
+    sink.emitNext(discoveryEvent, busyLooping(Duration.ofSeconds(3)));
   }
 
   private ServiceDiscoveryEvent toServiceDiscoveryEvent(MembershipEvent membershipEvent) {

services-discovery/src/test/java/io/scalecube/services/discovery/ScalecubeServiceDiscoveryTest.java

@@ -12,7 +12,7 @@
 import io.scalecube.cluster.membership.MembershipConfig;
 import io.scalecube.cluster.metadata.JdkMetadataCodec;
 import io.scalecube.cluster.metadata.MetadataCodec;
-import io.scalecube.net.Address;
+import io.scalecube.services.Address;
 import io.scalecube.services.ServiceEndpoint;
 import io.scalecube.services.ServiceMethodDefinition;
 import io.scalecube.services.ServiceRegistration;
@@ -234,13 +234,13 @@ private Mono<ServiceDiscovery> newServiceDiscovery(
                 .gossip(cfg -> GOSSIP_CONFIG)
                 .failureDetector(cfg -> FAILURE_DETECTOR_CONFIG)
                 .membership(cfg -> MEMBERSHIP_CONFIG)
-                .membership(cfg -> cfg.seedMembers(seedAddress)));
+                .membership(cfg -> cfg.seedMembers(seedAddress.toString())));
   }
 
   private void startSeed(MetadataCodec metadataCodec) {
     new ScalecubeServiceDiscovery()
         .transport(cfg -> cfg.transportFactory(new WebsocketTransportFactory()))
-        .membership(opts -> opts.seedMembers(SEED_ADDRESS))
+        .membership(opts -> opts.seedMembers(SEED_ADDRESS.toString()))
         .options(opts -> opts.metadata(newServiceEndpoint()))
         .options(opts -> opts.metadataCodec(metadataCodec))
         .gossip(cfg -> GOSSIP_CONFIG)