WIP on updated cluster RC

Author: artem-v

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

@@ -36,7 +36,7 @@ public class ServiceEndpoint implements Externalizable {
   public ServiceEndpoint() {}
 
   private ServiceEndpoint(Builder builder) {
-    this.id = Objects.requireNonNull(builder.id);
+    this.id = Objects.requireNonNull(builder.id, "ServiceEndpoint.id is required");
     this.address = builder.address;
     this.contentTypes = Collections.unmodifiableSet(new HashSet<>(builder.contentTypes));
     this.tags = Collections.unmodifiableMap(new HashMap<>(builder.tags));
@@ -96,7 +96,11 @@ public void writeExternal(ObjectOutput out) throws IOException {
     out.writeUTF(id);
 
     // address
-    out.writeUTF(address.toString());
+    boolean addressExists = address != null;
+    out.writeBoolean(addressExists);
+    if (addressExists) {
+      out.writeUTF(address.toString());
+    }
 
     // contentTypes
     out.writeInt(contentTypes.size());
@@ -124,12 +128,15 @@ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundExcept
     id = in.readUTF();
 
     // address
-    address = Address.from(in.readUTF());
+    boolean addressExists = in.readBoolean();
+    if (addressExists) {
+      address = Address.from(in.readUTF());
+    }
 
     // contentTypes
-    int capacitySize = in.readInt();
-    Set<String> contentTypes = new HashSet<>(capacitySize);
-    for (int i = 0; i < capacitySize; i++) {
+    int contentTypesSize = in.readInt();
+    Set<String> contentTypes = new HashSet<>(contentTypesSize);
+    for (int i = 0; i < contentTypesSize; i++) {
       contentTypes.add(in.readUTF());
     }
     this.contentTypes = Collections.unmodifiableSet(contentTypes);

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

@@ -47,7 +47,7 @@ public ServiceMethodDefinition(String action) {
    * @param auth is method protected by authentication
    */
   public ServiceMethodDefinition(String action, Map<String, String> tags, boolean auth) {
-    this.action = Objects.requireNonNull(action);
+    this.action = Objects.requireNonNull(action, "ServiceMethodDefinition.action is required");
     this.tags = Collections.unmodifiableMap(new HashMap<>(tags));
     this.auth = auth;
   }
@@ -90,7 +90,7 @@ public void writeExternal(ObjectOutput out) throws IOException {
   }
 
   @Override
-  public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+  public void readExternal(ObjectInput in) throws IOException {
     // namespace
     action = in.readUTF();
 

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

@@ -38,7 +38,7 @@ public ServiceRegistration() {}
    */
   public ServiceRegistration(
       String namespace, Map<String, String> tags, Collection<ServiceMethodDefinition> methods) {
-    this.namespace = Objects.requireNonNull(namespace);
+    this.namespace = Objects.requireNonNull(namespace, "ServiceRegistration.namespace is required");
     this.tags = Collections.unmodifiableMap(new HashMap<>(tags));
     this.methods = Collections.unmodifiableList(new ArrayList<>(methods));
   }

services-api/src/main/java/io/scalecube/services/exceptions/ServiceException.java

@@ -1,5 +1,7 @@
 package io.scalecube.services.exceptions;
 
+import java.util.StringJoiner;
+
 public abstract class ServiceException extends RuntimeException {
 
   private final int errorCode;
@@ -25,11 +27,9 @@ public int errorCode() {
 
   @Override
   public String toString() {
-    return getClass().getSimpleName()
-        + "{errorCode="
-        + errorCode
-        + ", errorMessage="
-        + getMessage()
-        + '}';
+    return new StringJoiner(", ", getClass().getSimpleName() + "[", "]")
+        .add("errorCode=" + errorCode)
+        .add("errorMessage=" + getMessage())
+        .toString();
   }
 }

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

@@ -21,6 +21,7 @@
 import javax.management.StandardMBean;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import reactor.core.Exceptions;
 import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
@@ -150,30 +151,29 @@ private ServiceDiscoveryEvent toServiceDiscoveryEvent(MembershipEvent membership
     ServiceDiscoveryEvent discoveryEvent = null;
 
     if (membershipEvent.isAdded() && membershipEvent.newMetadata() != null) {
-      ServiceEndpoint serviceEndpoint = decodeMetadata(membershipEvent.newMetadata());
-      discoveryEvent = ServiceDiscoveryEvent.newEndpointAdded(serviceEndpoint);
+      discoveryEvent =
+          ServiceDiscoveryEvent.newEndpointAdded(decodeMetadata(membershipEvent.newMetadata()));
     }
 
     if (membershipEvent.isRemoved() && membershipEvent.oldMetadata() != null) {
-      ServiceEndpoint serviceEndpoint = decodeMetadata(membershipEvent.oldMetadata());
-      discoveryEvent = ServiceDiscoveryEvent.newEndpointRemoved(serviceEndpoint);
+      discoveryEvent =
+          ServiceDiscoveryEvent.newEndpointRemoved(decodeMetadata(membershipEvent.oldMetadata()));
     }
 
     if (membershipEvent.isLeaving() && membershipEvent.newMetadata() != null) {
-      ServiceEndpoint serviceEndpoint = decodeMetadata(membershipEvent.newMetadata());
-      discoveryEvent = ServiceDiscoveryEvent.newEndpointLeaving(serviceEndpoint);
+      discoveryEvent =
+          ServiceDiscoveryEvent.newEndpointLeaving(decodeMetadata(membershipEvent.newMetadata()));
     }
 
     return discoveryEvent;
   }
 
   private ServiceEndpoint decodeMetadata(ByteBuffer byteBuffer) {
     try {
-      return (ServiceEndpoint) clusterConfig.metadataCodec().deserialize(byteBuffer);
-    } catch (Exception ex) {
-      // Gobble exception and don't ruin stack
-      LOGGER.error("Failed to read metadata: " + ex);
-      return null;
+      return (ServiceEndpoint) clusterConfig.metadataCodec().deserialize(byteBuffer.duplicate());
+    } catch (Exception e) {
+      LOGGER.error("Failed to read metadata: " + e);
+      throw Exceptions.propagate(e);
     }
   }
 

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

@@ -9,13 +9,22 @@
 import io.scalecube.cluster.fdetector.FailureDetectorConfig;
 import io.scalecube.cluster.gossip.GossipConfig;
 import io.scalecube.cluster.membership.MembershipConfig;
+import io.scalecube.cluster.metadata.JdkMetadataCodec;
 import io.scalecube.net.Address;
 import io.scalecube.services.ServiceEndpoint;
+import io.scalecube.services.ServiceMethodDefinition;
+import io.scalecube.services.ServiceRegistration;
 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.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Supplier;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Flux;
@@ -32,11 +41,61 @@ class ScalecubeServiceDiscoveryTest extends BaseTest {
   public static final MembershipConfig MEMBERSHIP_CONFIG = MembershipConfig.defaultLocalConfig();
   public static final int CLUSTER_SIZE = 3 + 1; // r1 + r2 + r3 (plus 1 for be sure)
 
+  private JdkMetadataCodec jdkMetadataCodec = new JdkMetadataCodec();
+
   @BeforeAll
   public static void setUp() {
     StepVerifier.setDefaultTimeout(TIMEOUT);
   }
 
+  @Test
+  public void testJdkMetadataCodec() {
+    ServiceEndpoint serviceEndpoint =
+        ServiceEndpoint.builder()
+            .id(UUID.randomUUID().toString())
+            .tags(Collections.singletonMap("K", "V"))
+            .contentTypes(Collections.singleton("json"))
+            .appendServiceRegistrations(
+                Collections.singletonList(
+                    new ServiceRegistration(
+                        "namespace",
+                        Collections.singletonMap("KK", "VV"),
+                        Collections.singletonList(
+                            new ServiceMethodDefinition(
+                                "action0", Collections.singletonMap("KKK0", "VVV"), true)))))
+            .appendServiceRegistrations(
+                Collections.singletonList(
+                    new ServiceRegistration(
+                        "namespace",
+                        Collections.singletonMap("KK", "VV"),
+                        Collections.singletonList(
+                            new ServiceMethodDefinition(
+                                "action1", Collections.singletonMap("KKK1", "VVV"), true)))))
+            .appendServiceRegistrations(
+                Collections.singletonList(
+                    new ServiceRegistration(
+                        "namespace",
+                        Collections.singletonMap("KK", "VV"),
+                        Collections.singletonList(
+                            new ServiceMethodDefinition(
+                                "action2", Collections.singletonMap("KKK2", "VVV"), true)))))
+            .build();
+
+    ByteBuffer buffer = jdkMetadataCodec.serialize(serviceEndpoint);
+    ServiceEndpoint serviceEndpoint1 = (ServiceEndpoint) jdkMetadataCodec.deserialize(buffer);
+    Assertions.assertEquals(serviceEndpoint.id(), serviceEndpoint1.id());
+    Assertions.assertEquals(1, serviceEndpoint1.tags().size());
+    Assertions.assertEquals(1, serviceEndpoint1.contentTypes().size());
+
+    List<ServiceRegistration> serviceRegistrations =
+        new ArrayList<>(serviceEndpoint1.serviceRegistrations());
+    Assertions.assertEquals(3, serviceRegistrations.size());
+    for (ServiceRegistration serviceRegistration : serviceRegistrations) {
+      Assertions.assertEquals(1, serviceRegistration.methods().size());
+      Assertions.assertEquals(1, serviceRegistration.tags().size());
+    }
+  }
+
   @Test
   public void testEndpointIsAddedThenRemoved() {
     Address seedAddress = startSeed();

services/src/test/java/io/scalecube/services/routings/sut/WeightedRandomRouter.java

@@ -13,7 +13,7 @@ public Optional<ServiceReference> route(ServiceRegistry serviceRegistry, Service
     RandomCollection<ServiceReference> weightedRandom = new RandomCollection<>();
     serviceRegistry
         .lookupService(request)
-        .forEach(sr -> weightedRandom.add(Double.valueOf(sr.tags().get("Weight")), sr));
+        .forEach(sr -> weightedRandom.add(Double.parseDouble(sr.tags().get("Weight")), sr));
     return Optional.of(weightedRandom.next());
   }
 }