IGNITE-61 - Direct marshalling

modules/codegen/src/main/java/org/apache/ignite/codegen/CommunicationMessageCodeGenerator.java

@@ -536,7 +536,7 @@ else if (type.isArray()) {
             returnFalseIfFailed(write, "writer.writeObjectArray", field, name,
                 type.getComponentType().getSimpleName() + ".class");
         }
-        else if (Collection.class.isAssignableFrom(type)) {
+        else if (Collection.class.isAssignableFrom(type) && !Set.class.isAssignableFrom(type)) {
             assert colItemType != null;
 
             returnFalseIfFailed(write, "writer.writeCollection", field, name, colItemType.getSimpleName() + ".class");
@@ -613,7 +613,7 @@ else if (type.isArray()) {
             returnFalseIfReadFailed(name, "reader.readObjectArray", field,
                 type.getComponentType().getSimpleName() + ".class");
         }
-        else if (Collection.class.isAssignableFrom(type)) {
+        else if (Collection.class.isAssignableFrom(type) && !Set.class.isAssignableFrom(type)) {
             assert colItemType != null;
 
             returnFalseIfReadFailed(name, "reader.readCollection", field,

modules/core/src/main/java/org/apache/ignite/internal/direct/GridTcpCommunicationMessageFactory.java

@@ -51,7 +51,7 @@
  */
 public class GridTcpCommunicationMessageFactory {
     /** Common message producers. */
-    private static final GridTcpCommunicationMessageProducer[] COMMON = new GridTcpCommunicationMessageProducer[83];
+    private static final GridTcpCommunicationMessageProducer[] COMMON = new GridTcpCommunicationMessageProducer[89];
 
     /**
      * Custom messages registry. Used for test purposes.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java

@@ -1136,7 +1136,7 @@ void onResult(GridDhtLockResponse<K, V> res) {
                     evictReaders(cctx, res.nearEvicted(), node.id(), res.messageId(), nearMapping);
                 }
 
-                Set<Integer> invalidParts = res.invalidPartitions();
+                Collection<Integer> invalidParts = res.invalidPartitions();
 
                 // Removing mappings for invalid partitions.
                 if (!F.isEmpty(invalidParts)) {

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockResponse.java

@@ -55,7 +55,7 @@ public class GridDhtLockResponse<K, V> extends GridDistributedLockResponse<K, V>
     /** Invalid partitions. */
     @GridToStringInclude
     @GridDirectCollection(int.class)
-    private Set<Integer> invalidParts = new GridLeanSet<>();
+    private Collection<Integer> invalidParts = new GridLeanSet<>();
 
     @GridDirectTransient
     /** Preload entries. */
@@ -138,7 +138,7 @@ public void addInvalidPartition(int part) {
     /**
      * @return Invalid partitions.
      */
-    public Set<Integer> invalidPartitions() {
+    public Collection<Integer> invalidPartitions() {
         return invalidParts;
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemandMessage.java

@@ -41,7 +41,7 @@ public class GridDhtPartitionDemandMessage<K, V> extends GridCacheMessage<K, V>
     /** Partition. */
     @GridToStringInclude
     @GridDirectCollection(int.class)
-    private Set<Integer> parts;
+    private Collection<Integer> parts;
 
     /** Topic. */
     @GridDirectTransient
@@ -112,7 +112,7 @@ void addPartition(int p) {
     /**
      * @return Partition.
      */
-    Set<Integer> partitions() {
+    Collection<Integer> partitions() {
         return parts;
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplyMessage.java

@@ -46,12 +46,12 @@ public class GridDhtPartitionSupplyMessage<K, V> extends GridCacheMessage<K, V>
 
     /** Partitions that have been fully sent. */
     @GridDirectCollection(int.class)
-    private Set<Integer> last;
+    private Collection<Integer> last;
 
     /** Partitions which were not found. */
     @GridToStringInclude
     @GridDirectCollection(int.class)
-    private Set<Integer> missed;
+    private Collection<Integer> missed;
 
     /** Entries. */
     @GridDirectTransient
@@ -130,7 +130,7 @@ boolean ack() {
     /**
      * @return Flag to indicate last message for partition.
      */
-    Set<Integer> last() {
+    Collection<Integer> last() {
         return last == null ? Collections.<Integer>emptySet() : last;
     }
 
@@ -166,7 +166,7 @@ void missed(int p) {
     /**
      * @return Missed partitions.
      */
-    Set<Integer> missed() {
+    Collection<Integer> missed() {
         return missed == null ? Collections.<Integer>emptySet() : missed;
     }
 

modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageAdapter.java

@@ -17,6 +17,8 @@
 
 package org.apache.ignite.plugin.extensions.communication;
 
+import org.apache.ignite.internal.direct.*;
+
 import java.io.*;
 import java.nio.*;
 
@@ -25,7 +27,7 @@
  */
 public abstract class MessageAdapter implements Serializable, Cloneable {
     /** Writer. */
-    protected MessageWriter writer;
+    protected final MessageWriter writer = new DirectMessageWriter();
 
     /** Reader. */
     protected MessageReader reader;
@@ -40,8 +42,8 @@ public abstract class MessageAdapter implements Serializable, Cloneable {
      * @param writer Writer.
      */
     public final void setWriter(MessageWriter writer) {
-        if (this.writer == null)
-            this.writer = writer;
+//        if (this.writer == null)
+//            this.writer = writer;
     }
 
     /**