WIP on jdk-cluster-codec and its externalizable dto implementaions

Author: artem-v

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

@@ -1,21 +1,26 @@
 package io.scalecube.cluster;
 
 import io.scalecube.net.Address;
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
 import java.util.Objects;
 import java.util.UUID;
 
 /**
  * Cluster member which represents node in the cluster and contains its id and address. This class
  * is essentially immutable.
  */
-public final class Member {
+public final class Member implements Externalizable {
+
+  private static final long serialVersionUID = 1L;
 
   private String id;
   private String alias;
   private Address address;
 
-  /** Instantiates empty member for deserialization purpose. */
-  Member() {}
+  public Member() {}
 
   /**
    * Constructor.
@@ -63,6 +68,16 @@ public int hashCode() {
     return Objects.hash(id, address);
   }
 
+  @Override
+  public void writeExternal(ObjectOutput out) throws IOException {
+    // todo
+  }
+
+  @Override
+  public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+    // todo
+  }
+
   @Override
   public String toString() {
     if (alias == null) {

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

@@ -1,10 +1,16 @@
 package io.scalecube.cluster.fdetector;
 
 import io.scalecube.cluster.Member;
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
 import java.util.StringJoiner;
 
 /** DTO class. Supports FailureDetector messages (Ping, Ack, PingReq). */
-final class PingData {
+final class PingData implements Externalizable {
+
+  private static final long serialVersionUID = 1L;
 
   enum AckType {
 
@@ -32,8 +38,7 @@ enum AckType {
   /** Ping response type. */
   private AckType ackType;
 
-  /** Instantiates empty ping data for deserialization purpose. */
-  PingData() {}
+  public PingData() {}
 
   private PingData(PingData other) {
     this.from = other.from;
@@ -78,6 +83,16 @@ public PingData withAckType(AckType ackType) {
     return p;
   }
 
+  @Override
+  public void writeExternal(ObjectOutput out) throws IOException {
+    // todo
+  }
+
+  @Override
+  public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+    // todo
+  }
+
   @Override
   public String toString() {
     return new StringJoiner(", ", PingData.class.getSimpleName() + "[", "]")

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

@@ -1,19 +1,24 @@
 package io.scalecube.cluster.gossip;
 
 import io.scalecube.cluster.transport.api.Message;
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
 import java.util.Objects;
 import java.util.StringJoiner;
 
 /** Data model for gossip, include gossip id, qualifier and object need to disseminate. */
-final class Gossip {
+final class Gossip implements Externalizable {
+
+  private static final long serialVersionUID = 1L;
 
   private String gossiperId;
   private Message message;
   // incremented counter
   private long sequenceId;
 
-  /** Instantiates empty gossip for deserialization purpose. */
-  Gossip() {}
+  public Gossip() {}
 
   public Gossip(String gossiperId, Message message, long sequenceId) {
     this.gossiperId = Objects.requireNonNull(gossiperId);
@@ -56,6 +61,16 @@ public int hashCode() {
     return Objects.hash(gossiperId, message, sequenceId);
   }
 
+  @Override
+  public void writeExternal(ObjectOutput out) throws IOException {
+    // todo
+  }
+
+  @Override
+  public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+    // todo
+  }
+
   @Override
   public String toString() {
     return new StringJoiner(", ", Gossip.class.getSimpleName() + "[", "]")

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

@@ -1,18 +1,23 @@
 package io.scalecube.cluster.gossip;
 
+import java.io.Externalizable;
+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.StringJoiner;
 
 /** Gossip request which be transmitted through the network, contains list of gossips. */
-final class GossipRequest {
+final class GossipRequest implements Externalizable {
+
+  private static final long serialVersionUID = 1L;
 
   private List<Gossip> gossips;
   private String from;
 
-  /** Instantiates empty gossip request for deserialization purpose. */
-  GossipRequest() {}
+  public GossipRequest() {}
 
   public GossipRequest(Gossip gossip, String from) {
     this(Collections.singletonList(gossip), from);
@@ -31,6 +36,16 @@ public String from() {
     return from;
   }
 
+  @Override
+  public void writeExternal(ObjectOutput out) throws IOException {
+    // todo
+  }
+
+  @Override
+  public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+    // todo
+  }
+
   @Override
   public String toString() {
     return new StringJoiner(", ", GossipRequest.class.getSimpleName() + "[", "]")

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

@@ -6,17 +6,22 @@
 import static io.scalecube.cluster.membership.MemberStatus.SUSPECT;
 
 import io.scalecube.cluster.Member;
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
 import java.util.Objects;
 
 /** Cluster membership record which represents member, status, and incarnation. */
-final class MembershipRecord {
+final class MembershipRecord implements Externalizable {
+
+  private static final long serialVersionUID = 1L;
 
   private Member member;
   private MemberStatus status;
   private int incarnation;
 
-  /** Instantiates empty membership record for deserialization purpose. */
-  MembershipRecord() {}
+  public MembershipRecord() {}
 
   /** Instantiates new instance of membership record with given member, status and incarnation. */
   public MembershipRecord(Member member, MemberStatus status, int incarnation) {
@@ -101,6 +106,16 @@ public int hashCode() {
     return Objects.hash(member, status, incarnation);
   }
 
+  @Override
+  public void writeExternal(ObjectOutput out) throws IOException {
+    // todo
+  }
+
+  @Override
+  public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+    // todo
+  }
+
   @Override
   public String toString() {
     return "{m: " + member + ", s: " + status + ", inc: " + incarnation + '}';

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

@@ -1,5 +1,9 @@
 package io.scalecube.cluster.membership;
 
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -9,7 +13,9 @@
  * A class containing full membership table from specific member and used full synchronization
  * between cluster members.
  */
-final class SyncData {
+final class SyncData implements Externalizable {
+
+  private static final long serialVersionUID = 1L;
 
   /** Full cluster membership table. */
   private List<MembershipRecord> membership;
@@ -19,8 +25,7 @@ final class SyncData {
    */
   private String syncGroup;
 
-  /** Instantiates empty sync data for deserialization purpose. */
-  SyncData() {}
+  public SyncData() {}
 
   public SyncData(Collection<MembershipRecord> membership, String syncGroup) {
     this.membership = new ArrayList<>(membership);
@@ -35,6 +40,16 @@ public String getSyncGroup() {
     return syncGroup;
   }
 
+  @Override
+  public void writeExternal(ObjectOutput out) throws IOException {
+    // todo
+  }
+
+  @Override
+  public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+    // todo
+  }
+
   @Override
   public String toString() {
     return new StringJoiner(", ", SyncData.class.getSimpleName() + "[", "]")

codec-parent/codec-jdk/pom.xml

@@ -0,0 +1,15 @@
+<?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">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>io.scalecube</groupId>
+    <artifactId>scalecube-codec-parent</artifactId>
+    <version>2.4.10-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>codec-jdk</artifactId>
+
+</project>

codec-parent/codec-jdk/src/main/java/io/scalecube/cluster/codec/jdk/JdkMessageCodec.java

@@ -0,0 +1,29 @@
+package io.scalecube.cluster.codec.jdk;
+
+import io.scalecube.cluster.transport.api.Message;
+import io.scalecube.cluster.transport.api.MessageCodec;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.OutputStream;
+
+public class JdkMessageCodec implements MessageCodec {
+
+  @Override
+  public Message deserialize(InputStream is) throws IOException, ClassNotFoundException {
+    Message message = new Message();
+    try (ObjectInputStream inputStream = new ObjectInputStream(is)) {
+      message.readExternal(inputStream);
+      return message;
+    }
+  }
+
+  @Override
+  public void serialize(Message message, OutputStream os) throws IOException {
+    try (ObjectOutputStream outputStream = new ObjectOutputStream(os)) {
+      message.writeExternal(outputStream);
+      outputStream.flush();
+    }
+  }
+}

codec-parent/codec-jdk/src/main/java/io/scalecube/cluster/codec/jdk/JdkMetadataCodec.java

@@ -0,0 +1,34 @@
+package io.scalecube.cluster.codec.jdk;
+
+import io.scalecube.cluster.metadata.MetadataCodec;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.nio.ByteBuffer;
+import reactor.core.Exceptions;
+
+public class JdkMetadataCodec implements MetadataCodec {
+
+  @Override
+  public Object deserialize(ByteBuffer buffer) {
+    byte[] bytes = buffer.array();
+    try (ObjectInputStream is = new ObjectInputStream(new ByteArrayInputStream(bytes))) {
+      return is.readObject();
+    } catch (Exception e) {
+      throw Exceptions.propagate(e);
+    }
+  }
+
+  @Override
+  public ByteBuffer serialize(Object metadata) {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    try (ObjectOutputStream os = new ObjectOutputStream(baos)) {
+      os.writeObject(metadata);
+      os.flush();
+      return ByteBuffer.wrap(baos.toByteArray());
+    } catch (Exception e) {
+      throw Exceptions.propagate(e);
+    }
+  }
+}

codec-parent/pom.xml

@@ -14,6 +14,7 @@
   <name>ScaleCube/ClusterCodec</name>
 
   <modules>
+    <module>codec-jdk</module>
     <module>codec-jackson</module>
   </modules>