IGNITE-17874 Added List and ByteBuffer support to message serializati…

…on (#1189)
apache · Oct 12, 2022 · 2aeb633 · 2aeb633
1 parent c4c6821
commit 2aeb633
Show file tree

Hide file tree

Showing 22 changed files with 289 additions and 17 deletions.
diff --git a/...ain/java/org/apache/ignite/plugin/extensions/communication/MessageCollectionItemType.java b/...ain/java/org/apache/ignite/plugin/extensions/communication/MessageCollectionItemType.java
@@ -77,6 +77,9 @@ public enum MessageCollectionItemType {
     /** Bit set. */
     BIT_SET,
 
+    /** Byte buffer. */
+    BYTE_BUFFER,
+
     /** UUID. */
     UUID,
 

diff --git a/...main/java/org/apache/ignite/internal/network/processor/messages/MessageImplGenerator.java b/...main/java/org/apache/ignite/internal/network/processor/messages/MessageImplGenerator.java
@@ -694,5 +694,4 @@ private enum MaybeMessageType {
         MESSAGE,
         MAP;
     }
-
 }
diff --git a/...va/org/apache/ignite/internal/network/processor/serialization/BaseMethodNameResolver.java b/...va/org/apache/ignite/internal/network/processor/serialization/BaseMethodNameResolver.java
@@ -17,8 +17,10 @@
 
 package org.apache.ignite.internal.network.processor.serialization;
 
+import java.nio.ByteBuffer;
 import java.util.BitSet;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import javax.annotation.processing.ProcessingEnvironment;
@@ -124,8 +126,12 @@ private String resolveReferenceMethodName(DeclaredType parameterType) {
             return "BitSet";
         } else if (typeUtils.isSameType(parameterType, Collection.class)) {
             return "Collection";
+        } else if (typeUtils.isSameType(parameterType, List.class)) {
+            return "List";
         } else if (typeUtils.isSameType(parameterType, Map.class)) {
             return "Map";
+        } else if (typeUtils.isSameType(parameterType, ByteBuffer.class)) {
+            return "ByteBuffer";
         } else {
             throw new ProcessingException("Unsupported reference type for message (de-)serialization: " + parameterType);
         }

diff --git a/...e/ignite/internal/network/processor/serialization/MessageCollectionItemTypeConverter.java b/...e/ignite/internal/network/processor/serialization/MessageCollectionItemTypeConverter.java
@@ -17,6 +17,7 @@
 
 package org.apache.ignite.internal.network.processor.serialization;
 
+import java.nio.ByteBuffer;
 import java.util.BitSet;
 import java.util.UUID;
 import javax.annotation.processing.ProcessingEnvironment;
@@ -125,6 +126,8 @@ private MessageCollectionItemType fromDeclaredType(DeclaredType parameterType) {
             return MessageCollectionItemType.MSG;
         } else if (typeUtils.isSameType(parameterType, BitSet.class)) {
             return MessageCollectionItemType.BIT_SET;
+        } else if (typeUtils.isSameType(parameterType, ByteBuffer.class)) {
+            return MessageCollectionItemType.BYTE_BUFFER;
         } else {
             throw new ProcessingException("Unsupported MessageCollectionItemType: " + parameterType);
         }

diff --git a/...g/apache/ignite/internal/network/processor/serialization/MessageReaderMethodResolver.java b/...g/apache/ignite/internal/network/processor/serialization/MessageReaderMethodResolver.java
@@ -75,6 +75,8 @@ CodeBlock resolveReadMethod(ExecutableElement getter) {
                 return resolveReadObjectArray((ArrayType) parameterType, parameterName);
             case "Collection":
                 return resolveReadCollection((DeclaredType) parameterType, parameterName);
+            case "List":
+                return resolveReadList((DeclaredType) parameterType, parameterName);
             case "Map":
                 return resolveReadMap((DeclaredType) parameterType, parameterName);
             default:
@@ -115,6 +117,22 @@ private CodeBlock resolveReadCollection(DeclaredType parameterType, String param
                 .build();
     }
 
+    /**
+     * Creates a {@link MessageReader#readList(String, MessageCollectionItemType)} method call.
+     */
+    private CodeBlock resolveReadList(DeclaredType parameterType, String parameterName) {
+        TypeMirror listGenericType = parameterType.getTypeArguments().get(0);
+
+        return CodeBlock.builder()
+                .add(
+                        "readList($S, $T.$L)",
+                        parameterName,
+                        MessageCollectionItemType.class,
+                        typeConverter.fromTypeMirror(listGenericType)
+                )
+                .build();
+    }
+
     /**
      * Creates a {@link MessageReader#readMap(String, MessageCollectionItemType, MessageCollectionItemType, boolean)} method call.
      */

diff --git a/...g/apache/ignite/internal/network/processor/serialization/MessageWriterMethodResolver.java b/...g/apache/ignite/internal/network/processor/serialization/MessageWriterMethodResolver.java
@@ -87,6 +87,8 @@ CodeBlock resolveWriteMethod(ExecutableElement getter) {
                 return resolveWriteObjectArray((ArrayType) getterReturnType, parameterName);
             case "Collection":
                 return resolveWriteCollection((DeclaredType) getterReturnType, parameterName);
+            case "List":
+                return resolveWriteList((DeclaredType) getterReturnType, parameterName);
             case "Map":
                 return resolveWriteMap((DeclaredType) getterReturnType, parameterName);
             default:
@@ -130,6 +132,23 @@ private CodeBlock resolveWriteCollection(DeclaredType parameterType, String para
                 .build();
     }
 
+    /**
+     * Creates a {@link MessageWriter#writeList(String, List, MessageCollectionItemType)} method call.
+     */
+    private CodeBlock resolveWriteList(DeclaredType parameterType, String parameterName) {
+        TypeMirror listGenericType = parameterType.getTypeArguments().get(0);
+
+        return CodeBlock.builder()
+                .add(
+                        "writeList($S, message.$L(), $T.$L)",
+                        parameterName,
+                        parameterName,
+                        MessageCollectionItemType.class,
+                        typeConverter.fromTypeMirror(listGenericType)
+                )
+                .build();
+    }
+
     /**
      * Creates a {@link MessageWriter#writeMap(String, Map, MessageCollectionItemType, MessageCollectionItemType)} method call.
      */

diff --git a/modules/network-api/src/main/java/org/apache/ignite/network/serialization/MessageReader.java b/modules/network-api/src/main/java/org/apache/ignite/network/serialization/MessageReader.java
@@ -21,6 +21,7 @@
 import java.util.BitSet;
 import java.util.Collection;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import org.apache.ignite.lang.IgniteUuid;
@@ -213,6 +214,14 @@ public interface MessageReader {
      */
     public BitSet readBitSet(String name);
 
+    /**
+     * Reads a {@link ByteBuffer}.
+     *
+     * @param name Field name.
+     * @return {@link ByteBuffer}.
+     */
+    ByteBuffer readByteBuffer(String name);
+
     /**
      * Reads an {@link UUID}.
      *
@@ -259,6 +268,16 @@ public interface MessageReader {
      */
     public <C extends Collection<?>> C readCollection(String name, MessageCollectionItemType itemType);
 
+    /**
+     * Reads a list.
+     *
+     * @param <C>      Type of collection.
+     * @param name     Field name.
+     * @param itemType An item type of the Collection.
+     * @return Collection.
+     */
+    public <C extends List<?>> C readList(String name, MessageCollectionItemType itemType);
+
     /**
      * Reads a map.
      *

diff --git a/modules/network-api/src/main/java/org/apache/ignite/network/serialization/MessageWriter.java b/modules/network-api/src/main/java/org/apache/ignite/network/serialization/MessageWriter.java
@@ -20,6 +20,7 @@
 import java.nio.ByteBuffer;
 import java.util.BitSet;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import org.apache.ignite.lang.IgniteUuid;
@@ -237,6 +238,15 @@ public interface MessageWriter {
      */
     public boolean writeBitSet(String name, BitSet val);
 
+    /**
+     * Writes a {@link ByteBuffer}.
+     *
+     * @param name Field name.
+     * @param val {@link ByteBuffer}.
+     * @return Whether a value was fully written.
+     */
+    boolean writeByteBuffer(String name, ByteBuffer val);
+
     /**
      * Writes an {@link UUID}.
      *
@@ -286,6 +296,17 @@ public interface MessageWriter {
      */
     public <T> boolean writeCollection(String name, Collection<T> col, MessageCollectionItemType itemType);
 
+    /**
+     * Writes a list.
+     *
+     * @param <T>      Type of collection.
+     * @param name     Field name.
+     * @param col      Collection.
+     * @param itemType An item type of the collection.
+     * @return Whether a value was fully written.
+     */
+    public <T> boolean writeList(String name, List<T> col, MessageCollectionItemType itemType);
+
     /**
      * Writes a map.
      *

diff --git a/...twork/src/main/java/org/apache/ignite/internal/network/direct/DirectMarshallingUtils.java b/...twork/src/main/java/org/apache/ignite/internal/network/direct/DirectMarshallingUtils.java
@@ -46,5 +46,4 @@ public static short getShort(ByteBuffer buffer) {
     private static short asShort(byte b0, byte b1) {
         return (short) ((b1 & 0xFF) << 8 | b0 & 0xFF);
     }
-
 }
diff --git a/.../network/src/main/java/org/apache/ignite/internal/network/direct/DirectMessageReader.java b/.../network/src/main/java/org/apache/ignite/internal/network/direct/DirectMessageReader.java
@@ -20,6 +20,7 @@
 import java.nio.ByteBuffer;
 import java.util.BitSet;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import org.apache.ignite.internal.network.direct.state.DirectMessageState;
@@ -310,6 +311,17 @@ public BitSet readBitSet(String name) {
         return val;
     }
 
+    @Override
+    public ByteBuffer readByteBuffer(String name) {
+        DirectByteBufferStream stream = state.item().stream;
+
+        ByteBuffer val = stream.readByteBuffer();
+
+        lastRead = stream.lastFinished();
+
+        return val;
+    }
+
     /** {@inheritDoc} */
     @Override
     public UUID readUuid(String name) {
@@ -371,6 +383,20 @@ public <C extends Collection<?>> C readCollection(String name, MessageCollection
         return col;
     }
 
+    /** {@inheritDoc} */
+    @Override
+    public <C extends List<?>> C readList(String name, MessageCollectionItemType itemType) {
+        DirectByteBufferStream stream = state.item().stream;
+
+        Collection<?> col = stream.readCollection(itemType, this);
+
+        lastRead = stream.lastFinished();
+
+        assert col == null || col instanceof List : col;
+
+        return (C) col;
+    }
+
     /** {@inheritDoc} */
     @Override
     public <M extends Map<?, ?>> M readMap(String name, MessageCollectionItemType keyType,

diff --git a/.../network/src/main/java/org/apache/ignite/internal/network/direct/DirectMessageWriter.java b/.../network/src/main/java/org/apache/ignite/internal/network/direct/DirectMessageWriter.java
@@ -20,6 +20,7 @@
 import java.nio.ByteBuffer;
 import java.util.BitSet;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import org.apache.ignite.internal.network.direct.state.DirectMessageState;
@@ -284,6 +285,15 @@ public boolean writeBitSet(String name, BitSet val) {
         return stream.lastFinished();
     }
 
+    @Override
+    public boolean writeByteBuffer(String name, ByteBuffer val) {
+        DirectByteBufferStream stream = state.item().stream;
+
+        stream.writeByteBuffer(val);
+
+        return stream.lastFinished();
+    }
+
     /** {@inheritDoc} */
     @Override
     public boolean writeUuid(String name, UUID val) {
@@ -334,6 +344,16 @@ public <T> boolean writeCollection(String name, Collection<T> col, MessageCollec
         return stream.lastFinished();
     }
 
+    /** {@inheritDoc} */
+    @Override
+    public <T> boolean writeList(String name, List<T> col, MessageCollectionItemType itemType) {
+        DirectByteBufferStream stream = state.item().stream;
+
+        stream.writeCollection(col, itemType, this);
+
+        return stream.lastFinished();
+    }
+
     /** {@inheritDoc} */
     @Override
     public <K, V> boolean writeMap(String name, Map<K, V> map, MessageCollectionItemType keyType,

diff --git a/...rc/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStream.java b/...rc/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStream.java
@@ -196,6 +196,13 @@ public interface DirectByteBufferStream {
      */
     void writeBitSet(BitSet val);
 
+    /**
+     * Writes {@link ByteBuffer}.
+     *
+     * @param val Value.
+     */
+    void writeByteBuffer(ByteBuffer val);
+
     /**
      * Writes {@link UUID}.
      *
@@ -377,6 +384,13 @@ <K, V> void writeMap(Map<K, V> map, MessageCollectionItemType keyType, MessageCo
      */
     BitSet readBitSet();
 
+    /**
+     * Reads {@link ByteBuffer}.
+     *
+     * @return Value.
+     */
+    ByteBuffer readByteBuffer();
+
     /**
      * Reads {@link UUID}.
      *