IGNITE-26829 Use MessageSerializer for GridTaskResultResponse (#12466)

Author: wernerdv

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java

@@ -99,6 +99,7 @@
 import org.apache.ignite.internal.codegen.GridQueryNextPageResponseSerializer;
 import org.apache.ignite.internal.codegen.GridTaskCancelRequestSerializer;
 import org.apache.ignite.internal.codegen.GridTaskResultRequestSerializer;
+import org.apache.ignite.internal.codegen.GridTaskResultResponseSerializer;
 import org.apache.ignite.internal.codegen.HandshakeMessageSerializer;
 import org.apache.ignite.internal.codegen.HandshakeWaitMessageSerializer;
 import org.apache.ignite.internal.codegen.IgniteDhtDemandedPartitionsMapSerializer;
@@ -355,7 +356,7 @@ public class GridIoMessageFactory implements MessageFactoryProvider {
         factory.register((short)62, DataStreamerRequest::new);
         factory.register((short)63, DataStreamerResponse::new);
         factory.register((short)76, GridTaskResultRequest::new, new GridTaskResultRequestSerializer());
-        factory.register((short)77, GridTaskResultResponse::new);
+        factory.register((short)77, GridTaskResultResponse::new, new GridTaskResultResponseSerializer());
         factory.register((short)78, MissingMappingRequestMessage::new, new MissingMappingRequestMessageSerializer());
         factory.register((short)79, MissingMappingResponseMessage::new, new MissingMappingResponseMessageSerializer());
         factory.register((short)80, MetadataRequestMessage::new, new MetadataRequestMessageSerializer());

modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskCommandHandler.java

@@ -132,10 +132,8 @@ public GridTaskCommandHandler(final GridKernalContext ctx) {
 
                         if (err != null)
                             res.error(err.getMessage());
-                        else {
-                            res.result(desc.result());
-                            res.resultBytes(U.marshal(ctx, desc.result()));
-                        }
+                        else
+                            res.marshalResult(ctx, desc.result());
                     }
                     else
                         res.found(false);
@@ -433,7 +431,7 @@ else if (!nodeId.equals(resHolderId))
                     res = (GridTaskResultResponse)msg;
 
                 try {
-                    res.result(U.unmarshal(ctx, res.resultBytes(), U.resolveClassLoader(ctx.config())));
+                    res.unmarshalResult(ctx);
                 }
                 catch (IgniteCheckedException e) {
                     U.error(log, "Failed to unmarshal task result: " + res, e);

modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskResultResponse.java

@@ -17,31 +17,34 @@
 
 package org.apache.ignite.internal.processors.rest.handlers.task;
 
-import java.nio.ByteBuffer;
-import org.apache.ignite.internal.GridDirectTransient;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.plugin.extensions.communication.MessageReader;
-import org.apache.ignite.plugin.extensions.communication.MessageWriter;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * Task result response.
  */
 public class GridTaskResultResponse implements Message {
     /** Result. */
-    @GridDirectTransient
-    private Object res;
+    private @Nullable Object res;
 
     /** Serialized result. */
+    @Order(value = 0, method = "resultBytes")
     private byte[] resBytes;
 
     /** Finished flag. */
+    @Order(1)
     private boolean finished;
 
     /** Flag indicating that task has ever been launched on node. */
+    @Order(2)
     private boolean found;
 
     /** Error. */
+    @Order(value = 3, method = "error")
     private String err;
 
     /**
@@ -51,13 +54,6 @@ public class GridTaskResultResponse implements Message {
         return res;
     }
 
-    /**
-     * @param res Task result.
-     */
-    public void result(@Nullable Object res) {
-        this.res = res;
-    }
-
     /**
      * @param resBytes Serialized result.
      */
@@ -115,90 +111,31 @@ public void error(String err) {
     }
 
     /** {@inheritDoc} */
-    @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
-        writer.setBuffer(buf);
-
-        if (!writer.isHeaderWritten()) {
-            if (!writer.writeHeader(directType()))
-                return false;
-
-            writer.onHeaderWritten();
-        }
-
-        switch (writer.state()) {
-            case 0:
-                if (!writer.writeString(err))
-                    return false;
-
-                writer.incrementState();
-
-            case 1:
-                if (!writer.writeBoolean(finished))
-                    return false;
-
-                writer.incrementState();
-
-            case 2:
-                if (!writer.writeBoolean(found))
-                    return false;
-
-                writer.incrementState();
-
-            case 3:
-                if (!writer.writeByteArray(resBytes))
-                    return false;
-
-                writer.incrementState();
-
-        }
-
-        return true;
+    @Override public short directType() {
+        return 77;
     }
 
-    /** {@inheritDoc} */
-    @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
-        reader.setBuffer(buf);
-
-        switch (reader.state()) {
-            case 0:
-                err = reader.readString();
-
-                if (!reader.isLastRead())
-                    return false;
-
-                reader.incrementState();
-
-            case 1:
-                finished = reader.readBoolean();
-
-                if (!reader.isLastRead())
-                    return false;
-
-                reader.incrementState();
-
-            case 2:
-                found = reader.readBoolean();
-
-                if (!reader.isLastRead())
-                    return false;
-
-                reader.incrementState();
-
-            case 3:
-                resBytes = reader.readByteArray();
-
-                if (!reader.isLastRead())
-                    return false;
+    /**
+     * Marshals task result to byte array.
+     *
+     * @param ctx Context.
+     * @param res Task result.
+     */
+    public void marshalResult(GridKernalContext ctx, @Nullable Object res) throws IgniteCheckedException {
+        resBytes = U.marshal(ctx, res);
+    }
 
-                reader.incrementState();
+    /**
+     * Unmarshals task result from byte array.
+     *
+     * @param ctx Context.
+     */
+    public void unmarshalResult(GridKernalContext ctx) throws IgniteCheckedException {
+        if (resBytes != null) {
+            res = U.unmarshal(ctx, resBytes, U.resolveClassLoader(ctx.config()));
 
+            // It is not required anymore.
+            resBytes = null;
         }
-
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override public short directType() {
-        return 77;
     }
 }