Introducing NIO server

Introducing NIO server.

Added:

- Netty server instead of simple IO-based.

Changed:

- The way of serialization/deserialization of EPMD messages.

Author: xxlabaza

.codestyle/checkstyle.xml

@@ -191,7 +191,7 @@ limitations under the License.
     </module>
     <module name="MethodLength">
       <property name="tokens" value="METHOD_DEF"/>
-      <property name="max" value="30"/>
+      <property name="max" value="60"/>
       <property name="countEmpty" value="false"/>
     </module>
     <module name="ParameterNumber">

CHANGELOG.md

@@ -11,6 +11,18 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 - Add more unit and integration tests.
 
+## [0.4.0](https://github.com/appulse-projects/epmd-java/releases/tag/0.4.0) - 2018-02-16
+
+Introducing NIO server.
+
+### Added
+
+- Netty server instead of simple IO-based.
+
+### Changed
+
+- The way of serialization/deserialization of EPMD messages.
+
 ## [0.3.3](https://github.com/appulse-projects/epmd-java/releases/tag/0.3.3) - 2018-02-09
 
 Minor fix, using unsigned numbers where they are needed

client/README.md

@@ -14,7 +14,7 @@ Include the dependency to your project's pom.xml file:
     <dependency>
         <groupId>io.appulse.epmd.java</groupId>
         <artifactId>client</artifactId>
-        <version>0.3.3</version>
+        <version>0.4.0</version>
     </dependency>
     ...
 </dependencies>
@@ -23,7 +23,7 @@ Include the dependency to your project's pom.xml file:
 or Gradle:
 
 ```groovy
-compile 'io.appulse.epmd.java:client:0.3.3'
+compile 'io.appulse.epmd.java:client:0.4.0'
 ```
 
 ### Create client

client/pom.xml

@@ -25,7 +25,7 @@ limitations under the License.
   <parent>
     <groupId>io.appulse</groupId>
     <artifactId>epmd-java</artifactId>
-    <version>0.3.3</version>
+    <version>0.4.0</version>
   </parent>
 
   <groupId>io.appulse.epmd.java</groupId>

core/pom.xml

@@ -25,7 +25,7 @@ limitations under the License.
   <parent>
     <groupId>io.appulse</groupId>
     <artifactId>epmd-java</artifactId>
-    <version>0.3.3</version>
+    <version>0.4.0</version>
   </parent>
 
   <groupId>io.appulse.epmd.java</groupId>

core/src/main/java/io/appulse/epmd/java/core/mapper/deserializer/DataDeserializer.java

@@ -18,8 +18,14 @@
 
 import io.appulse.epmd.java.core.mapper.DataSerializable;
 import io.appulse.epmd.java.core.mapper.deserializer.exception.DeserializationException;
+import io.appulse.epmd.java.core.mapper.deserializer.exception.InvalidReceivedMessageTagException;
+import io.appulse.epmd.java.core.model.Tag;
+import io.appulse.epmd.java.core.model.TaggedMessage;
 import io.appulse.utils.Bytes;
 
