TcpDiscoveryCollectionMessage fixes

Author: NSAmelchev

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryClientReconnectMessage.java

@@ -20,14 +20,11 @@
 import java.util.Collection;
 import java.util.Objects;
 import java.util.UUID;
-import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.managers.discovery.DiscoveryMessageFactory;
 import org.apache.ignite.internal.util.tostring.GridToStringExclude;
 import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteUuid;
-import org.apache.ignite.marshaller.Marshaller;
 import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
 
 /**
@@ -48,11 +45,8 @@ public class TcpDiscoveryClientReconnectMessage extends TcpDiscoveryAbstractMess
 
     /** Pending messages. */
     @GridToStringExclude
-    private Collection<TcpDiscoveryAbstractMessage> msgs;
-
-    /** Srialized bytes of {@link #msgs}. */
     @Order(2)
-    byte[] msgsBytes;
+    TcpDiscoveryCollectionMessage msgs;
 
     /** Constructor for {@link DiscoveryMessageFactory}. */
     public TcpDiscoveryClientReconnectMessage() {
@@ -89,14 +83,14 @@ public IgniteUuid lastMessageId() {
      * @param msgs Pending messages.
      */
     public void pendingMessages(Collection<TcpDiscoveryAbstractMessage> msgs) {
-        this.msgs = msgs;
+        this.msgs = new TcpDiscoveryCollectionMessage(msgs);
     }
 
     /**
      * @return Pending messages.
      */
     public Collection<TcpDiscoveryAbstractMessage> pendingMessages() {
-        return msgs;
+        return msgs.messages();
     }
 
     /**
@@ -128,18 +122,6 @@ public boolean success() {
             Objects.equals(lastMsgId, other.lastMsgId);
     }
 
-    /** {@inheritDoc} */
-    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
-        if (msgs != null)
-            msgsBytes = U.marshal(marsh, msgs);
-    }
-
-    /** {@inheritDoc} */
-    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
-        if (msgsBytes != null)
-            msgs = U.unmarshal(marsh, msgsBytes, clsLdr);
-    }
-
     /** {@inheritDoc} */
     @Override public short directType() {
         return 28;

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryCollectionMessage.java

@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.spi.discovery.tcp.messages;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.discovery.DiscoveryMessageFactory;
+import org.apache.ignite.internal.util.tostring.GridToStringExclude;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * TODO: Remove/revise after https://issues.apache.org/jira/browse/IGNITE-27627
+ * Message to transfer a collection of {@link TcpDiscoveryAbstractMessage} with the original order.
+ * Several of them might be a {@link Message} without {@link Serializable} serialization.
+ */
+public class TcpDiscoveryCollectionMessage implements MarshallableMessage {
+    /** {@link TcpDiscoveryAbstractMessage} pending messages which are a {@link Message}. */
+    @Order(0)
+    @Nullable Map<Integer, Message> writableMsgs;
+
+    /** Marshallable or Java-serializable pending messages which are not a {@link Message}. */
+    @Nullable private Map<Integer, TcpDiscoveryAbstractMessage> marshallableMsgs;
+
+    /** Marshalled {@link #marshallableMsgs}. */
+    @Order(1)
+    @GridToStringExclude
+    @Nullable byte[] marshallableMsgsBytes;
+
+    /** Constructor for {@link DiscoveryMessageFactory}. */
+    public TcpDiscoveryCollectionMessage() {
+        // No-op.
+    }
+
+    /** @param msgs Discovery messages to hold. */
+    public TcpDiscoveryCollectionMessage(Collection<TcpDiscoveryAbstractMessage> msgs) {
+        if (F.isEmpty(msgs))
+            return;
+
+        // Keeps the original message order.
+        int idx = 0;
+
+        for (TcpDiscoveryAbstractMessage m : msgs) {
+            if (m instanceof Message) {
+                if (writableMsgs == null)
+                    writableMsgs = U.newHashMap(msgs.size());
+
+                writableMsgs.put(idx++, (Message)m);
+
+                continue;
+            }
+
+            if (marshallableMsgs == null)
+                marshallableMsgs = U.newHashMap(msgs.size());
+
+            marshallableMsgs.put(idx++, m);
+        }
+    }
+
+    /** @param marsh marshaller. */
+    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+        if (marshallableMsgs != null && marshallableMsgsBytes == null)
+            marshallableMsgsBytes = U.marshal(marsh, marshallableMsgs);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+        if (marshallableMsgsBytes != null && marshallableMsgs == null) {
+            marshallableMsgs = U.unmarshal(marsh, marshallableMsgsBytes, clsLdr);
+
+            marshallableMsgsBytes = null;
+        }
+    }
+
+    /**
+     * Gets pending messages sent to new node by its previous.
+     *
+     * @return Pending messages from previous node.
+     */
+    public Collection<TcpDiscoveryAbstractMessage> messages() {
+        if (F.isEmpty(writableMsgs) && F.isEmpty(marshallableMsgs))
+            return Collections.emptyList();
+
+        int totalSz = (F.isEmpty(writableMsgs) ? 0 : writableMsgs.size())
+            + (F.isEmpty(marshallableMsgs) ? 0 : marshallableMsgs.size());
+
+        List<TcpDiscoveryAbstractMessage> res = new ArrayList<>(totalSz);
+
+        for (int i = 0; i < totalSz; ++i) {
+            Message m = F.isEmpty(writableMsgs) ? null : writableMsgs.get(i);
+
+            if (m == null) {
+                TcpDiscoveryAbstractMessage adm = marshallableMsgs.get(i);
+
+                assert adm != null;
+
+                res.add(adm);
+            }
+            else {
+                assert marshallableMsgs == null || marshallableMsgs.get(i) == null;
+                assert m instanceof TcpDiscoveryAbstractMessage;
+
+                res.add((TcpDiscoveryAbstractMessage)m);
+            }
+        }
+
+        return res;
+    }
+
+    /** {@inheritDoc} */
+    @Override public short directType() {
+        return -108;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(TcpDiscoveryCollectionMessage.class, this, "super", super.toString());
+    }
+}