IGNITE-28221 : combining of the message factories v2 (#13003)

Author: Vladsz83

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/message/CalciteCommunicationMessageSerializationTest.java

@@ -17,7 +17,7 @@
 
 package org.apache.ignite.internal.processors.query.calcite.message;
 
-import org.apache.ignite.internal.managers.communication.AbstractMessageSerializationTest;
+import org.apache.ignite.internal.managers.AbstractMessageSerializationTest;
 import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
 
 /** */

modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java

@@ -19,7 +19,6 @@
 
 import java.util.List;
 import java.util.Objects;
-import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
 import org.apache.ignite.internal.util.typedef.internal.U;
@@ -31,7 +30,7 @@ public final class ExchangeInfo extends IgniteDiagnosticRequest.DiagnosticBaseIn
     AffinityTopologyVersion topVer;
 
     /**
-     * Empty constructor required by {@link GridIoMessageFactory}.
+     * Empty constructor required by {@link CoreMessagesProvider}.
      */
     public ExchangeInfo() {
         // No-op.

modules/core/src/main/java/org/apache/ignite/internal/ExchangeInfo.java

@@ -84,6 +84,7 @@
 import org.apache.ignite.maintenance.MaintenanceRegistry;
 import org.apache.ignite.plugin.PluginNotFoundException;
 import org.apache.ignite.plugin.PluginProvider;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 
 /**
  *
@@ -209,6 +210,13 @@ public interface GridKernalContext extends Iterable<GridComponent> {
      */
     public MaintenanceRegistry maintenanceRegistry();
 
+    /**
+     * Gets core message factoy.
+     *
+     * @return Core message factory.
+     */
+    public MessageFactory messageFactory();
+
     /**
      * Gets transformation processor.
      *

modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java

@@ -114,6 +114,7 @@
 import org.apache.ignite.maintenance.MaintenanceRegistry;
 import org.apache.ignite.plugin.PluginNotFoundException;
 import org.apache.ignite.plugin.PluginProvider;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.jetbrains.annotations.Nullable;
 
 import static org.apache.ignite.internal.IgniteComponentType.SPRING;
@@ -371,17 +372,14 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
     private Thread.UncaughtExceptionHandler hnd;
 
     /** */
-    private IgniteEx grid;
+    private IgniteKernal grid;
 
     /** */
     private IgniteConfiguration cfg;
 
     /** */
     private GridKernalGateway gw;
 
-    /** Network segmented flag. */
-    private volatile boolean segFlag;
-
     /** Performance suggestions. */
     private final GridPerformanceSuggestions perf = new GridPerformanceSuggestions();
 
@@ -430,7 +428,7 @@ public GridKernalContextImpl() {
     @SuppressWarnings("TypeMayBeWeakened")
     protected GridKernalContextImpl(
         GridLoggerProxy log,
-        IgniteEx grid,
+        IgniteKernal grid,
         IgniteConfiguration cfg,
         GridKernalGateway gw,
         List<PluginProvider> plugins,
@@ -614,18 +612,9 @@ else if (!(comp instanceof DiscoveryNodeValidationProcessor
             comps.add(comp);
     }
 
-    /**
-     * @param helper Helper to add.
-     */
-    public void addHelper(Object helper) {
-        assert helper != null;
-
-        assert false : "Unknown helper class: " + helper.getClass();
-    }
-
     /** {@inheritDoc} */
     @Override public boolean isStopping() {
-        return ((IgniteKernal)grid).isStopping();
+        return grid.isStopping();
     }
 
     /** */
@@ -703,6 +692,11 @@ public void addHelper(Object helper) {
         return maintenanceProc;
     }
 
+    /** {@inheritDoc} */
+    @Override public MessageFactory messageFactory() {
+        return grid.messageFactory();
+    }
+
     /** {@inheritDoc} */
     @Override public CacheObjectTransformerProcessor transformer() {
         return transProc;

modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java

@@ -24,7 +24,6 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
-import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.plugin.extensions.communication.Message;
 import org.jetbrains.annotations.Nullable;
@@ -49,7 +48,7 @@ public class IgniteDiagnosticRequest implements Message {
     private final Map<Object, List<String>> msgs = new LinkedHashMap<>();
 
     /**
-     * Default constructor required by {@link GridIoMessageFactory}.
+     * Default constructor required by {@link CoreMessagesProvider}.
      */
     public IgniteDiagnosticRequest() {
         // No-op.

modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticRequest.java

@@ -17,7 +17,6 @@
 
 package org.apache.ignite.internal;
 
-import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.plugin.extensions.communication.Message;
 import org.jetbrains.annotations.Nullable;
@@ -33,7 +32,7 @@ public class IgniteDiagnosticResponse implements Message {
     @Nullable String respInfo;
 
     /**
-     * Default constructor required by {@link GridIoMessageFactory}.
+     * Default constructor required by {@link CoreMessagesProvider}.
      */
     public IgniteDiagnosticResponse() {
         // No-op.

modules/core/src/main/java/org/apache/ignite/internal/IgniteDiagnosticResponse.java

@@ -99,6 +99,7 @@
 import org.apache.ignite.internal.managers.checkpoint.GridCheckpointManager;
 import org.apache.ignite.internal.managers.collision.GridCollisionManager;
 import org.apache.ignite.internal.managers.communication.GridIoManager;
+import org.apache.ignite.internal.managers.communication.IgniteMessageFactoryImpl;
 import org.apache.ignite.internal.managers.deployment.GridDeploymentManager;
 import org.apache.ignite.internal.managers.discovery.DiscoveryLocalJoinData;
 import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
@@ -209,6 +210,8 @@
 import org.apache.ignite.plugin.IgnitePlugin;
 import org.apache.ignite.plugin.PluginNotFoundException;
 import org.apache.ignite.plugin.PluginProvider;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
 import org.apache.ignite.spi.IgniteSpi;
 import org.apache.ignite.spi.IgniteSpiVersionCheckException;
 import org.apache.ignite.spi.discovery.isolated.IsolatedDiscoverySpi;
@@ -436,6 +439,9 @@ public class IgniteKernal implements IgniteEx, Externalizable {
     /** The state object is used when reconnection occurs. See {@link IgniteKernal#onReconnected(boolean)}. */
     private final ReconnectState reconnectState = new ReconnectState();
 
+    /** Core message factory. */
+    private MessageFactory msgFactory;
+
     /**
      * No-arg constructor is required by externalization.
      */
@@ -999,7 +1005,11 @@ public void start(
             }
             startManager(new GridMetricManager(ctx));
             startManager(new GridSystemViewManager(ctx));
+
+            initMessageFactory();
+
             startManager(new GridIoManager(ctx));
+
             startManager(new GridCheckpointManager(ctx));
 
             startManager(new GridEventStorageManager(ctx));
@@ -1301,6 +1311,31 @@ else if (e instanceof IgniteCheckedException)
         startTimer.finishGlobalStage("Await exchange");
     }
 
+    /** */
+    private void initMessageFactory() throws IgniteCheckedException {
+        MessageFactoryProvider[] msgs = ctx.plugins().extensions(MessageFactoryProvider.class);
+
+        if (msgs == null)
+            msgs = new MessageFactoryProvider[0];
+
+        List<MessageFactoryProvider> compMsgs = new ArrayList<>();
+
+        compMsgs.add(new CoreMessagesProvider(ctx.marshaller(), ctx.marshallerContext().jdkMarshaller(),
+            U.resolveClassLoader(ctx.config())));
+
+        for (IgniteComponentType compType : IgniteComponentType.values()) {
+            MessageFactoryProvider f = compType.messageFactory();
+
+            if (f != null)
+                compMsgs.add(f);
+        }
+
+        if (!compMsgs.isEmpty())
+            msgs = F.concat(msgs, compMsgs.toArray(new MessageFactoryProvider[compMsgs.size()]));
+
+        msgFactory = new IgniteMessageFactoryImpl(msgs);
+    }
+
     /**
      * @return Ignite security processor. See {@link IgniteSecurity} for details.
      */
@@ -3028,6 +3063,11 @@ private void checkClusterState() throws IgniteException {
         }
     }
 
+    /** @return Core message factory. */
+    MessageFactory messageFactory() {
+        return msgFactory;
+    }
+
     /**
      * Method is responsible for handling the {@link EventType#EVT_CLIENT_NODE_DISCONNECTED} event. Notify all the
      * GridComponents that the such even has been occurred (e.g. if the local client node disconnected from the cluster

modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java

@@ -21,7 +21,6 @@
 import java.util.HashSet;
 import java.util.Objects;
 import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
 import org.apache.ignite.internal.processors.cache.GridCacheContext;
 import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
 import org.apache.ignite.internal.processors.cache.KeyCacheObject;
@@ -38,7 +37,7 @@ public final class TxEntriesInfo extends IgniteDiagnosticRequest.DiagnosticBaseI
     Collection<KeyCacheObject> keys;
 
     /**
-     * Empty constructor required by {@link GridIoMessageFactory}.
+     * Empty constructor required by {@link CoreMessagesProvider}.
      */
     public TxEntriesInfo() {
         // No-op.

modules/core/src/main/java/org/apache/ignite/internal/TxEntriesInfo.java

@@ -18,7 +18,6 @@
 package org.apache.ignite.internal;
 
 import java.util.Objects;
-import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.util.typedef.internal.U;
@@ -34,7 +33,7 @@ public final class TxInfo extends IgniteDiagnosticRequest.DiagnosticBaseInfo {
     GridCacheVersion nearVer;
 
     /**
-     * Empty constructor required by {@link GridIoMessageFactory}.
+     * Empty constructor required by {@link CoreMessagesProvider}.
      */
     public TxInfo() {
         // No-op.