+import lombok.NonNull;
+import lombok.val;
+
 /**
  *
  * @author Artem Labazin
@@ -28,19 +34,30 @@
 class DataDeserializer implements Deserializer {
 
   @Override
-  public <T> T deserialize (Bytes bytes, Class<T> type) throws DeserializationException {
+  public <T> T deserialize (@NonNull Bytes bytes, @NonNull Class<T> type) throws DeserializationException {
     T result;
     try {
       result = type.newInstance();
     } catch (IllegalAccessException | InstantiationException ex) {
       throw new DeserializationException(ex);
     }
+
+    if (result instanceof TaggedMessage) {
+      val expectedTag = ((TaggedMessage) result).getTag();
+      val tag = Tag.of(bytes.getByte());
+      if (!expectedTag.equals(tag)) {
+        val message = String.format("Expected tag is: %s, but actual tag is: %s",
+                                    expectedTag, tag);
+        throw new InvalidReceivedMessageTagException(message);
+      }
+    }
+
     ((DataSerializable) result).read(bytes);
     return result;
   }
 
   @Override
-  public boolean isApplicable (Class<?> type) {
+  public boolean isApplicable (@NonNull Class<?> type) {
     return DataSerializable.class.isAssignableFrom(type);
   }
 }

core/src/main/java/io/appulse/epmd/java/core/mapper/deserializer/EnumDeserializer.java

@@ -18,13 +18,13 @@
 
 import static java.nio.charset.StandardCharsets.ISO_8859_1;
 import static java.util.Arrays.asList;
+import static java.util.Collections.synchronizedSet;
 import static java.util.Optional.empty;
 import static java.util.Optional.ofNullable;
 
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
-import java.util.Collections;
 import java.util.HashSet;
 import java.util.Optional;
 import java.util.Set;
@@ -46,12 +46,12 @@ class EnumDeserializer implements Deserializer {
   private static final Set<String> ENUM_UNKNOWN_VALUE;
 
   static {
-    ENUM_CREATE_METHODS_NAMES = Collections.synchronizedSet(new HashSet<>(asList(
+    ENUM_CREATE_METHODS_NAMES = synchronizedSet(new HashSet<>(asList(
         "of",
         "parse",
         "from"
     )));
-    ENUM_UNKNOWN_VALUE = Collections.synchronizedSet(new HashSet<>(asList(
+    ENUM_UNKNOWN_VALUE = synchronizedSet(new HashSet<>(asList(
         "UNDEFINED",
         "UNKNOWN"
     )));

core/src/main/java/io/appulse/epmd/java/core/mapper/deserializer/MessageDeserializer.java

@@ -16,23 +16,15 @@
 
 package io.appulse.epmd.java.core.mapper.deserializer;
 
-import static io.appulse.epmd.java.core.model.Tag.UNDEFINED;
-import static io.appulse.utils.BytesUtils.asInteger;
 import static java.util.Arrays.asList;
-import static java.util.Optional.ofNullable;
 
 import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 
-import io.appulse.epmd.java.core.mapper.Message;
-import io.appulse.epmd.java.core.mapper.deserializer.exception.InvalidReceivedMessageLengthException;
-import io.appulse.epmd.java.core.mapper.deserializer.exception.InvalidReceivedMessageTagException;
 import io.appulse.epmd.java.core.mapper.deserializer.exception.NoApplicableDeserializerException;
-import io.appulse.epmd.java.core.mapper.exception.MessageAnnotationMissingException;
-import io.appulse.epmd.java.core.model.Tag;
 import io.appulse.utils.Bytes;
 
-import lombok.SneakyThrows;
-import lombok.val;
+import lombok.NonNull;
 
 /**
  *
@@ -44,44 +36,22 @@ public final class MessageDeserializer {
   private static final List<Deserializer> DESERIALIZERS;
 
   static {
-    DESERIALIZERS = asList(
+    DESERIALIZERS = new CopyOnWriteArrayList<>(asList(
+        new RequestDeserializer(),
         new DataDeserializer(),
         new EnumDeserializer()
-    );
+    ));
   }
 
-  @SneakyThrows
-  public <T> T deserialize (byte[] bytes, Class<T> type) {
-    val buffer = ofNullable(bytes)
-        .map(Bytes::wrap)
-        .orElseThrow(NullPointerException::new);
-
-    val annotation = ofNullable(type)
-        .map(it -> it.getAnnotation(Message.class))
-        .orElseThrow(MessageAnnotationMissingException::new);
-
-    if (annotation.lengthBytes() > 0) {
-      val receivedMessageLength = asInteger(buffer.getBytes(annotation.lengthBytes()));
-      if (receivedMessageLength != buffer.remaining()) {
-        val message = String.format("Expected length is %d - %d bytes, but actual length is %d bytes.",
-                                    annotation.lengthBytes(), receivedMessageLength, buffer.remaining());
-        throw new InvalidReceivedMessageLengthException(message);
-      }
-    }
-
-    if (annotation.value() != UNDEFINED) {
-      val tag = Tag.of(buffer.getByte());
-      if (annotation.value() != tag) {
-        val message = String.format("Expected tag is: %s, but actual tag is: %s",
-                                    annotation.value(), tag);
-        throw new InvalidReceivedMessageTagException(message);
-      }
-    }
+  public <T> T deserialize (@NonNull byte[] bytes, @NonNull Class<T> type) {
+    return deserialize(Bytes.wrap(bytes), type);
+  }
 
+  public <T> T deserialize (@NonNull Bytes bytes, @NonNull Class<T> type) {
     return DESERIALIZERS.stream()
         .filter(it -> it.isApplicable(type))
         .findAny()
         .orElseThrow(NoApplicableDeserializerException::new)
-        .deserialize(buffer, type);
+        .deserialize(bytes, type);
   }
 }

core/src/main/java/io/appulse/epmd/java/core/mapper/deserializer/RequestDeserializer.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.appulse.epmd.java.core.mapper.deserializer;
+
+import io.appulse.epmd.java.core.mapper.deserializer.exception.DeserializationException;
+import io.appulse.epmd.java.core.mapper.deserializer.exception.InvalidReceivedMessageLengthException;
+import io.appulse.epmd.java.core.mapper.deserializer.exception.InvalidReceivedMessageTagException;
+import io.appulse.epmd.java.core.model.Tag;
+import io.appulse.epmd.java.core.model.request.Request;
+import io.appulse.utils.Bytes;
+
+import lombok.NonNull;
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
+
+/**
+ *
+ * @author Artem Labazin
+ * @since 0.4.0
+ */
+@Slf4j
+class RequestDeserializer implements Deserializer {
+
+  @Override
+  public <T> T deserialize (@NonNull Bytes bytes, @NonNull Class<T> type) throws DeserializationException {
+    val length = bytes.getShort();
+    if (length != bytes.remaining()) {
+      val message = String.format("Expected length is %d - %d bytes, but actual length is %d bytes.",
+                                  2, length, bytes.remaining());
+      log.error(message);
+      throw new InvalidReceivedMessageLengthException(message);
+    }
+
+    T result;
+    try {
+      result = type.newInstance();
+    } catch (IllegalAccessException | InstantiationException ex) {
+      log.error("Deserialized type instantiation error", ex);
+      throw new DeserializationException(ex);
+    }
+
+    if (!(result instanceof Request)) {
+      val message = String.format("Deserializing type '%s' is not an instance of '%s'",
+                                  type.getSimpleName(), Request.class.getSimpleName());
+      log.error(message);
+      throw new DeserializationException(message);
+    }
+    val request = (Request) result;
+
+    val tag = Tag.of(bytes.getByte());
+    if (tag != request.getTag()) {
+      val message = String.format("Expected tag is: %s, but actual tag is: %s",
+                                  request.getTag(), tag);
+      log.error(message);
+      throw new InvalidReceivedMessageTagException(message);
+    }
+
+    request.read(bytes);
+    return result;
+  }
+
+  @Override
+  public boolean isApplicable (@NonNull Class<?> type) {
+    return Request.class.isAssignableFrom(type);
+  }
+}

