Repository: scalecube/scalecube-services
Issue: Data type can be lost if message is passed via proxy node (e.g. gossip)

Since message doesn't contain field dataType, but only in deserialized form at MessageSchema when message pass via some proxy which can't resolve dataType via class by name (e.g. gossip which pass to some node which doesn't have required data model) then data type will be lost on this node and won't be transferred on next hop.


Changed files:
transport/src/main/java/io/servicefabric/transport/TransportHeaders.java, transport/src/main/java/io/servicefabric/transport/protocol/MessageSchema.java

diff --git a/transport/src/main/java/io/servicefabric/transport/TransportHeaders.java b/transport/src/main/java/io/servicefabric/transport/TransportHeaders.java
index 5bc3ef79..ef85fa9d 100644
--- a/transport/src/main/java/io/servicefabric/transport/TransportHeaders.java
+++ b/transport/src/main/java/io/servicefabric/transport/TransportHeaders.java
@@ -7,10 +7,23 @@ import rx.functions.Func1;
  */
 public final class TransportHeaders {
 
+  /**
+   * This header is supposed to be used by application in case if same data type can be reused for different message
+   * meanings so it will allow to qualify the specific meaning of data.
+   */
   public static final String QUALIFIER = "q";
 
+  /**
+   * This header is supposed to be used by application in order to correlate request and response messages.
+   */
   public static final String CORRELATION_ID = "cid";
 
+  /**
+   * This is a system header which used by transport for serialization and deserialization purpose. It is not supposed
+   * to be used by application directly and it is subject to changes in future releases.
+   */
+  public static final String DATA_TYPE = "_type";
+
   private TransportHeaders() {
     // Do not instantiate
   }
diff --git a/transport/src/main/java/io/servicefabric/transport/protocol/MessageSchema.java b/transport/src/main/java/io/servicefabric/transport/protocol/MessageSchema.java
index 249b16b9..299d3888 100644
--- a/transport/src/main/java/io/servicefabric/transport/protocol/MessageSchema.java
+++ b/transport/src/main/java/io/servicefabric/transport/protocol/MessageSchema.java
@@ -3,6 +3,7 @@ package io.servicefabric.transport.protocol;
 import static io.protostuff.LinkedBuffer.MIN_BUFFER_SIZE;
 import static io.servicefabric.transport.utils.RecyclableLinkedBuffer.DEFAULT_MAX_CAPACITY;
 
+import io.servicefabric.transport.TransportHeaders;
 import io.servicefabric.transport.utils.RecyclableLinkedBuffer;
 import io.servicefabric.transport.utils.memoization.Computable;
 import io.servicefabric.transport.utils.memoization.ConcurrentMapMemoizer;
@@ -34,15 +35,13 @@ final class MessageSchema implements Schema<Message> {
 
   private static final int HEADER_KEYS_FIELD_NUMBER = 1;
   private static final int HEADER_VALUES_FIELD_NUMBER = 2;
-  private static final int DATA_PAYLOAD_FIELD_NUMBER = 3;
-  private static final int DATA_TYPE_FIELD_NUMBER = 4;
+  private static final int DATA_FIELD_NUMBER = 3;
 
   private static final RecyclableLinkedBuffer recyclableLinkedBuffer = new RecyclableLinkedBuffer(MIN_BUFFER_SIZE,
       DEFAULT_MAX_CAPACITY);
 
   private static final Map<String, Integer> fieldMap = ImmutableMap.of("headerKeys", HEADER_KEYS_FIELD_NUMBER,
-      "headerValues", HEADER_VALUES_FIELD_NUMBER, "dataPayload", DATA_PAYLOAD_FIELD_NUMBER, "dataType",
-      DATA_TYPE_FIELD_NUMBER);
+      "headerValues", HEADER_VALUES_FIELD_NUMBER, "data", DATA_FIELD_NUMBER);
 
   private final ConcurrentMapMemoizer<String, Optional<Class>> classCache = new ConcurrentMapMemoizer<>(
       new Computable<String, Optional<Class>>() {
@@ -64,10 +63,8 @@ final class MessageSchema implements Schema<Message> {
         return "headerKeys";
       case HEADER_VALUES_FIELD_NUMBER:
         return "headerValues";
-      case DATA_PAYLOAD_FIELD_NUMBER:
-        return "dataPayload";
-      case DATA_TYPE_FIELD_NUMBER:
-        return "dataType";
+      case DATA_FIELD_NUMBER:
+        return "data";
       default:
         return null;
     }
@@ -109,8 +106,7 @@ final class MessageSchema implements Schema<Message> {
     boolean iterate = true;
     List<String> headerKeys = new ArrayList<>();
     List<String> headerValues = new ArrayList<>();
-    byte[] dataPayload = null;
-    String dataType = null;
+    byte[] dataBytes = null;
     while (iterate) {
       int number = input.readFieldNumber(this);
       switch (number) {
@@ -123,11 +119,8 @@ final class MessageSchema implements Schema<Message> {
         case HEADER_VALUES_FIELD_NUMBER:
           headerValues.add(input.readString());
           break;
-        case DATA_PAYLOAD_FIELD_NUMBER:
-          dataPayload = input.readByteArray();
-          break;
-        case DATA_TYPE_FIELD_NUMBER:
-          dataType = input.readString();
+        case DATA_FIELD_NUMBER:
+          dataBytes = input.readByteArray();
           break;
         default:
           input.handleUnknownField(number, this);
@@ -135,38 +128,43 @@ final class MessageSchema implements Schema<Message> {
     }
 
     // Deserialize headers
+    Map<String, String> headers = new HashMap<>(headerKeys.size());
     if (!headerKeys.isEmpty()) {
-      Map<String, String> headers = new HashMap<>(headerKeys.size());
       ListIterator<String> headerValuesIterator = headerValues.listIterator();
       for (String key : headerKeys) {
         String value = headerValuesIterator.next();
         headers.put(key, value);
       }
-      message.setHeaders(headers);
     }
 
     // Deserialize data
-    if (dataPayload != null) {
+    Object data = null;
+    if (dataBytes != null) {
+      String dataType = headers.get(TransportHeaders.DATA_TYPE);
       if (dataType == null) {
-        message.setData(dataPayload);
+        data = dataBytes;
       } else {
         Optional<Class> optionalDataClass = classCache.get(dataType);
         if (optionalDataClass.isPresent()) {
+          headers.remove(TransportHeaders.DATA_TYPE);
           Class<?> dataClass = optionalDataClass.get();
           Schema dataSchema = RuntimeSchema.getSchema(dataClass);
-          Object data = dataSchema.newMessage();
+          data = dataSchema.newMessage();
           try {
-            ProtostuffIOUtil.mergeFrom(dataPayload, data, dataSchema);
+            ProtostuffIOUtil.mergeFrom(dataBytes, data, dataSchema);
           } catch (Throwable e) {
             LOGGER.error("Failed to deserialize : {}", message);
             throw e;
           }
-          message.setData(data);
         } else {
-          message.setData(dataPayload);
+          data = dataBytes;
         }
       }
     }
+
+    // Set message
+    message.setHeaders(headers);
+    message.setData(data);
   }
 
   @Override
@@ -183,15 +181,20 @@ final class MessageSchema implements Schema<Message> {
     Object originalData = message.data();
     if (originalData != null) {
       if (originalData instanceof byte[]) {
-        output.writeByteArray(DATA_PAYLOAD_FIELD_NUMBER, (byte[]) originalData, false);
+        // Write data byte array as is
+        output.writeByteArray(DATA_FIELD_NUMBER, (byte[]) originalData, false);
       } else {
+        // Write data class as an additional header
         Class<?> dataClass = originalData.getClass();
+        output.writeString(HEADER_KEYS_FIELD_NUMBER, TransportHeaders.DATA_TYPE, true);
+        output.writeString(HEADER_VALUES_FIELD_NUMBER, dataClass.getName(), true);
+
+        // Write data as serialized byte array
         Schema dataSchema = RuntimeSchema.getSchema(dataClass);
         try (RecyclableLinkedBuffer rlb = recyclableLinkedBuffer.get()) {
           byte[] array = ProtostuffIOUtil.toByteArray(originalData, dataSchema, rlb.buffer());
-          output.writeByteArray(DATA_PAYLOAD_FIELD_NUMBER, array, false);
+          output.writeByteArray(DATA_FIELD_NUMBER, array, false);
         }
-        output.writeString(DATA_TYPE_FIELD_NUMBER, dataClass.getName(), false);
       }
     }
   }