WIP on externalizable

Author: artem-v

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

@@ -70,12 +70,29 @@ public int hashCode() {
 
   @Override
   public void writeExternal(ObjectOutput out) throws IOException {
-    // todo
+    // id
+    out.writeUTF(id);
+    // alias
+    boolean aliasNotNull = alias != null;
+    out.writeBoolean(aliasNotNull);
+    if (aliasNotNull) {
+      out.writeUTF(alias);
+    }
+    // address
+    out.writeUTF(address.toString());
   }
 
   @Override
-  public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-    // todo
+  public void readExternal(ObjectInput in) throws IOException {
+    // id
+    id = in.readUTF();
+    // alias
+    boolean aliasNotNull = in.readBoolean();
+    if (aliasNotNull) {
+      alias = in.readUTF();
+    }
+    // address
+    address = Address.from(in.readUTF());
   }
 
   @Override

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

@@ -1,12 +1,14 @@
 package io.scalecube.cluster.metadata;
 
+import io.scalecube.cluster.codec.jdk.JdkMetadataCodec;
 import io.scalecube.utils.ServiceLoaderUtil;
 import java.nio.ByteBuffer;
 
 /** Contains methods for metadata serializing/deserializing logic. */
 public interface MetadataCodec {
 
-  MetadataCodec INSTANCE = ServiceLoaderUtil.findFirst(MetadataCodec.class).orElse(null);
+  MetadataCodec INSTANCE =
+      ServiceLoaderUtil.findFirst(MetadataCodec.class).orElseGet(JdkMetadataCodec::new);
 
   /**
    * Deserializes metadata from buffer.

cluster-testlib/pom.xml

@@ -22,11 +22,6 @@
       <artifactId>scalecube-cluster-api</artifactId>
       <version>${project.version}</version>
     </dependency>
-    <dependency>
-      <groupId>io.scalecube</groupId>
-      <artifactId>scalecube-codec-jackson</artifactId>
-      <version>${project.version}</version>
-    </dependency>
   </dependencies>
 
 </project>

cluster/pom.xml

@@ -51,12 +51,6 @@
       <version>${log4j.version}</version>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>io.scalecube</groupId>
-      <artifactId>scalecube-codec-jackson</artifactId>
-      <version>${project.version}</version>
-      <scope>test</scope>
-    </dependency>
   </dependencies>
 
 </project>

cluster/src/main/java/io/scalecube/cluster/ClusterImpl.java

@@ -20,6 +20,8 @@
 import io.scalecube.cluster.transport.api.TransportConfig;
 import io.scalecube.net.Address;
 import io.scalecube.transport.netty.TransportImpl;
+import io.scalecube.utils.ServiceLoaderUtil;
+import java.io.Serializable;
 import java.lang.management.ManagementFactory;
 import java.nio.ByteBuffer;
 import java.util.Collection;
@@ -225,11 +227,7 @@ public Cluster startAwait() {
   }
 
   private Mono<Cluster> doStart() {
-    return Mono.defer(
-        () -> {
-          validateConfiguration();
-          return doStart0();
-        });
+    return Mono.fromRunnable(this::validateConfiguration).then(Mono.fromRunnable(this::doStart0));
   }
 
   private Mono<Cluster> doStart0() {
@@ -295,28 +293,31 @@ private Mono<Cluster> doStart0() {
   }
 
   private void validateConfiguration() {
-    MetadataDecoder metadataDecoder = config.metadataDecoder();
-    MetadataEncoder metadataEncoder = config.metadataEncoder();
-    MetadataCodec metadataCodec = config.metadataCodec();
-
-    if (metadataDecoder == null && metadataEncoder == null && metadataCodec == null) {
-      throw new IllegalArgumentException("Invalid cluster config");
-    }
+    final MetadataDecoder metadataDecoder = config.metadataDecoder();
+    final MetadataEncoder metadataEncoder = config.metadataEncoder();
+    final MetadataCodec metadataCodec =
+        ServiceLoaderUtil.findFirst(MetadataCodec.class).orElse(null);
 
     if (metadataDecoder != null && metadataEncoder != null && metadataCodec != null) {
-      throw new IllegalArgumentException("Invalid cluster config");
+      throw new IllegalArgumentException(
+          "Invalid cluster config: either pair of [metadataDecoder, metadataEncoder] "
+              + "or metadataCodec must be specified, not both");
     }
 
-    if (metadataCodec == null) {
-      Objects.requireNonNull(
-          metadataDecoder, "Invalid cluster config: metadataDecoder must be specified");
-      Objects.requireNonNull(
-          metadataEncoder, "Invalid cluster config: metadataEncoder must be specified");
+    if ((metadataDecoder == null && metadataEncoder != null)
+        || (metadataDecoder != null && metadataEncoder == null)) {
+      throw new IllegalArgumentException(
+          "Invalid cluster config: both of [metadataDecoder, metadataEncoder]  must be specified");
     }
 
     if (metadataDecoder == null && metadataEncoder == null) {
-      Objects.requireNonNull(
-          metadataCodec, "Invalid cluster config: metadataCodec must be specified");
+      if (metadataCodec == null) {
+        Object metadata = config.metadata();
+        if (metadata != null && !(metadata instanceof Serializable)) {
+          throw new IllegalArgumentException(
+              "Invalid cluster config: metadata must be Serializable");
+        }
+      }
     }
 
     Objects.requireNonNull(

cluster/src/main/java/io/scalecube/cluster/fdetector/PingData.java

@@ -85,12 +85,26 @@ public PingData withAckType(AckType ackType) {
 
   @Override
   public void writeExternal(ObjectOutput out) throws IOException {
-    // todo
+    // from
+    out.writeObject(from);
+    // to
+    out.writeObject(to);
+    // originalIssuer
+    out.writeObject(originalIssuer);
+    // ackType
+    out.writeObject(ackType);
   }
 
   @Override
   public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-    // todo
+    // from
+    from = (Member) in.readObject();
+    // to
+    to = (Member) in.readObject();
+    // originalIssuer
+    originalIssuer = (Member) in.readObject();
+    // ackType
+    ackType = (AckType) in.readObject();
   }
 
   @Override

cluster/src/main/java/io/scalecube/cluster/gossip/Gossip.java

@@ -63,12 +63,22 @@ public int hashCode() {
 
   @Override
   public void writeExternal(ObjectOutput out) throws IOException {
-    // todo
+    // gossiperId
+    out.writeUTF(gossiperId);
+    // message
+    out.writeObject(message);
+    // sequenceId
+    out.writeLong(sequenceId);
   }
 
   @Override
   public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-    // todo
+    // gossiperId
+    gossiperId = in.readUTF();
+    // message
+    message = (Message) in.readObject();
+    // sequenceId
+    sequenceId = in.readLong();
   }
 
   @Override

cluster/src/main/java/io/scalecube/cluster/gossip/GossipRequest.java

@@ -7,6 +7,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 import java.util.StringJoiner;
 
 /** Gossip request which be transmitted through the network, contains list of gossips. */
@@ -24,6 +25,8 @@ public GossipRequest(Gossip gossip, String from) {
   }
 
   public GossipRequest(List<Gossip> gossips, String from) {
+    Objects.requireNonNull(gossips);
+    Objects.requireNonNull(from);
     this.gossips = new ArrayList<>(gossips);
     this.from = from;
   }
@@ -38,12 +41,25 @@ public String from() {
 
   @Override
   public void writeExternal(ObjectOutput out) throws IOException {
-    // todo
+    // gossips
+    out.writeInt(gossips.size());
+    for (Gossip gossip : gossips) {
+      out.writeObject(gossip);
+    }
+    // from
+    out.writeUTF(from);
   }
 
   @Override
   public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-    // todo
+    // gossips
+    int size = in.readInt();
+    gossips = new ArrayList<>(size);
+    for (int i = 0; i < size; i++) {
+      gossips.add((Gossip) in.readObject());
+    }
+    // from
+    from = in.readUTF();
   }
 
   @Override

cluster/src/main/java/io/scalecube/cluster/membership/MembershipRecord.java

@@ -108,12 +108,22 @@ public int hashCode() {
 
   @Override
   public void writeExternal(ObjectOutput out) throws IOException {
-    // todo
+    // member
+    out.writeObject(member);
+    // status
+    out.writeObject(status);
+    // incarnation
+    out.writeInt(incarnation);
   }
 
   @Override
   public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-    // todo
+    // member
+    member = (Member) in.readObject();
+    // status
+    status = (MemberStatus) in.readObject();
+    // incarnation
+    incarnation = in.readInt();
   }
 
   @Override

cluster/src/main/java/io/scalecube/cluster/membership/SyncData.java

@@ -7,6 +7,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.Objects;
 import java.util.StringJoiner;
 
 /**
@@ -28,6 +29,8 @@ final class SyncData implements Externalizable {
   public SyncData() {}
 
   public SyncData(Collection<MembershipRecord> membership, String syncGroup) {
+    Objects.requireNonNull(membership);
+    Objects.requireNonNull(syncGroup);
     this.membership = new ArrayList<>(membership);
     this.syncGroup = syncGroup;
   }
@@ -42,12 +45,25 @@ public String getSyncGroup() {
 
   @Override
   public void writeExternal(ObjectOutput out) throws IOException {
-    // todo
+    // membership
+    out.writeInt(membership.size());
+    for (MembershipRecord record : membership) {
+      out.writeObject(record);
+    }
+    // syncGroup
+    out.writeUTF(syncGroup);
   }
 
   @Override
   public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-    // todo
+    // membership
+    int size = in.readInt();
+    membership = new ArrayList<>(size);
+    for (int i = 0; i < size; i++) {
+      membership.add((MembershipRecord) in.readObject());
+    }
+    // syncGroup
+    syncGroup = (String) in.readObject();
   }
 
   @Override