core/src/main/java/io/appulse/epmd/java/core/mapper/serializer/DataSerializer.java

@@ -18,8 +18,10 @@
 
 import io.appulse.epmd.java.core.mapper.DataSerializable;
 import io.appulse.epmd.java.core.mapper.serializer.exception.SerializationException;
+import io.appulse.epmd.java.core.model.TaggedMessage;
 import io.appulse.utils.Bytes;
 
+import lombok.NonNull;
 import lombok.val;
 
 /**
@@ -30,14 +32,18 @@
 class DataSerializer implements Serializer {
 
   @Override
-  public byte[] serialize (Object object, Class<?> type) throws SerializationException {
-    val body = Bytes.allocate();
-    ((DataSerializable) object).write(body);
-    return body.array();
+  public byte[] serialize (@NonNull Object object, @NonNull Class<?> type) throws SerializationException {
+    val serializable = (DataSerializable) object;
+    val bytes = serializable instanceof TaggedMessage
+                ? Bytes.allocate().put1B(((TaggedMessage) object).getTag().getCode())
+                : Bytes.allocate();
+
+    serializable.write(bytes);
+    return bytes.array();
   }
 
   @Override
-  public boolean isApplicable (Class<?> type) {
+  public boolean isApplicable (@NonNull Class<?> type) {
     return DataSerializable.class.isAssignableFrom(type);
   }
 }