diff --git a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java index ffabd9754894b..73a27fbe00c06 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java @@ -175,7 +175,9 @@ import org.apache.ignite.internal.processors.cache.query.GridCacheQueryResponse; import org.apache.ignite.internal.processors.cache.query.GridCacheSqlQuery; import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryBatchAck; +import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryDeployableObject; import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryEntry; +import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler; import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry; import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey; import org.apache.ignite.internal.processors.cache.transactions.TxEntryValueHolder; @@ -193,7 +195,12 @@ import org.apache.ignite.internal.processors.cluster.ClusterUpdateNotifierDataBagItem; import org.apache.ignite.internal.processors.cluster.NodeFullMetricsMessage; import org.apache.ignite.internal.processors.cluster.NodeMetricsMessage; +import org.apache.ignite.internal.processors.continuous.ContinousRoutineDiscoveryData; +import org.apache.ignite.internal.processors.continuous.ContinousRoutineDiscoveryDataItem; +import org.apache.ignite.internal.processors.continuous.ContinousRoutineLocalInfo; +import org.apache.ignite.internal.processors.continuous.ContinuousRoutineInfo; import org.apache.ignite.internal.processors.continuous.ContinuousRoutineStartResultMessage; +import org.apache.ignite.internal.processors.continuous.ContinuousRoutinesJoiningNodeDiscoveryData; import org.apache.ignite.internal.processors.continuous.GridContinuousMessage; import org.apache.ignite.internal.processors.continuous.StartRequestData; import org.apache.ignite.internal.processors.continuous.StartRoutineAckDiscoveryMessage; @@ -613,6 +620,15 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh) { withNoSchema(QueryProposalsDataBagItem.class); withNoSchema(QueryEntityMessage.class); withNoSchema(QueryEntityExMessage.class); + withNoSchema(ContinuousRoutineInfo.class); + withNoSchema(ContinuousRoutinesJoiningNodeDiscoveryData.class); + withNoSchema(CacheContinuousQueryDeployableObject.class); + withSchema(CacheContinuousQueryHandler.class); + withSchema(GridEventConsumeHandler.class); + withSchema(GridMessageListenHandler.class); + withSchema(ContinousRoutineLocalInfo.class); + withSchema(ContinousRoutineDiscoveryDataItem.class); + withNoSchema(ContinousRoutineDiscoveryData.class); // [11200 - 11300]: Compute, distributed process messages. msgIdx = 11200; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java index cad29b7b12909..031a1a94a9675 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java @@ -61,12 +61,10 @@ /** * Continuous routine handler for remote event listening. */ -class GridEventConsumeHandler implements GridContinuousHandler { - /** */ - private static final long serialVersionUID = 0L; - +@UseBinaryMarshaller +public final class GridEventConsumeHandler implements GridContinuousHandler, MarshallableMessage { /** Default callback. */ - private static final IgniteBiPredicate DFLT_CALLBACK = new P2() { + private static final IgniteBiPredicate DFLT_CALLBACK = new P2<>() { @Override public boolean apply(UUID uuid, Event e) { return true; } @@ -76,28 +74,32 @@ class GridEventConsumeHandler implements GridContinuousHandler { private IgniteBiPredicate cb; /** Filter. */ - private IgnitePredicate filter; + @Nullable volatile IgnitePredicate filter; - /** Serialized filter. */ - private byte[] filterBytes; + /** Marshaled {@link #filter}. */ + @Order(0) + @Nullable volatile byte[] filterBytes; - /** Deployment class name. */ - private String clsName; + /** Deployment class name. Is {@code null} if P2P deployment is disabled. */ + @Order(1) + @Nullable volatile String clsName; - /** Deployment info. */ - private GridDeploymentInfo depInfo; + /** Deployment info. Is {@code null} if P2P deployment is disabled. */ + @Order(2) + @Nullable volatile GridDeploymentInfoBean depInfo; /** Types. */ - private int[] types; + @Order(3) + int[] types; /** Listener. */ private GridLocalEventListener lsnr; /** P2P unmarshalling future. */ - private IgniteInternalFuture p2pUnmarshalFut = new GridFinishedFuture<>(); + private volatile IgniteInternalFuture p2pUnmarshalFut = new GridFinishedFuture<>(); /** - * Required by {@link Externalizable}. + * Empty constructor for serialization purposes. */ public GridEventConsumeHandler() { // No-op. @@ -225,8 +227,6 @@ private void initFilter(IgnitePredicate filter, GridKernalContext ctx) th EventWrapper wrapper = new EventWrapper(evt); if (evt instanceof CacheEvent) { - String cacheName = ((CacheEvent)evt).cacheName(); - ClusterNode node = ctx.discovery().node(t3.get1()); if (node == null) @@ -390,11 +390,10 @@ private boolean filterDropsEvent(Event evt) { /** {@inheritDoc} */ @Override public void p2pMarshal(GridKernalContext ctx) throws IgniteCheckedException { - assert ctx != null; assert ctx.config().isPeerClassLoadingEnabled(); if (filter != null) { - Class cls = U.detectClass(filter); + Class cls = U.detectClass(filter); clsName = cls.getName(); @@ -412,7 +411,6 @@ private boolean filterDropsEvent(Event evt) { /** {@inheritDoc} */ @Override public void p2pUnmarshal(UUID nodeId, GridKernalContext ctx) throws IgniteCheckedException { assert nodeId != null; - assert ctx != null; assert ctx.config().isPeerClassLoadingEnabled(); if (filterBytes != null) { @@ -471,36 +469,27 @@ private boolean filterDropsEvent(Event evt) { } /** {@inheritDoc} */ - @Override public void writeExternal(ObjectOutput out) throws IOException { - boolean b = filterBytes != null; + @Override public void marshal(Marshaller marsh) throws IgniteCheckedException { + assert (clsName == null) == (depInfo == null); - out.writeBoolean(b); - - if (b) { - U.writeByteArray(out, filterBytes); - U.writeString(out, clsName); - out.writeObject(depInfo); - } - else - out.writeObject(filter); - - out.writeObject(types); + /** Are marshaled in {@link #p2pUnmarshal(UUID, GridKernalContext)}. */ + if (filter != null && depInfo == null) + filterBytes = marsh.marshal(filter); } /** {@inheritDoc} */ - @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { - boolean b = in.readBoolean(); + @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException { + assert (clsName == null) == (depInfo == null); - if (b) { + /** Are unmarshaled in {@link #p2pUnmarshal(UUID, GridKernalContext)}. */ + if (depInfo != null) { p2pUnmarshalFut = new GridFutureAdapter<>(); - filterBytes = U.readByteArray(in); - clsName = U.readString(in); - depInfo = (GridDeploymentInfo)in.readObject(); + + return; } - else - filter = (IgnitePredicate)in.readObject(); - types = (int[])in.readObject(); + if (filterBytes != null) + filter = marsh.unmarshal(filterBytes, clsLdr); } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridMessageListenHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/GridMessageListenHandler.java index e59873f4b449f..2efee6e39e937 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/GridMessageListenHandler.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/GridMessageListenHandler.java @@ -17,10 +17,6 @@ package org.apache.ignite.internal; -import java.io.Externalizable; -import java.io.IOException; -import java.io.ObjectInput; -import java.io.ObjectOutput; import java.util.Collection; import java.util.Collections; import java.util.Map; @@ -39,41 +35,41 @@ import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.lang.IgniteBiPredicate; +import org.apache.ignite.marshaller.Marshaller; import org.jetbrains.annotations.Nullable; /** * Continuous handler for message subscription. */ -public class GridMessageListenHandler implements GridContinuousHandler { +@UseBinaryMarshaller +public final class GridMessageListenHandler implements GridContinuousHandler, MarshallableMessage { /** */ - private static final long serialVersionUID = 0L; + private volatile @Nullable Object topic; - /** */ - private Object topic; - - /** */ - private IgniteBiPredicate pred; - - /** */ - private byte[] topicBytes; + /** Marshalled {@link #topic}. */ + @Order(0) + @Nullable volatile byte[] topicBytes; /** */ - private byte[] predBytes; + private volatile IgniteBiPredicate pred; - /** */ - private String clsName; + /** Marshalled {@link #pred}. */ + @Order(1) + volatile byte[] predBytes; - /** */ - private GridDeploymentInfoBean depInfo; + /** Class name of {@link #pred}. Is {@code null} if the P2P deployment is disabled. */ + @Order(2) + @Nullable volatile String clsName; - /** */ - private boolean depEnabled; + /** P2P deploy info of {@link #pred}. Is {@code null} if the P2P deployment is disabled. */ + @Order(3) + @Nullable volatile GridDeploymentInfoBean predDepInfo; /** P2P unmarshalling future. */ - private IgniteInternalFuture p2pUnmarshalFut = new GridFinishedFuture<>(); + private volatile IgniteInternalFuture p2pUnmarshalFut = new GridFinishedFuture<>(); /** - * Required by {@link Externalizable}. + * Empty constructor for serialization purposes */ public GridMessageListenHandler() { // No-op. @@ -166,9 +162,19 @@ public GridMessageListenHandler(@Nullable Object topic, IgniteBiPredicate(); + + return; + } + + topic = marsh.unmarshal(topicBytes, clsLdr); + pred = marsh.unmarshal(predBytes, clsLdr); + } + /** {@inheritDoc} */ @Override public GridContinuousBatch createBatch() { return new GridContinuousBatchAdapter(); @@ -235,39 +256,6 @@ public GridMessageListenHandler(@Nullable Object topic, IgniteBiPredicate(); - topicBytes = U.readByteArray(in); - predBytes = U.readByteArray(in); - clsName = U.readString(in); - depInfo = (GridDeploymentInfoBean)in.readObject(); - } - else { - topic = in.readObject(); - pred = (IgniteBiPredicate)in.readObject(); - } - } - /** {@inheritDoc} */ @Override public String toString() { return S.toString(GridMessageListenHandler.class, this); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentInfoBean.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentInfoBean.java index ba9f48c3ca24a..b0aab5d4548da 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentInfoBean.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentInfoBean.java @@ -30,7 +30,7 @@ /** * Deployment info bean. */ -public class GridDeploymentInfoBean implements Message, GridDeploymentInfo, Serializable { +public final class GridDeploymentInfoBean implements Message, GridDeploymentInfo, Serializable { /** */ private static final long serialVersionUID = 0L; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryDeployableObject.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryDeployableObject.java index c4e4005095ff4..647e4feeae175 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryDeployableObject.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryDeployableObject.java @@ -17,40 +17,37 @@ package org.apache.ignite.internal.processors.cache.query.continuous; -import java.io.Externalizable; -import java.io.IOException; -import java.io.ObjectInput; -import java.io.ObjectOutput; import java.util.UUID; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.internal.GridKernalContext; import org.apache.ignite.internal.IgniteDeploymentCheckedException; +import org.apache.ignite.internal.Order; import org.apache.ignite.internal.managers.deployment.GridDeployment; -import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo; import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean; 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.plugin.extensions.communication.Message; /** * Deployable object. */ -class CacheContinuousQueryDeployableObject implements Externalizable { - /** */ - private static final long serialVersionUID = 0L; - +public class CacheContinuousQueryDeployableObject implements Message { /** Serialized object. */ @GridToStringExclude - private byte[] bytes; + @Order(0) + byte[] bytes; /** Deployment class name. */ - private String clsName; + @Order(1) + String clsName; /** Deployment info. */ - private GridDeploymentInfo depInfo; + @Order(2) + GridDeploymentInfoBean depInfo; /** - * Required by {@link Externalizable}. + * Empty constructor for serialization purposes. */ public CacheContinuousQueryDeployableObject() { // No-op. @@ -97,20 +94,6 @@ T unmarshal(UUID nodeId, GridKernalContext ctx) throws IgniteCheckedExceptio return U.unmarshal(ctx, bytes, U.resolveClassLoader(dep.classLoader(), ctx.config())); } - /** {@inheritDoc} */ - @Override public void writeExternal(ObjectOutput out) throws IOException { - U.writeByteArray(out, bytes); - U.writeString(out, clsName); - out.writeObject(depInfo); - } - - /** {@inheritDoc} */ - @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { - bytes = U.readByteArray(in); - clsName = U.readString(in); - depInfo = (GridDeploymentInfo)in.readObject(); - } - /** {@inheritDoc} */ @Override public String toString() { return S.toString(CacheContinuousQueryDeployableObject.class, this); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java index 805ebda1e0044..67c63ee574481 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java @@ -17,9 +17,7 @@ package org.apache.ignite.internal.processors.cache.query.continuous; -import java.io.Externalizable; import java.io.IOException; -import java.io.ObjectInput; import java.io.ObjectOutput; import java.util.ArrayList; import java.util.Collection; @@ -51,6 +49,10 @@ import org.apache.ignite.events.CacheQueryReadEvent; import org.apache.ignite.internal.GridKernalContext; import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.internal.Marshalled; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.UseBinaryMarshaller; import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; import org.apache.ignite.internal.managers.communication.GridIoPolicy; import org.apache.ignite.internal.managers.communication.MessageMarshalling; @@ -83,6 +85,7 @@ import org.apache.ignite.lang.IgniteAsyncCallback; import org.apache.ignite.lang.IgniteBiTuple; import org.apache.ignite.lang.IgniteClosure; +import org.apache.ignite.marshaller.Marshaller; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -96,10 +99,8 @@ /** * Continuous query handler. */ -public class CacheContinuousQueryHandler implements GridContinuousHandler { - /** */ - private static final long serialVersionUID = 0L; - +@UseBinaryMarshaller +public final class CacheContinuousQueryHandler implements GridContinuousHandler, MarshallableMessage { /** @see #IGNITE_CONTINUOUS_QUERY_BACKUP_ACK_THRESHOLD */ public static final int DFLT_CONTINUOUS_QUERY_BACKUP_ACK_THRESHOLD = 100; @@ -131,8 +132,8 @@ public class CacheContinuousQueryHandler implements GridContinuousHandler * Transformer implementation for processing received remote events. * They are already transformed so we simply return transformed value for event. */ - private transient IgniteClosure, ?> returnValTrans = - new IgniteClosure, Object>() { + private IgniteClosure, ?> returnValTrans = + new IgniteClosure<>() { @Override public Object apply(CacheEntryEvent evt) { assert evt.getKey() == null; @@ -141,124 +142,154 @@ public class CacheContinuousQueryHandler implements GridContinuousHandler }; /** Cache name. */ - private String cacheName; + @Order(0) + String cacheName; /** Topic for ordered messages. */ - private Object topic; + @Marshalled("topicBytes") + Object topic; + + /** Marshalled {@link #topic}. */ + @Order(1) + byte[] topicBytes; /** P2P unmarshalling future. */ - protected transient IgniteInternalFuture p2pUnmarshalFut = new GridFinishedFuture<>(); + protected volatile IgniteInternalFuture p2pUnmarshalFut = new GridFinishedFuture<>(); /** Initialization future. */ - protected transient IgniteInternalFuture initFut; + protected IgniteInternalFuture initFut; /** Local listener. */ - private transient CacheEntryUpdatedListener locLsnr; + private CacheEntryUpdatedListener locLsnr; /** Remote filter. */ - private CacheEntryEventSerializableFilter rmtFilter; + volatile CacheEntryEventSerializableFilter rmtFilter; - /** Deployable object for filter. */ - private CacheContinuousQueryDeployableObject rmtFilterDep; + /** Deployable object for {@link #rmtFilter}. Is {@code null} if no external marsshalling used. */ + @Order(2) + @Nullable volatile CacheContinuousQueryDeployableObject rmtFilterDep; + + /** Marshalled {@link #rmtFilter} if {@link #rmtFilterDep} is {@code null}. */ + @Order(3) + @Nullable volatile byte[] rmtFilterBytes; /** Remote filter factory. */ - private Factory rmtFilterFactory; + @Nullable volatile Factory rmtFilterFactory; + + /** Deployable object for {@link #rmtFilterFactory}. Is {@code null} if no external marsshalling used. */ + @Order(4) + volatile CacheContinuousQueryDeployableObject rmtFilterFactoryDep; - /** Deployable object for filter factory. */ - private CacheContinuousQueryDeployableObject rmtFilterFactoryDep; + /** Marshalled {@link #rmtFilterFactory} if {@link #rmtFilterFactoryDep} is {@code null}. */ + @Order(5) + @Nullable volatile byte[] rmtFilterFactoryBytes; /** Remote filter created by {@link #rmtFilterFactory}. */ - private transient CacheEntryEventFilter rmtFilterFromFactory; + private CacheEntryEventFilter rmtFilterFromFactory; /** Event types for JCache API. */ - private byte types; + @Order(6) + byte types; /** Remote transformer factory. */ - private Factory, ?>> rmtTransFactory; + volatile Factory, ?>> rmtTransFactory; - /** Deployable object for transformer factory. */ - private CacheContinuousQueryDeployableObject rmtTransFactoryDep; + + /** Deployable object for {@link #rmtTransFactory}. Is {@code null} if no external marsshalling used. */ + @Order(7) + volatile CacheContinuousQueryDeployableObject rmtTransFactoryDep; + + /** Marshalled {@link #rmtTransFactory} if {@link #rmtTransFactoryDep} is {@code null}. */ + @Order(8) + @Nullable volatile byte[] rmtTransFactoryBytes; /** Remote transformer created by {@link #rmtTransFactory}. */ - private transient IgniteClosure, ?> rmtTrans; + private IgniteClosure, ?> rmtTrans; /** Local listener for transformed events. */ - private transient EventListener locTransLsnr; + private EventListener locTransLsnr; /** Internal flag. */ - private boolean internal; + @Order(9) + boolean internal; /** Notify existing flag. */ - private boolean notifyExisting; + @Order(10) + boolean notifyExisting; /** Old value required flag. */ - private boolean oldValRequired; + @Order(11) + boolean oldValRequired; /** Synchronous flag. */ - private boolean sync; + @Order(12) + boolean sync; /** Ignore expired events flag. */ - private boolean ignoreExpired; + @Order(13) + boolean ignoreExpired; /** Task name hash code. */ - private int taskHash; + @Order(14) + int taskHash; /** Whether to skip primary check for REPLICATED cache. */ - private transient boolean skipPrimaryCheck; + boolean skipPrimaryCheck; /** */ - private transient boolean locOnly; + private boolean locOnly; /** */ - private boolean keepBinary; + @Order(15) + boolean keepBinary; /** */ - private transient ConcurrentMap rcvs; + private ConcurrentMap rcvs; /** */ - private transient ConcurrentMap entryBufs; + private ConcurrentMap entryBufs; /** */ - private transient CacheContinuousQueryAcknowledgeBuffer ackBuf; + private CacheContinuousQueryAcknowledgeBuffer ackBuf; /** */ - private transient int cacheId; + private int cacheId; /** */ - private transient volatile Map initUpdCntrs; + private volatile Map initUpdCntrs; /** */ - private transient volatile Map> initUpdCntrsPerNode; + private volatile Map> initUpdCntrsPerNode; /** */ - private transient volatile AffinityTopologyVersion initTopVer; + private volatile AffinityTopologyVersion initTopVer; /** */ - private transient volatile boolean nodeLeft; + private volatile boolean nodeLeft; /** */ - private transient boolean ignoreClsNotFound; + private boolean ignoreClsNotFound; /** */ - transient boolean asyncCb; + boolean asyncCb; /** */ - private transient UUID nodeId; + private UUID nodeId; /** */ - private transient UUID routineId; + private UUID routineId; /** Local update counters values on listener start. Used for skipping events fired before the listener start. */ - private transient volatile Map locInitUpdCntrs; + private volatile Map locInitUpdCntrs; /** */ - private transient GridKernalContext ctx; + private GridKernalContext ctx; /** */ - private transient IgniteLogger log; + private IgniteLogger log; /** - * Required by {@link Externalizable}. + * Empty constructor for serialization purposes. */ public CacheContinuousQueryHandler() { // No-op. @@ -942,8 +973,6 @@ private void prepareEntry(GridCacheContext cctx, UUID nodeId, CacheContinuousQue * @throws IgniteCheckedException In case of error. */ void waitTopologyFuture(GridKernalContext ctx) throws IgniteCheckedException { - GridCacheContext cctx = cacheContext(ctx); - AffinityTopologyVersion topVer = initTopVer; cacheContext(ctx).shared().exchange().affinityReadyFuture(topVer).get(); @@ -1402,23 +1431,66 @@ CacheContinuousQueryEventBuffer partitionBuffer(GridCacheContext cctx, int rmtTransFactoryDep = new CacheContinuousQueryDeployableObject(rmtTransFactory, ctx); } + /** {@inheritDoc} */ + @Override public void marshal(Marshaller marsh) throws IgniteCheckedException { + if (rmtFilter != null && rmtFilterDep == null) + rmtFilterBytes = marsh.marshal(rmtFilter); + + if (rmtFilterFactory != null && rmtFilterFactoryDep == null) + rmtFilterFactoryBytes = marsh.marshal(rmtFilterFactory); + + if (rmtTransFactory != null && rmtTransFactoryDep == null) + rmtTransFactoryBytes = marsh.marshal(rmtTransFactory); + } + + /** {@inheritDoc} */ + @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException { + if (rmtFilterBytes != null && rmtFilterDep == null) + rmtFilter = marsh.unmarshal(rmtFilterBytes, clsLdr); + + if (rmtFilterFactoryBytes != null && rmtFilterFactoryDep == null) + rmtFilterFactory = marsh.unmarshal(rmtFilterFactoryBytes, clsLdr); + + if (rmtTransFactoryBytes != null && rmtTransFactoryDep == null) + rmtTransFactory = marsh.unmarshal(rmtTransFactoryBytes, clsLdr); + + if (rmtFilterDep != null || rmtFilterFactoryDep != null || rmtTransFactoryDep != null) + p2pUnmarshalFut = new GridFutureAdapter<>(); + + cacheId = CU.cacheId(cacheName); + } + /** {@inheritDoc} */ @Override public void p2pUnmarshal(UUID nodeId, GridKernalContext ctx) throws IgniteCheckedException { assert nodeId != null; assert ctx != null; assert ctx.config().isPeerClassLoadingEnabled(); - if (rmtFilterDep != null) - rmtFilter = p2pUnmarshal(rmtFilterDep, nodeId, ctx); + try { + if (rmtFilterDep != null) + rmtFilter = rmtFilterDep.unmarshal(nodeId, ctx); + + if (rmtFilterFactoryDep != null) + rmtFilterFactory = rmtFilterFactoryDep.unmarshal(nodeId, ctx); - if (rmtFilterFactoryDep != null) - rmtFilterFactory = p2pUnmarshal(rmtFilterFactoryDep, nodeId, ctx); + if (rmtTransFactoryDep != null) + rmtTransFactory = rmtTransFactoryDep.unmarshal(nodeId, ctx); + + if (!p2pUnmarshalFut.isDone()) + ((GridFutureAdapter)p2pUnmarshalFut).onDone(); + } + catch (IgniteCheckedException e) { + ((GridFutureAdapter)p2pUnmarshalFut).onDone(e); + + throw e; + } + catch (ExceptionInInitializerError e) { + IgniteCheckedException err = new IgniteCheckedException("Failed to unmarshal deployable object.", e); - if (rmtTransFactoryDep != null) - rmtTransFactory = p2pUnmarshal(rmtTransFactoryDep, nodeId, ctx); + ((GridFutureAdapter)p2pUnmarshalFut).onDone(err); - if (!p2pUnmarshalFut.isDone()) - ((GridFutureAdapter)p2pUnmarshalFut).onDone(); + throw err; + } } /** @@ -1430,37 +1502,6 @@ public boolean isMarshalled() { && (!requiresDeployment(rmtTransFactory) || rmtTransFactoryDep != null); } - /** - * @param depObj Deployable object to unmarshal. - * @param nodeId Sender node Id. - * @param ctx Kernal context. - * @param Result type. - * @return Unmarshalled object. - * @throws IgniteCheckedException In case of unmarshalling failures. - */ - protected T p2pUnmarshal(CacheContinuousQueryDeployableObject depObj, - UUID nodeId, GridKernalContext ctx) throws IgniteCheckedException { - if (depObj != null) { - try { - return depObj.unmarshal(nodeId, ctx); - } - catch (IgniteCheckedException e) { - ((GridFutureAdapter)p2pUnmarshalFut).onDone(e); - - throw e; - } - catch (ExceptionInInitializerError e) { - IgniteCheckedException err = new IgniteCheckedException("Failed to unmarshal deployable object.", e); - - ((GridFutureAdapter)p2pUnmarshalFut).onDone(err); - - throw err; - } - } - else - return null; - } - /** {@inheritDoc} */ @Override public GridContinuousBatch createBatch() { return new GridContinuousQueryBatch(); @@ -1554,78 +1595,6 @@ private void sendBackupAcknowledge(final IgniteBiTuple, Set(); - } - else - rmtFilter = (CacheEntryEventSerializableFilter)in.readObject(); - - internal = in.readBoolean(); - notifyExisting = in.readBoolean(); - oldValRequired = in.readBoolean(); - sync = in.readBoolean(); - ignoreExpired = in.readBoolean(); - taskHash = in.readInt(); - keepBinary = in.readBoolean(); - - b = in.readBoolean(); - - if (b) { - rmtFilterFactoryDep = (CacheContinuousQueryDeployableObject)in.readObject(); - - if (p2pUnmarshalFut.isDone()) - p2pUnmarshalFut = new GridFutureAdapter<>(); - } - else - rmtFilterFactory = (Factory)in.readObject(); - - types = in.readByte(); - - b = in.readBoolean(); - - if (b) { - rmtTransFactoryDep = (CacheContinuousQueryDeployableObject)in.readObject(); - - if (p2pUnmarshalFut.isDone()) - p2pUnmarshalFut = new GridFutureAdapter<>(); - } - else - rmtTransFactory = (Factory, ?>>)in.readObject(); - - cacheId = CU.cacheId(cacheName); - } - /** */ private static void writeDeployable(ObjectOutput out, Object obj, CacheContinuousQueryDeployableObject dep) throws IOException { boolean b = dep != null; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinousRoutineDiscoveryData.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinousRoutineDiscoveryData.java new file mode 100644 index 0000000000000..d6d1340649f4b --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinousRoutineDiscoveryData.java @@ -0,0 +1,72 @@ +/* + * 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.internal.processors.continuous; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Map; +import java.util.UUID; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.util.tostring.GridToStringInclude; +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.plugin.extensions.communication.Message; + +/** Discovery data. */ +public class ContinousRoutineDiscoveryData implements Message { + /** Node ID. */ + @Order(0) + UUID nodeId; + + /** Items. */ + @GridToStringInclude + @Order(1) + Collection items; + + /** */ + @Order(2) + Map> clientInfos; + + /** Empty constructor for serialization purposes. */ + public ContinousRoutineDiscoveryData() { + // No-op. + } + + /** + * @param nodeId Node ID. + * @param clientInfos Client information. + */ + ContinousRoutineDiscoveryData(UUID nodeId, Map> clientInfos) { + assert nodeId != null; + + this.nodeId = nodeId; + + this.clientInfos = clientInfos; + + items = new ArrayList<>(); + } + + /** @param item Item. */ + public void addItem(ContinousRoutineDiscoveryDataItem item) { + items.add(item); + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ContinousRoutineDiscoveryData.class, this); + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinousRoutineDiscoveryDataItem.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinousRoutineDiscoveryDataItem.java new file mode 100644 index 0000000000000..247c5be996cb2 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinousRoutineDiscoveryDataItem.java @@ -0,0 +1,98 @@ +/* + * 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.internal.processors.continuous; + +import java.util.UUID; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.internal.Marshalled; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.UseBinaryMarshaller; +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.lang.IgnitePredicate; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.jetbrains.annotations.Nullable; + +/** Discovery data item. */ +@UseBinaryMarshaller +public class ContinousRoutineDiscoveryDataItem implements Message { + /** */ + @Order(0) + UUID routineId; + + /** */ + @Marshalled("prjPredBytes") + IgnitePredicate prjPred; + + /** Marshalled {@link #prjPred}. */ + @Order(1) + byte[] prjPredBytes; + + /** Handler. */ + @Order(2) + GridContinuousHandler hnd; + + /** Buffer size. */ + @Order(3) + int bufSize; + + /** Time interval. */ + @Order(4) + long interval; + + /** Automatic unsubscribe flag. */ + @Order(5) + boolean autoUnsubscribe; + + /** Empty constructor for serialization porposes. */ + public ContinousRoutineDiscoveryDataItem() { + // No-op. + } + + /** + * @param routineId Consume ID. + * @param prjPred Projection predicate. + * @param hnd Handler. + * @param bufSize Buffer size. + * @param interval Time interval. + * @param autoUnsubscribe Automatic unsubscribe flag. + */ + ContinousRoutineDiscoveryDataItem(UUID routineId, + @Nullable IgnitePredicate prjPred, + GridContinuousHandler hnd, + int bufSize, + long interval, + boolean autoUnsubscribe + ) { + assert routineId != null; + assert hnd != null; + assert bufSize > 0; + assert interval >= 0; + + this.routineId = routineId; + this.prjPred = prjPred; + this.hnd = hnd; + this.bufSize = bufSize; + this.interval = interval; + this.autoUnsubscribe = autoUnsubscribe; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ContinousRoutineDiscoveryDataItem.class, this); + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinousRoutineLocalInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinousRoutineLocalInfo.java new file mode 100644 index 0000000000000..900600c54954b --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinousRoutineLocalInfo.java @@ -0,0 +1,133 @@ +/* + * 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.internal.processors.continuous; + +import java.util.UUID; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.internal.Marshalled; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.UseBinaryMarshaller; +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.lang.IgnitePredicate; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.jetbrains.annotations.Nullable; + +/** Local routine info. */ +@UseBinaryMarshaller +public class ContinousRoutineLocalInfo implements Message, GridContinuousProcessor.RoutineInfo { + /** Source node id. */ + @Order(0) + UUID nodeId; + + /** Projection predicate. */ + @Marshalled("prjPredBytes") + IgnitePredicate prjPred; + + /** Marshalled {@link #prjPred}. */ + @Order(1) + byte[] prjPredBytes; + + /** Continuous routine handler. */ + @Order(2) + GridContinuousHandler hnd; + + /** Buffer size. */ + @Order(3) + int bufSize; + + /** Time interval. */ + @Order(4) + long interval; + + /** Automatic unsubscribe flag. */ + @Order(5) + boolean autoUnsubscribe; + + /** Empty constructor for serialization purposes. */ + public ContinousRoutineLocalInfo() { + // No-op. + } + + /** + * @param nodeId Node id. + * @param prjPred Projection predicate. + * @param hnd Continuous routine handler. + * @param bufSize Buffer size. + * @param interval Interval. + * @param autoUnsubscribe Automatic unsubscribe flag. + */ + ContinousRoutineLocalInfo( + UUID nodeId, + @Nullable IgnitePredicate prjPred, + GridContinuousHandler hnd, + int bufSize, + long interval, + boolean autoUnsubscribe + ) { + assert hnd != null; + assert bufSize > 0; + assert interval >= 0; + + this.nodeId = nodeId; + this.prjPred = prjPred; + this.hnd = hnd; + this.bufSize = bufSize; + this.interval = interval; + this.autoUnsubscribe = autoUnsubscribe; + } + + /** {@inheritDoc} */ + @Override public GridContinuousHandler handler() { + return hnd; + } + + /** {@inheritDoc} */ + @Override public int bufferSize() { + return bufSize; + } + + /** {@inheritDoc} */ + @Override public long interval() { + return interval; + } + + /** {@inheritDoc} */ + @Override public boolean autoUnsubscribe() { + return autoUnsubscribe; + } + + /** {@inheritDoc} */ + @Override public long lastSendTime() { + return -1; + } + + /** {@inheritDoc} */ + @Override public boolean delayedRegister() { + return false; + } + + /** {@inheritDoc} */ + @Override public UUID nodeId() { + return nodeId; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ContinousRoutineLocalInfo.class, this); + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutineInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutineInfo.java index 207a8f4fda984..86f7b0db9803b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutineInfo.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutineInfo.java @@ -17,45 +17,55 @@ package org.apache.ignite.internal.processors.continuous; -import java.io.Serializable; import java.util.UUID; +import org.apache.ignite.internal.Order; import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.plugin.extensions.communication.Message; /** * */ -class ContinuousRoutineInfo implements Serializable { - /** */ - private static final long serialVersionUID = 0L; - +public class ContinuousRoutineInfo implements Message { /** */ + @Order(0) UUID srcNodeId; /** */ - final UUID routineId; + @Order(1) + UUID routineId; /** */ - final byte[] hnd; + @Order(2) + GridContinuousHandler hnd; /** */ - final byte[] nodeFilter; + @Order(3) + byte[] nodeFilter; /** */ - final int bufSize; + @Order(4) + int bufSize; /** */ - final long interval; + @Order(5) + long interval; /** */ - final boolean autoUnsubscribe; + @Order(6) + boolean autoUnsubscribe; - /** */ - transient boolean disconnected; + /** Transient. */ + boolean disconnected; + + /** Empty constructor for serialization purposes. */ + public ContinuousRoutineInfo() { + // No-op. + } /** * @param srcNodeId Source node ID. * @param routineId Routine ID. - * @param hnd Marshalled handler. + * @param hnd Handler. * @param nodeFilter Marshalled node filter. * @param bufSize Handler buffer size. * @param interval Time interval. @@ -64,7 +74,7 @@ class ContinuousRoutineInfo implements Serializable { ContinuousRoutineInfo( UUID srcNodeId, UUID routineId, - byte[] hnd, + GridContinuousHandler hnd, byte[] nodeFilter, int bufSize, long interval, diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesJoiningNodeDiscoveryData.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesJoiningNodeDiscoveryData.java index 9be6ef8e07eab..1849e472672de 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesJoiningNodeDiscoveryData.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/ContinuousRoutinesJoiningNodeDiscoveryData.java @@ -17,19 +17,23 @@ package org.apache.ignite.internal.processors.continuous; -import java.io.Serializable; import java.util.List; +import org.apache.ignite.internal.Order; import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.plugin.extensions.communication.Message; /** * */ -public class ContinuousRoutinesJoiningNodeDiscoveryData implements Serializable { +public class ContinuousRoutinesJoiningNodeDiscoveryData implements Message { /** */ - private static final long serialVersionUID = 0L; + @Order(0) + List startedRoutines; - /** */ - final List startedRoutines; + /** Empty constructor for serialization purposes. */ + public ContinuousRoutinesJoiningNodeDiscoveryData() { + // No-op. + } /** * @param startedRoutines Routines registered on nodes, to be started in cluster. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousHandler.java index b1a3812f61217..03ca0defa6d37 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousHandler.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousHandler.java @@ -17,20 +17,20 @@ package org.apache.ignite.internal.processors.continuous; -import java.io.Externalizable; import java.util.Collection; import java.util.Map; import java.util.UUID; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.internal.GridKernalContext; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; +import org.apache.ignite.plugin.extensions.communication.Message; import org.jetbrains.annotations.Nullable; /** * Continuous routine handler. */ @SuppressWarnings("PublicInnerClass") -public interface GridContinuousHandler extends Externalizable, Cloneable { +public interface GridContinuousHandler extends Cloneable, Message { /** * Listener registration status. */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java index fecd065ff7551..e2d3350a7a201 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java @@ -17,12 +17,6 @@ package org.apache.ignite.internal.processors.continuous; -import java.io.Externalizable; -import java.io.IOException; -import java.io.ObjectInput; -import java.io.ObjectOutput; -import java.io.Serializable; -import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -80,7 +74,6 @@ import org.apache.ignite.internal.util.future.GridFutureAdapter; import org.apache.ignite.internal.util.lang.GridPlainRunnable; import org.apache.ignite.internal.util.lang.gridfunc.ReadOnlyCollectionView2X; -import org.apache.ignite.internal.util.tostring.GridToStringInclude; import org.apache.ignite.internal.util.typedef.CI1; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.X; @@ -125,10 +118,10 @@ public class GridContinuousProcessor extends GridProcessorAdapter { public static final String CQ_SYS_VIEW_DESC = "Continuous queries"; /** Local infos. */ - private final ConcurrentMap locInfos = new ConcurrentHashMap<>(); + private final ConcurrentMap locInfos = new ConcurrentHashMap<>(); /** Local infos. */ - private final ConcurrentMap> clientInfos = new ConcurrentHashMap<>(); + private final ConcurrentMap> clientInfos = new ConcurrentHashMap<>(); /** Remote infos. */ private final ConcurrentMap rmtInfos = new ConcurrentHashMap<>(); @@ -336,12 +329,12 @@ Map remoteRoutineInfos() { } /** */ - Map localRoutineInfos() { + Map localRoutineInfos() { return Collections.unmodifiableMap(locInfos); } /** */ - Map> clientRoutineInfos() { + Map> clientRoutineInfos() { return Collections.unmodifiableMap(clientInfos); } @@ -408,7 +401,7 @@ public void unlockStopping() { return; } - Serializable data = getDiscoveryData(dataBag.joiningNodeId()); + ContinousRoutineDiscoveryData data = getDiscoveryData(dataBag.joiningNodeId()); if (data != null) dataBag.addJoiningNodeData(CONTINUOUS_PROC.ordinal(), data); @@ -422,7 +415,7 @@ public void unlockStopping() { return; } - Serializable data = getDiscoveryData(dataBag.joiningNodeId()); + ContinousRoutineDiscoveryData data = getDiscoveryData(dataBag.joiningNodeId()); if (data != null) dataBag.addNodeSpecificData(CONTINUOUS_PROC.ordinal(), data); @@ -431,7 +424,7 @@ public void unlockStopping() { /** * @param joiningNodeId Joining node id. */ - private Serializable getDiscoveryData(UUID joiningNodeId) { + private @Nullable ContinousRoutineDiscoveryData getDiscoveryData(UUID joiningNodeId) { if (log.isDebugEnabled()) { log.debug("collectDiscoveryData [node=" + joiningNodeId + ", loc=" + ctx.localNodeId() + @@ -441,26 +434,26 @@ private Serializable getDiscoveryData(UUID joiningNodeId) { } if (!joiningNodeId.equals(ctx.localNodeId()) || !locInfos.isEmpty()) { - Map> clientInfos0 = copyClientInfos(clientInfos); + Map> clientInfos0 = copyClientInfos(clientInfos); if (joiningNodeId.equals(ctx.localNodeId()) && ctx.discovery().localNode().isClient()) { - Map infos = copyLocalInfos(locInfos); + Map infos = copyLocalInfos(locInfos); clientInfos0.put(ctx.localNodeId(), infos); } - DiscoveryData data = new DiscoveryData(ctx.localNodeId(), clientInfos0); + ContinousRoutineDiscoveryData data = new ContinousRoutineDiscoveryData(ctx.localNodeId(), clientInfos0); // Collect listeners information (will be sent to joining node during discovery process). - for (Map.Entry e : locInfos.entrySet()) { + for (Map.Entry e : locInfos.entrySet()) { UUID routineId = e.getKey(); - LocalRoutineInfo info = e.getValue(); + ContinousRoutineLocalInfo info = e.getValue(); assert !ctx.config().isPeerClassLoadingEnabled() || !(info.hnd instanceof CacheContinuousQueryHandler) || - ((CacheContinuousQueryHandler)info.hnd).isMarshalled(); + ((CacheContinuousQueryHandler)info.hnd).isMarshalled(); - data.addItem(new DiscoveryDataItem(routineId, + data.addItem(new ContinousRoutineDiscoveryDataItem(routineId, info.prjPred, info.hnd, info.bufSize, @@ -477,13 +470,13 @@ private Serializable getDiscoveryData(UUID joiningNodeId) { /** * @param clientInfos Client infos. */ - private Map> copyClientInfos(Map> clientInfos) { - Map> res = U.newHashMap(clientInfos.size()); + private Map> copyClientInfos(Map> clientInfos) { + Map> res = U.newHashMap(clientInfos.size()); - for (Map.Entry> e : clientInfos.entrySet()) { - Map cp = U.newHashMap(e.getValue().size()); + for (Map.Entry> e : clientInfos.entrySet()) { + Map cp = U.newHashMap(e.getValue().size()); - for (Map.Entry e0 : e.getValue().entrySet()) + for (Map.Entry e0 : e.getValue().entrySet()) cp.put(e0.getKey(), e0.getValue()); res.put(e.getKey(), cp); @@ -495,10 +488,10 @@ private Map> copyClientInfos(Map copyLocalInfos(Map locInfos) { - Map res = U.newHashMap(locInfos.size()); + private Map copyLocalInfos(Map locInfos) { + Map res = U.newHashMap(locInfos.size()); - for (Map.Entry e : locInfos.entrySet()) + for (Map.Entry e : locInfos.entrySet()) res.put(e.getKey(), e.getValue()); return res; @@ -547,10 +540,10 @@ private Map copyLocalInfos(Map l } } else { - Map nodeSpecData = data.nodeSpecificData(); + Map nodeSpecData = data.nodeSpecificData(); if (nodeSpecData != null) { - for (DiscoveryData val : nodeSpecData.values()) + for (ContinousRoutineDiscoveryData val : nodeSpecData.values()) onDiscoveryDataReceivedMutable(val); } } @@ -562,37 +555,37 @@ private Map copyLocalInfos(Map l * * @param data received discovery data. */ - private void onDiscoveryDataReceivedMutable(DiscoveryData data) { + private void onDiscoveryDataReceivedMutable(ContinousRoutineDiscoveryData data) { if (data != null) { - for (DiscoveryDataItem item : data.items) { + for (ContinousRoutineDiscoveryDataItem item : data.items) { if (!locInfos.containsKey(item.routineId)) { registerHandlerOnJoin(data.nodeId, item.routineId, item.prjPred, item.hnd, item.bufSize, item.interval, item.autoUnsubscribe); } if (!item.autoUnsubscribe) { - locInfos.putIfAbsent(item.routineId, new LocalRoutineInfo(data.nodeId, + locInfos.putIfAbsent(item.routineId, new ContinousRoutineLocalInfo(data.nodeId, item.prjPred, item.hnd, item.bufSize, item.interval, item.autoUnsubscribe)); } } // Process CQs started on clients. - for (Map.Entry> entry : data.clientInfos.entrySet()) { + for (Map.Entry> entry : data.clientInfos.entrySet()) { UUID clientNodeId = entry.getKey(); if (!ctx.localNodeId().equals(clientNodeId)) { - Map clientRoutineMap = entry.getValue(); + Map clientRoutineMap = entry.getValue(); - for (Map.Entry e : clientRoutineMap.entrySet()) { + for (Map.Entry e : clientRoutineMap.entrySet()) { UUID routineId = e.getKey(); - LocalRoutineInfo info = e.getValue(); + ContinousRoutineLocalInfo info = e.getValue(); registerHandlerOnJoin(clientNodeId, routineId, info.prjPred, info.hnd, info.bufSize, info.interval, info.autoUnsubscribe); } } - Map map = + Map map = clientInfos.computeIfAbsent(clientNodeId, k -> new HashMap<>()); map.putAll(entry.getValue()); @@ -628,23 +621,8 @@ private void onDiscoveryDataReceivedImmutable(ContinuousRoutineInfo routineInfo) return; } - GridContinuousHandler hnd; - - try { - hnd = U.unmarshal(marsh, routineInfo.hnd, U.resolveClassLoader(ctx.config())); - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to unmarshal continuous routine handler [" + - "routineId=" + routineInfo.routineId + - ", srcNodeId=" + routineInfo.srcNodeId + ']', e); - - ctx.failure().process(new FailureContext(FailureType.CRITICAL_ERROR, e)); - - return; - } - registerHandlerOnJoin(routineInfo.srcNodeId, routineInfo.routineId, nodeFilter, - hnd, routineInfo.bufSize, routineInfo.interval, routineInfo.autoUnsubscribe); + routineInfo.hnd, routineInfo.bufSize, routineInfo.interval, routineInfo.autoUnsubscribe); } /** @@ -782,7 +760,7 @@ public UUID registerStaticRoutine( final UUID routineId = UUID.randomUUID(); - LocalRoutineInfo routineInfo = new LocalRoutineInfo(ctx.localNodeId(), prjPred, hnd, 1, 0, true); + ContinousRoutineLocalInfo routineInfo = new ContinousRoutineLocalInfo(ctx.localNodeId(), prjPred, hnd, 1, 0, true); if (immutableDiscoCustomMsg) { routinesInfo.addRoutineInfo(createRoutineInfo( @@ -822,14 +800,13 @@ private ContinuousRoutineInfo createRoutineInfo( long interval, boolean autoUnsubscribe) throws IgniteCheckedException { - byte[] hndBytes = marsh.marshal(hnd); byte[] filterBytes = nodeFilter != null ? marsh.marshal(nodeFilter) : null; return new ContinuousRoutineInfo( srcNodeId, routineId, - hndBytes, + hnd, filterBytes, bufSize, interval, @@ -866,7 +843,7 @@ public IgniteInternalFuture startRoutine(GridContinuousHandler hnd, // Register routine locally. locInfos.put(routineId, - new LocalRoutineInfo(ctx.localNodeId(), prjPred, hnd, bufSize, interval, autoUnsubscribe)); + new ContinousRoutineLocalInfo(ctx.localNodeId(), prjPred, hnd, bufSize, interval, autoUnsubscribe)); if (locOnly) { try { @@ -1058,7 +1035,7 @@ public IgniteInternalFuture stopRoutine(UUID routineId) { boolean stop = false; // Unregister routine locally. - LocalRoutineInfo routine = locInfos.remove(routineId); + ContinousRoutineLocalInfo routine = locInfos.remove(routineId); if (routine != null) { stop = true; @@ -1123,7 +1100,7 @@ public void addBackupNotification(UUID nodeId, sendNotification(nodeId, routineId, null, toSnd, orderedTopic, true, null); } else { - LocalRoutineInfo locRoutineInfo = locInfos.get(routineId); + ContinousRoutineLocalInfo locRoutineInfo = locInfos.get(routineId); if (locRoutineInfo != null) locRoutineInfo.handler().notifyCallback(nodeId, routineId, objs, ctx); @@ -1248,7 +1225,7 @@ public void addNotification(UUID nodeId, unregisterRemote(e.getKey()); } - for (LocalRoutineInfo routine : locInfos.values()) + for (ContinousRoutineLocalInfo routine : locInfos.values()) routine.hnd.onClientDisconnected(); rmtInfos.clear(); @@ -1315,7 +1292,7 @@ private void processStopRequest(ClusterNode snd, StopRoutineDiscoveryMessage msg unregisterRemote(routineId); } - for (Map clientInfo : clientInfos.values()) { + for (Map clientInfo : clientInfos.values()) { if (clientInfo.remove(msg.routineId()) != null) break; } @@ -1374,17 +1351,17 @@ private void processStartRequestMutable(ClusterNode node, StartRoutineDiscoveryM GridContinuousHandler hnd = data.handler(); if (node.isClient()) { - Map clientRoutineMap = clientInfos.get(node.id()); + Map clientRoutineMap = clientInfos.get(node.id()); if (clientRoutineMap == null) { clientRoutineMap = new HashMap<>(); - Map old = clientInfos.put(node.id(), clientRoutineMap); + Map old = clientInfos.put(node.id(), clientRoutineMap); assert old == null; } - clientRoutineMap.put(routineId, new LocalRoutineInfo(node.id(), + clientRoutineMap.put(routineId, new ContinousRoutineLocalInfo(node.id(), data.nodeFilter(), hnd, data.bufferSize(), @@ -1422,7 +1399,7 @@ private void processStartRequestMutable(ClusterNode node, StartRoutineDiscoveryM if (!data.autoUnsubscribe()) // Register routine locally. - locInfos.putIfAbsent(routineId, new LocalRoutineInfo( + locInfos.putIfAbsent(routineId, new ContinousRoutineLocalInfo( node.id(), prjPred, hnd, data.bufferSize(), data.interval(), data.autoUnsubscribe())); } catch (IgniteCheckedException e) { @@ -1468,7 +1445,7 @@ private void processStartRequestImmutable(final AffinityTopologyVersion topVer, ContinuousRoutineInfo routineInfo = new ContinuousRoutineInfo(snd.id(), msg.routineId(), - reqData.hndBytes, + reqData.hnd, reqData.nodeFilterBytes, reqData.bufferSize(), reqData.interval(), @@ -1607,7 +1584,7 @@ private void processNotification(UUID nodeId, GridContinuousMessage msg) { UUID routineId = msg.routineId(); try { - LocalRoutineInfo routine = locInfos.get(routineId); + ContinousRoutineLocalInfo routine = locInfos.get(routineId); if (routine != null) routine.hnd.notifyCallback(nodeId, routineId, (Collection)msg.data(), ctx); @@ -1769,7 +1746,7 @@ private void unregisterHandler(UUID routineId, GridContinuousHandler hnd, boolea @SuppressWarnings("TooBroadScope") private void unregisterRemote(UUID routineId) { RemoteRoutineInfo remote; - LocalRoutineInfo loc; + ContinousRoutineLocalInfo loc; stopLock.lock(); @@ -1971,100 +1948,6 @@ public static interface RoutineInfo { boolean delayedRegister(); } - /** - * Local routine info. - */ - public static class LocalRoutineInfo implements Serializable, RoutineInfo { - /** */ - private static final long serialVersionUID = 0L; - - /** Source node id. */ - private final UUID nodeId; - - /** Projection predicate. */ - private final IgnitePredicate prjPred; - - /** Continuous routine handler. */ - private final GridContinuousHandler hnd; - - /** Buffer size. */ - private final int bufSize; - - /** Time interval. */ - private final long interval; - - /** Automatic unsubscribe flag. */ - private boolean autoUnsubscribe; - - /** - * @param nodeId Node id. - * @param prjPred Projection predicate. - * @param hnd Continuous routine handler. - * @param bufSize Buffer size. - * @param interval Interval. - * @param autoUnsubscribe Automatic unsubscribe flag. - */ - LocalRoutineInfo( - UUID nodeId, - @Nullable IgnitePredicate prjPred, - GridContinuousHandler hnd, - int bufSize, - long interval, - boolean autoUnsubscribe - ) { - assert hnd != null; - assert bufSize > 0; - assert interval >= 0; - - this.nodeId = nodeId; - this.prjPred = prjPred; - this.hnd = hnd; - this.bufSize = bufSize; - this.interval = interval; - this.autoUnsubscribe = autoUnsubscribe; - } - - /** {@inheritDoc} */ - @Override public GridContinuousHandler handler() { - return hnd; - } - - /** {@inheritDoc} */ - @Override public int bufferSize() { - return bufSize; - } - - /** {@inheritDoc} */ - @Override public long interval() { - return interval; - } - - /** {@inheritDoc} */ - @Override public boolean autoUnsubscribe() { - return autoUnsubscribe; - } - - /** {@inheritDoc} */ - @Override public long lastSendTime() { - return -1; - } - - /** {@inheritDoc} */ - @Override public boolean delayedRegister() { - return false; - } - - /** {@inheritDoc} */ - @Override public UUID nodeId() { - return nodeId; - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(LocalRoutineInfo.class, this); - } - } - /** * Remote routine info. */ @@ -2286,157 +2169,6 @@ IgniteBiTuple checkInterval() { } } - /** - * Discovery data. - */ - private static class DiscoveryData implements Externalizable { - /** */ - private static final long serialVersionUID = 0L; - - /** Node ID. */ - private UUID nodeId; - - /** Items. */ - @GridToStringInclude - private Collection items; - - /** */ - private Map> clientInfos; - - /** - * Required by {@link Externalizable}. - */ - public DiscoveryData() { - // No-op. - } - - /** - * @param nodeId Node ID. - * @param clientInfos Client information. - */ - DiscoveryData(UUID nodeId, Map> clientInfos) { - assert nodeId != null; - - this.nodeId = nodeId; - - this.clientInfos = clientInfos; - - items = new ArrayList<>(); - } - - /** - * @param item Item. - */ - public void addItem(DiscoveryDataItem item) { - items.add(item); - } - - /** {@inheritDoc} */ - @Override public void writeExternal(ObjectOutput out) throws IOException { - U.writeUuid(out, nodeId); - U.writeCollection(out, items); - U.writeMap(out, clientInfos); - } - - /** {@inheritDoc} */ - @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { - nodeId = U.readUuid(in); - items = U.readCollection(in); - clientInfos = U.readMap(in); - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(DiscoveryData.class, this); - } - } - - /** - * Discovery data item. - */ - private static class DiscoveryDataItem implements Externalizable { - /** */ - private static final long serialVersionUID = 0L; - - /** Consume ID. */ - private UUID routineId; - - /** Projection predicate. */ - private IgnitePredicate prjPred; - - /** Handler. */ - private GridContinuousHandler hnd; - - /** Buffer size. */ - private int bufSize; - - /** Time interval. */ - private long interval; - - /** Automatic unsubscribe flag. */ - private boolean autoUnsubscribe; - - /** - * Required by {@link Externalizable}. - */ - public DiscoveryDataItem() { - // No-op. - } - - /** - * @param routineId Consume ID. - * @param prjPred Projection predicate. - * @param hnd Handler. - * @param bufSize Buffer size. - * @param interval Time interval. - * @param autoUnsubscribe Automatic unsubscribe flag. - */ - DiscoveryDataItem(UUID routineId, - @Nullable IgnitePredicate prjPred, - GridContinuousHandler hnd, - int bufSize, - long interval, - boolean autoUnsubscribe - ) { - assert routineId != null; - assert hnd != null; - assert bufSize > 0; - assert interval >= 0; - - this.routineId = routineId; - this.prjPred = prjPred; - this.hnd = hnd; - this.bufSize = bufSize; - this.interval = interval; - this.autoUnsubscribe = autoUnsubscribe; - } - - /** {@inheritDoc} */ - @Override public void writeExternal(ObjectOutput out) throws IOException { - U.writeUuid(out, routineId); - out.writeObject(prjPred); - out.writeObject(hnd); - out.writeInt(bufSize); - out.writeLong(interval); - out.writeBoolean(autoUnsubscribe); - } - - /** {@inheritDoc} */ - @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { - routineId = U.readUuid(in); - prjPred = (IgnitePredicate)in.readObject(); - hnd = (GridContinuousHandler)in.readObject(); - bufSize = in.readInt(); - interval = in.readLong(); - autoUnsubscribe = in.readBoolean(); - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(DiscoveryDataItem.class, this); - } - } - /** * Future for start routine. */ @@ -2521,7 +2253,7 @@ private void onAllRemoteRegistered( Map cntrs) { try { if (errs == null || errs.isEmpty()) { - LocalRoutineInfo routine = locInfos.get(routineId); + ContinousRoutineLocalInfo routine = locInfos.get(routineId); // Update partition counters. if (routine != null && routine.handler().isQuery()) { diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRequestData.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRequestData.java index 9571c231bbf9a..d400c35641d55 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRequestData.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRequestData.java @@ -51,11 +51,8 @@ public class StartRequestData implements Message { GridDeploymentInfoBean depInfo; /** Handler. */ - private GridContinuousHandler hnd; - - /** Serialized handler. */ @Order(3) - byte[] hndBytes; + GridContinuousHandler hnd; /** Buffer size. */ @Order(4) @@ -177,8 +174,6 @@ public void marshal(GridKernalContext ctx) throws IgniteCheckedException { // Handle peer deployment for other handler-specific objects. hnd.p2pMarshal(ctx); } - - hndBytes = U.marshal(ctx.marshaller(), hnd); } if (nodeFilter != null) @@ -210,17 +205,13 @@ public void unmarshal(GridKernalContext ctx, UUID sndId) throws IgniteCheckedExc U.resolveClassLoader(ctx.config())); } - if (hndBytes != null) { - hnd = U.unmarshal(ctx.marshaller(), hndBytes, U.resolveClassLoader(ctx.config())); + if (ctx.config().isPeerClassLoadingEnabled()) + hnd.p2pUnmarshal(sndId, ctx); - if (ctx.config().isPeerClassLoadingEnabled()) - hnd.p2pUnmarshal(sndId, ctx); + if (keepBinary) { + assert hnd instanceof CacheContinuousQueryHandler : hnd; - if (keepBinary) { - assert hnd instanceof CacheContinuousQueryHandler : hnd; - - ((CacheContinuousQueryHandler)hnd).keepBinary(true); - } + ((CacheContinuousQueryHandler)hnd).keepBinary(true); } } } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntriesExpireTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntriesExpireTest.java index 736b1742befa5..317076bc92a47 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntriesExpireTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntriesExpireTest.java @@ -28,7 +28,7 @@ import org.apache.ignite.configuration.CacheConfiguration; import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.processors.cache.GridCacheContext; -import org.apache.ignite.internal.processors.continuous.GridContinuousProcessor; +import org.apache.ignite.internal.processors.continuous.ContinousRoutineLocalInfo; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.internal.CU; import org.apache.ignite.testframework.GridTestUtils; @@ -57,7 +57,7 @@ public void testBackupQOnEntriesExpire() throws Exception { for (int i = 0; i < 1_000; i++) cache.put(i, i); - ConcurrentMap locInfos = + ConcurrentMap locInfos = getFieldValue(srv1.context().continuous(), "locInfos"); assertEquals(1, locInfos.size()); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java index 47b107edeee24..96e6e1d50cb12 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java @@ -63,7 +63,6 @@ import static org.apache.ignite.events.EventType.EVT_JOB_STARTED; import static org.apache.ignite.events.EventType.EVT_NODE_FAILED; import static org.apache.ignite.events.EventType.EVT_NODE_LEFT; -import static org.apache.ignite.internal.processors.continuous.GridContinuousProcessor.LocalRoutineInfo; import static org.apache.ignite.testframework.GridTestUtils.noop; /** @@ -157,13 +156,15 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest { * @param proc Continuous processor. * @return Local event routines. */ - private Collection localRoutines(GridContinuousProcessor proc) { - return F.view(U.>field(proc, "locInfos").values(), - new IgnitePredicate() { - @Override public boolean apply(LocalRoutineInfo info) { + private Collection localRoutines(GridContinuousProcessor proc) { + return F.view( + U.>field(proc, "locInfos").values(), + new IgnitePredicate<>() { + @Override public boolean apply(ContinousRoutineLocalInfo info) { return info.handler().isEvents(); } - }); + } + ); } /**