IGNITE-28294 Fix nested Java collections serialization

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

@@ -442,14 +442,14 @@ public ByteBuffer getBuffer() {
     }
 
     /** {@inheritDoc} */
-    @Override public void beforeInnerMessageRead() {
+    @Override public void beforeNestedRead() {
         state.forward();
 
         state.item().stream.setBuffer(buf);
     }
 
     /** {@inheritDoc} */
-    @Override public void afterInnerMessageRead(boolean finished) {
+    @Override public void afterNestedRead(boolean finished) {
         state.backward(finished);
     }
 

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

@@ -420,14 +420,14 @@ public ByteBuffer getBuffer() {
     }
 
     /** {@inheritDoc} */
-    @Override public void beforeInnerMessageWrite() {
+    @Override public void beforeNestedWrite() {
         state.forward();
 
         state.item().stream.setBuffer(buf);
     }
 
     /** {@inheritDoc} */
-    @Override public void afterInnerMessageWrite(boolean finished) {
+    @Override public void afterNestedWrite(boolean finished) {
         state.backward(finished);
     }
 

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

@@ -923,12 +923,12 @@ public void writeMessage(Message msg, MessageWriter writer) {
         if (msg != null) {
             if (buf.hasRemaining()) {
                 try {
-                    writer.beforeInnerMessageWrite();
+                    writer.beforeNestedWrite();
 
                     lastFinished = msgFactory.serializer(msg.directType()).writeTo(msg, writer);
                 }
                 finally {
-                    writer.afterInnerMessageWrite(lastFinished);
+                    writer.afterNestedWrite(lastFinished);
                 }
             }
             else
@@ -1574,12 +1574,12 @@ public <T extends Message> T readMessage(MessageReader reader) {
 
         if (msg != null) {
             try {
-                reader.beforeInnerMessageRead();
+                reader.beforeNestedRead();
 
                 lastFinished = msgFactory.serializer(msg.directType()).readFrom(msg, reader);
             }
             finally {
-                reader.afterInnerMessageRead(lastFinished);
+                reader.afterNestedRead(lastFinished);
             }
         }
         else
@@ -2117,30 +2117,51 @@ protected <K, V> void write(MessageType type, Object val, MessageWriter writer)
                 break;
 
             case MAP:
-                writeMap((Map<K, V>)val, (MessageMapType)type, writer);
+                try {
+                    writer.beforeNestedWrite();
+
+                    lastFinished = writer.writeMap((Map<K, V>)val, (MessageMapType)type);
+                }
+                finally {
+                    writer.afterNestedWrite(lastFinished);
+                }
 
                 break;
 
             case COLLECTION:
-                writeCollection((Collection<V>)val, (MessageCollectionType)type, writer);
+                try {
+                    writer.beforeNestedWrite();
+
+                    lastFinished = writer.writeCollection((Collection<V>)val, (MessageCollectionType)type);
+                }
+                finally {
+                    writer.afterNestedWrite(lastFinished);
+                }
 
                 break;
 
             case ARRAY:
-                writeObjectArray((V[])val, (MessageArrayType)type, writer);
+                try {
+                    writer.beforeNestedWrite();
+
+                    lastFinished = writer.writeObjectArray((V[])val, (MessageArrayType)type);
+                }
+                finally {
+                    writer.afterNestedWrite(lastFinished);
+                }
 
                 break;
 
             case MSG:
                 try {
                     if (val != null)
-                        writer.beforeInnerMessageWrite();
+                        writer.beforeNestedWrite();
 
                     writeMessage((Message)val, writer);
                 }
                 finally {
                     if (val != null)
-                        writer.afterInnerMessageWrite(lastFinished);
+                        writer.afterNestedWrite(lastFinished);
                 }
 
                 break;
@@ -2230,13 +2251,46 @@ protected Object read(MessageType type, MessageReader reader) {
                 return readGridLongList();
 
             case MAP:
-                return readMap((MessageMapType)type, reader);
+                try {
+                    reader.beforeNestedRead();
+
+                    Object val = reader.readMap((MessageMapType)type);
+
+                    lastFinished = reader.isLastRead();
+
+                    return val;
+                }
+                finally {
+                    reader.afterNestedRead(lastFinished);
+                }
 
             case COLLECTION:
-                return readCollection((MessageCollectionType)type, reader);
+                try {
+                    reader.beforeNestedRead();
+
+                    Object val = reader.readCollection((MessageCollectionType)type);
+
+                    lastFinished = reader.isLastRead();
+
+                    return val;
+                }
+                finally {
+                    reader.afterNestedRead(lastFinished);
+                }
 
             case ARRAY:
-                return readObjectArray((MessageArrayType)type, reader);
+                try {
+                    reader.beforeNestedRead();
+
+                    Object val = reader.readObjectArray((MessageArrayType)type);
+
+                    lastFinished = reader.isLastRead();
+
+                    return val;
+                }
+                finally {
+                    reader.afterNestedRead(lastFinished);
+                }
 
             case MSG:
                 return readMessage(reader);

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

@@ -297,16 +297,16 @@ public default <T extends Message> T readMessage() {
     public void decrementState();
 
     /**
-     * Callback called before inner message is read.
+     * Callback called before nested value is read.
      */
-    public void beforeInnerMessageRead();
+    public void beforeNestedRead();
 
     /**
-     * Callback called after inner message is read.
+     * Callback called after nested value is read.
      *
-     * @param finished Whether message was fully read.
+     * @param finished Whether value was fully read.
      */
-    public void afterInnerMessageRead(boolean finished);
+    public void afterNestedRead(boolean finished);
 
     /**
      * Resets this reader.

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

@@ -356,16 +356,16 @@ public default <K, V> boolean writeMap(Map<K, V> map, MessageMapType type) {
     public void decrementState();
 
     /**
-     * Callback called before inner message is written.
+     * Callback called before nested value is written.
      */
-    public void beforeInnerMessageWrite();
+    public void beforeNestedWrite();
 
     /**
-     * Callback called after inner message is written.
+     * Callback called after nested value is written.
      *
-     * @param finished Whether message was fully written.
+     * @param finished Whether value was fully written.
      */
-    public void afterInnerMessageWrite(boolean finished);
+    public void afterNestedWrite(boolean finished);
 
     /**
      * Resets this writer.