You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ni...@apache.org on 2018/02/08 12:01:20 UTC
[2/2] ignite git commit: IGNITE-425: Implementation of
ContinuousQueryWithTransformer
IGNITE-425: Implementation of ContinuousQueryWithTransformer
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/a83f3038
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/a83f3038
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/a83f3038
Branch: refs/heads/master
Commit: a83f3038a9424a5eb826cf35da8d22496a551b50
Parents: c917327
Author: Nikolay Izhikov <ni...@apache.org>
Authored: Thu Feb 8 14:51:32 2018 +0300
Committer: Nikolay Izhikov <ni...@apache.org>
Committed: Thu Feb 8 14:51:32 2018 +0300
----------------------------------------------------------------------
.../cache/query/AbstractContinuousQuery.java | 202 ++++++++
.../ignite/cache/query/ContinuousQuery.java | 172 +------
.../query/ContinuousQueryWithTransformer.java | 192 +++++++
.../processors/cache/IgniteCacheProxyImpl.java | 82 ++-
.../continuous/CacheContinuousQueryEntry.java | 9 +-
.../continuous/CacheContinuousQueryHandler.java | 149 +++++-
.../CacheContinuousQueryHandlerV2.java | 9 +-
.../CacheContinuousQueryHandlerV3.java | 185 +++++++
.../continuous/CacheContinuousQueryManager.java | 44 +-
...acheContinuousQueryRandomOperationsTest.java | 156 ++++--
...ContinuousWithTransformerClientSelfTest.java | 40 ++
...heContinuousWithTransformerFailoverTest.java | 309 +++++++++++
...eContinuousWithTransformerLocalSelfTest.java | 29 ++
...nuousWithTransformerPartitionedSelfTest.java | 29 ++
...uousWithTransformerRandomOperationsTest.java | 31 ++
...inuousWithTransformerReplicatedSelfTest.java | 511 +++++++++++++++++++
.../IgniteCacheQuerySelfTestSuite3.java | 13 +
17 files changed, 1919 insertions(+), 243 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/a83f3038/modules/core/src/main/java/org/apache/ignite/cache/query/AbstractContinuousQuery.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/query/AbstractContinuousQuery.java b/modules/core/src/main/java/org/apache/ignite/cache/query/AbstractContinuousQuery.java
new file mode 100644
index 0000000..2a615ee
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/cache/query/AbstractContinuousQuery.java
@@ -0,0 +1,202 @@
+/*
+ * 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.cache.query;
+
+import javax.cache.Cache;
+import javax.cache.configuration.Factory;
+import javax.cache.event.CacheEntryEventFilter;
+import javax.cache.event.EventType;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.lang.IgniteAsyncCallback;
+
+/**
+ * Base class for continuous query.
+ *
+ * @see ContinuousQuery
+ * @see ContinuousQueryWithTransformer
+ */
+public abstract class AbstractContinuousQuery<K, V> extends Query<Cache.Entry<K, V>> {
+ /**
+ * Default page size. Size of {@code 1} means that all entries
+ * will be sent to master node immediately (buffering is disabled).
+ */
+ public static final int DFLT_PAGE_SIZE = 1;
+
+ /** Maximum default time interval after which buffer will be flushed (if buffering is enabled). */
+ public static final long DFLT_TIME_INTERVAL = 0;
+
+ /**
+ * Default value for automatic unsubscription flag. Remote filters
+ * will be unregistered by default if master node leaves topology.
+ */
+ public static final boolean DFLT_AUTO_UNSUBSCRIBE = true;
+
+ /** Initial query. */
+ private Query<Cache.Entry<K, V>> initQry;
+
+ /** Remote filter factory. */
+ private Factory<? extends CacheEntryEventFilter<K, V>> rmtFilterFactory;
+
+ /** Time interval. */
+ private long timeInterval = DFLT_TIME_INTERVAL;
+
+ /** Automatic unsubscription flag. */
+ private boolean autoUnsubscribe = DFLT_AUTO_UNSUBSCRIBE;
+
+ /** Whether to notify about {@link EventType#EXPIRED} events. */
+ private boolean includeExpired;
+
+ /**
+ * Sets initial query.
+ * <p>
+ * This query will be executed before continuous listener is registered
+ * which allows to iterate through entries which already existed at the
+ * time continuous query is executed.
+ *
+ * @param initQry Initial query.
+ * @return {@code this} for chaining.
+ */
+ public AbstractContinuousQuery<K, V> setInitialQuery(Query<Cache.Entry<K, V>> initQry) {
+ this.initQry = initQry;
+
+ return this;
+ }
+
+ /**
+ * Gets initial query.
+ *
+ * @return Initial query.
+ */
+ public Query<Cache.Entry<K, V>> getInitialQuery() {
+ return initQry;
+ }
+
+ /**
+ * Sets optional key-value filter factory. This factory produces filter is called before entry is
+ * sent to the master node.
+ * <p>
+ * <b>WARNING:</b> all operations that involve any kind of JVM-local or distributed locking
+ * (e.g., synchronization or transactional cache operations), should be executed asynchronously
+ * without blocking the thread that called the filter. Otherwise, you can get deadlocks.
+ * <p>
+ * If remote filter are annotated with {@link IgniteAsyncCallback} then it is executed in async callback
+ * pool (see {@link IgniteConfiguration#getAsyncCallbackPoolSize()}) that allow to perform a cache operations.
+ *
+ * @param rmtFilterFactory Key-value filter factory.
+ * @return {@code this} for chaining.
+ * @see IgniteAsyncCallback
+ * @see IgniteConfiguration#getAsyncCallbackPoolSize()
+ */
+ public AbstractContinuousQuery<K, V> setRemoteFilterFactory(
+ Factory<? extends CacheEntryEventFilter<K, V>> rmtFilterFactory) {
+ this.rmtFilterFactory = rmtFilterFactory;
+
+ return this;
+ }
+
+ /**
+ * Gets remote filter.
+ *
+ * @return Remote filter.
+ */
+ public Factory<? extends CacheEntryEventFilter<K, V>> getRemoteFilterFactory() {
+ return rmtFilterFactory;
+ }
+
+ /**
+ * Sets time interval.
+ * <p>
+ * When a cache update happens, entry is first put into a buffer. Entries from buffer will
+ * be sent to the master node only if the buffer is full (its size can be provided via {@link #setPageSize(int)}
+ * method) or time provided via this method is exceeded.
+ * <p>
+ * Default time interval is {@code 0} which means that
+ * time check is disabled and entries will be sent only when buffer is full.
+ *
+ * @param timeInterval Time interval.
+ * @return {@code this} for chaining.
+ */
+ public AbstractContinuousQuery<K, V> setTimeInterval(long timeInterval) {
+ if (timeInterval < 0)
+ throw new IllegalArgumentException("Time interval can't be negative.");
+
+ this.timeInterval = timeInterval;
+
+ return this;
+ }
+
+ /**
+ * Gets time interval.
+ *
+ * @return Time interval.
+ */
+ public long getTimeInterval() {
+ return timeInterval;
+ }
+
+ /**
+ * Sets automatic unsubscribe flag.
+ * <p>
+ * This flag indicates that query filters on remote nodes should be
+ * automatically unregistered if master node (node that initiated the query) leaves topology. If this flag is
+ * {@code false}, filters will be unregistered only when the query is cancelled from master node, and won't ever be
+ * unregistered if master node leaves grid.
+ * <p>
+ * Default value for this flag is {@code true}.
+ *
+ * @param autoUnsubscribe Automatic unsubscription flag.
+ * @return {@code this} for chaining.
+ */
+ public AbstractContinuousQuery<K, V> setAutoUnsubscribe(boolean autoUnsubscribe) {
+ this.autoUnsubscribe = autoUnsubscribe;
+
+ return this;
+ }
+
+ /**
+ * Gets automatic unsubscription flag value.
+ *
+ * @return Automatic unsubscription flag.
+ */
+ public boolean isAutoUnsubscribe() {
+ return autoUnsubscribe;
+ }
+
+ /**
+ * Sets the flag value defining whether to notify about {@link EventType#EXPIRED} events.
+ * If {@code true}, then the remote listener will get notifications about entries
+ * expired in cache. Otherwise, only {@link EventType#CREATED}, {@link EventType#UPDATED}
+ * and {@link EventType#REMOVED} events will be fired in the remote listener.
+ * <p>
+ * This flag is {@code false} by default, so {@link EventType#EXPIRED} events are disabled.
+ *
+ * @param includeExpired Whether to notify about {@link EventType#EXPIRED} events.
+ */
+ public void setIncludeExpired(boolean includeExpired) {
+ this.includeExpired = includeExpired;
+ }
+
+ /**
+ * Gets the flag value defining whether to notify about {@link EventType#EXPIRED} events.
+ *
+ * @return Whether to notify about {@link EventType#EXPIRED} events.
+ */
+ public boolean isIncludeExpired() {
+ return includeExpired;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a83f3038/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQuery.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQuery.java b/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQuery.java
index 49d471e..9a8fbca 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQuery.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQuery.java
@@ -21,9 +21,9 @@ import javax.cache.Cache;
import javax.cache.configuration.Factory;
import javax.cache.event.CacheEntryEventFilter;
import javax.cache.event.CacheEntryUpdatedListener;
-import javax.cache.event.EventType;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
+import org.apache.ignite.cache.query.ContinuousQueryWithTransformer.EventListener;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.lang.IgniteAsyncCallback;
@@ -103,49 +103,20 @@ import org.apache.ignite.lang.IgniteAsyncCallback;
* is executed in async callback pool (see {@link IgniteConfiguration#getAsyncCallbackPoolSize()})
* and notification order is kept the same as update order for given cache key.
*
+ * @see ContinuousQueryWithTransformer
* @see IgniteAsyncCallback
* @see IgniteConfiguration#getAsyncCallbackPoolSize()
*/
-public final class ContinuousQuery<K, V> extends Query<Cache.Entry<K, V>> {
+public final class ContinuousQuery<K, V> extends AbstractContinuousQuery<K, V> {
/** */
private static final long serialVersionUID = 0L;
- /**
- * Default page size. Size of {@code 1} means that all entries
- * will be sent to master node immediately (buffering is disabled).
- */
- public static final int DFLT_PAGE_SIZE = 1;
-
- /** Maximum default time interval after which buffer will be flushed (if buffering is enabled). */
- public static final long DFLT_TIME_INTERVAL = 0;
-
- /**
- * Default value for automatic unsubscription flag. Remote filters
- * will be unregistered by default if master node leaves topology.
- */
- public static final boolean DFLT_AUTO_UNSUBSCRIBE = true;
-
- /** Initial query. */
- private Query<Cache.Entry<K, V>> initQry;
-
/** Local listener. */
private CacheEntryUpdatedListener<K, V> locLsnr;
/** Remote filter. */
private CacheEntryEventSerializableFilter<K, V> rmtFilter;
- /** Remote filter factory. */
- private Factory<? extends CacheEntryEventFilter<K, V>> rmtFilterFactory;
-
- /** Time interval. */
- private long timeInterval = DFLT_TIME_INTERVAL;
-
- /** Automatic unsubscription flag. */
- private boolean autoUnsubscribe = DFLT_AUTO_UNSUBSCRIBE;
-
- /** Whether to notify about {@link EventType#EXPIRED} events. */
- private boolean includeExpired;
-
/**
* Creates new continuous query.
*/
@@ -153,29 +124,9 @@ public final class ContinuousQuery<K, V> extends Query<Cache.Entry<K, V>> {
setPageSize(DFLT_PAGE_SIZE);
}
- /**
- * Sets initial query.
- * <p>
- * This query will be executed before continuous listener is registered
- * which allows to iterate through entries which already existed at the
- * time continuous query is executed.
- *
- * @param initQry Initial query.
- * @return {@code this} for chaining.
- */
+ /** {@inheritDoc} */
public ContinuousQuery<K, V> setInitialQuery(Query<Cache.Entry<K, V>> initQry) {
- this.initQry = initQry;
-
- return this;
- }
-
- /**
- * Gets initial query.
- *
- * @return Initial query.
- */
- public Query<Cache.Entry<K, V>> getInitialQuery() {
- return initQry;
+ return (ContinuousQuery<K, V>)super.setInitialQuery(initQry);
}
/**
@@ -197,6 +148,7 @@ public final class ContinuousQuery<K, V> extends Query<Cache.Entry<K, V>> {
* @return {@code this} for chaining.
* @see IgniteAsyncCallback
* @see IgniteConfiguration#getAsyncCallbackPoolSize()
+ * @see ContinuousQueryWithTransformer#setLocalListener(EventListener)
*/
public ContinuousQuery<K, V> setLocalListener(CacheEntryUpdatedListener<K, V> locLsnr) {
this.locLsnr = locLsnr;
@@ -246,118 +198,14 @@ public final class ContinuousQuery<K, V> extends Query<Cache.Entry<K, V>> {
return rmtFilter;
}
- /**
- * Sets optional key-value filter factory. This factory produces filter is called before entry is
- * sent to the master node.
- * <p>
- * <b>WARNING:</b> all operations that involve any kind of JVM-local or distributed locking
- * (e.g., synchronization or transactional cache operations), should be executed asynchronously
- * without blocking the thread that called the filter. Otherwise, you can get deadlocks.
- * <p>
- * If remote filter are annotated with {@link IgniteAsyncCallback} then it is executed in async callback
- * pool (see {@link IgniteConfiguration#getAsyncCallbackPoolSize()}) that allow to perform a cache operations.
- *
- * @param rmtFilterFactory Key-value filter factory.
- * @return {@code this} for chaining.
- * @see IgniteAsyncCallback
- * @see IgniteConfiguration#getAsyncCallbackPoolSize()
- */
- public ContinuousQuery<K, V> setRemoteFilterFactory(
- Factory<? extends CacheEntryEventFilter<K, V>> rmtFilterFactory) {
- this.rmtFilterFactory = rmtFilterFactory;
-
- return this;
- }
-
- /**
- * Gets remote filter.
- *
- * @return Remote filter.
- */
- public Factory<? extends CacheEntryEventFilter<K, V>> getRemoteFilterFactory() {
- return rmtFilterFactory;
- }
-
- /**
- * Sets time interval.
- * <p>
- * When a cache update happens, entry is first put into a buffer. Entries from buffer will
- * be sent to the master node only if the buffer is full (its size can be provided via {@link #setPageSize(int)}
- * method) or time provided via this method is exceeded.
- * <p>
- * Default time interval is {@code 0} which means that
- * time check is disabled and entries will be sent only when buffer is full.
- *
- * @param timeInterval Time interval.
- * @return {@code this} for chaining.
- */
+ /** {@inheritDoc} */
public ContinuousQuery<K, V> setTimeInterval(long timeInterval) {
- if (timeInterval < 0)
- throw new IllegalArgumentException("Time interval can't be negative.");
-
- this.timeInterval = timeInterval;
-
- return this;
+ return (ContinuousQuery<K, V>)super.setTimeInterval(timeInterval);
}
- /**
- * Gets time interval.
- *
- * @return Time interval.
- */
- public long getTimeInterval() {
- return timeInterval;
- }
-
- /**
- * Sets automatic unsubscribe flag.
- * <p>
- * This flag indicates that query filters on remote nodes should be
- * automatically unregistered if master node (node that initiated the query) leaves topology. If this flag is
- * {@code false}, filters will be unregistered only when the query is cancelled from master node, and won't ever be
- * unregistered if master node leaves grid.
- * <p>
- * Default value for this flag is {@code true}.
- *
- * @param autoUnsubscribe Automatic unsubscription flag.
- * @return {@code this} for chaining.
- */
+ /** {@inheritDoc} */
public ContinuousQuery<K, V> setAutoUnsubscribe(boolean autoUnsubscribe) {
- this.autoUnsubscribe = autoUnsubscribe;
-
- return this;
- }
-
- /**
- * Gets automatic unsubscription flag value.
- *
- * @return Automatic unsubscription flag.
- */
- public boolean isAutoUnsubscribe() {
- return autoUnsubscribe;
- }
-
- /**
- * Sets the flag value defining whether to notify about {@link EventType#EXPIRED} events.
- * If {@code true}, then the remote listener will get notifications about entries
- * expired in cache. Otherwise, only {@link EventType#CREATED}, {@link EventType#UPDATED}
- * and {@link EventType#REMOVED} events will be fired in the remote listener.
- * <p>
- * This flag is {@code false} by default, so {@link EventType#EXPIRED} events are disabled.
- *
- * @param includeExpired Whether to notify about {@link EventType#EXPIRED} events.
- */
- public void setIncludeExpired(boolean includeExpired) {
- this.includeExpired = includeExpired;
- }
-
- /**
- * Gets the flag value defining whether to notify about {@link EventType#EXPIRED} events.
- *
- * @return Whether to notify about {@link EventType#EXPIRED} events.
- */
- public boolean isIncludeExpired() {
- return includeExpired;
+ return (ContinuousQuery<K, V>)super.setAutoUnsubscribe(autoUnsubscribe);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/a83f3038/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQueryWithTransformer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQueryWithTransformer.java b/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQueryWithTransformer.java
new file mode 100644
index 0000000..122410f
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQueryWithTransformer.java
@@ -0,0 +1,192 @@
+/*
+ * 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.cache.query;
+
+import javax.cache.Cache;
+import javax.cache.configuration.Factory;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryEventFilter;
+import javax.cache.event.CacheEntryUpdatedListener;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.lang.IgniteAsyncCallback;
+import org.apache.ignite.lang.IgniteClosure;
+
+/**
+ * API for configuring continuous cache queries with transformer.
+ * <p>
+ * Continuous queries allow to register a remote filter and a local listener
+ * for cache updates. If an update event passes the filter, it will be transformed with transformer and sent to
+ * the node that executed the query and local listener will be notified.
+ * <p>
+ * Additionally, you can execute initial query to get currently existing data.
+ * Query can be of any type (SQL, TEXT or SCAN) and can be set via {@link #setInitialQuery(Query)}
+ * method.
+ * <p>
+ * Query can be executed either on all nodes in topology using {@link IgniteCache#query(Query)}
+ * method, or only on the local node, if {@link Query#setLocal(boolean)} parameter is set to {@code true}.
+ * Note that in case query is distributed and a new node joins, it will get the remote
+ * filter for the query during discovery process before it actually joins topology,
+ * so no updates will be missed.
+ * This will execute query on all nodes that have cache you are working with and
+ * listener will start to receive notifications for cache updates.
+ * <p>
+ * To stop receiving updates call {@link QueryCursor#close()} method.
+ * Note that this works even if you didn't provide initial query. Cursor will
+ * be empty in this case, but it will still unregister listeners when {@link QueryCursor#close()}
+ * is called.
+ * <p>
+ * {@link IgniteAsyncCallback} annotation is supported for {@link CacheEntryEventFilter}
+ * (see {@link #setRemoteFilterFactory(Factory)}) and {@link CacheEntryUpdatedListener}
+ * (see {@link #setRemoteTransformerFactory(Factory)}) and {@link CacheEntryUpdatedListener}
+ * (see {@link #setLocalListener(EventListener)} and {@link EventListener}).
+ * If filter and/or listener are annotated with {@link IgniteAsyncCallback} then annotated callback
+ * is executed in async callback pool (see {@link IgniteConfiguration#getAsyncCallbackPoolSize()})
+ * and notification order is kept the same as update order for given cache key.
+ *
+ * @see ContinuousQuery
+ * @see IgniteAsyncCallback
+ * @see IgniteConfiguration#getAsyncCallbackPoolSize()
+ */
+public final class ContinuousQueryWithTransformer<K, V, T> extends AbstractContinuousQuery<K, V> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Remote transformer factory. */
+ private Factory<? extends IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, T>> rmtTransFactory;
+
+ /** Local listener of transformed event */
+ private EventListener<T> locLsnr;
+
+ /**
+ * Creates new continuous query with transformer.
+ */
+ public ContinuousQueryWithTransformer() {
+ setPageSize(DFLT_PAGE_SIZE);
+ }
+
+ /** {@inheritDoc} */
+ public ContinuousQueryWithTransformer<K, V, T> setInitialQuery(Query<Cache.Entry<K, V>> initQry) {
+ return (ContinuousQueryWithTransformer<K, V, T>)super.setInitialQuery(initQry);
+ }
+
+ /** {@inheritDoc} */
+ public ContinuousQueryWithTransformer<K, V, T> setRemoteFilterFactory(
+ Factory<? extends CacheEntryEventFilter<K, V>> rmtFilterFactory) {
+ return (ContinuousQueryWithTransformer<K, V, T>)super.setRemoteFilterFactory(rmtFilterFactory);
+ }
+
+ /**
+ * Sets transformer factory. This factory produces transformer is called after and only if entry passes the filter.
+ * <p>
+ * <b>WARNING:</b> all operations that involve any kind of JVM-local or distributed locking
+ * (e.g., synchronization or transactional cache operations), should be executed asynchronously
+ * without blocking the thread that called the filter. Otherwise, you can get deadlocks.
+ * <p>
+ *
+ * @param factory Remote transformer factory.
+ * @return {@code this} for chaining.
+ */
+ public ContinuousQueryWithTransformer<K, V, T> setRemoteTransformerFactory(
+ Factory<? extends IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, T>> factory) {
+ this.rmtTransFactory = factory;
+
+ return this;
+ }
+
+ /**
+ * Gets remote transformer factory
+ *
+ * @return Remote Transformer Factory
+ */
+ public Factory<? extends IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, T>> getRemoteTransformerFactory() {
+ return rmtTransFactory;
+ }
+
+ /**
+ * Sets local callback. This callback is called only in local node when new updates are received.
+ * <p>
+ * The callback predicate accepts results of transformed by {@link #getRemoteFilterFactory()} events
+ * <p>
+ * <b>WARNING:</b> all operations that involve any kind of JVM-local or distributed locking (e.g.,
+ * synchronization or transactional cache operations), should be executed asynchronously without
+ * blocking the thread that called the callback. Otherwise, you can get deadlocks.
+ * <p>
+ * If local listener are annotated with {@link IgniteAsyncCallback} then it is executed in async callback pool
+ * (see {@link IgniteConfiguration#getAsyncCallbackPoolSize()}) that allow to perform a cache operations.
+ *
+ * @param locLsnr Local callback.
+ * @return {@code this} for chaining.
+ *
+ * @see IgniteAsyncCallback
+ * @see IgniteConfiguration#getAsyncCallbackPoolSize()
+ * @see ContinuousQuery#setLocalListener(CacheEntryUpdatedListener)
+ */
+ public ContinuousQueryWithTransformer<K, V, T> setLocalListener(EventListener<T> locLsnr) {
+ this.locLsnr = locLsnr;
+
+ return this;
+ }
+
+ /**
+ * Gets local transformed event listener
+ *
+ * @return local transformed event listener
+ */
+ public EventListener<T> getLocalListener() {
+ return locLsnr;
+ }
+
+ /** {@inheritDoc} */
+ public ContinuousQueryWithTransformer<K, V, T> setTimeInterval(long timeInterval) {
+ return (ContinuousQueryWithTransformer<K, V, T>)super.setTimeInterval(timeInterval);
+ }
+
+ /** {@inheritDoc} */
+ public ContinuousQueryWithTransformer<K, V, T> setAutoUnsubscribe(boolean autoUnsubscribe) {
+ return (ContinuousQueryWithTransformer<K, V, T>)super.setAutoUnsubscribe(autoUnsubscribe);
+ }
+
+ /** {@inheritDoc} */
+ @Override public ContinuousQueryWithTransformer<K, V, T> setPageSize(int pageSize) {
+ return (ContinuousQueryWithTransformer<K, V, T>)super.setPageSize(pageSize);
+ }
+
+ /** {@inheritDoc} */
+ @Override public ContinuousQueryWithTransformer<K, V, T> setLocal(boolean loc) {
+ return (ContinuousQueryWithTransformer<K, V, T>)super.setLocal(loc);
+ }
+
+ /**
+ * Interface for local listener of {@link ContinuousQueryWithTransformer} to implement.
+ * Invoked if an cache entry is updated, created or if a batch call is made,
+ * after the entries are updated and transformed.
+ *
+ * @param <T> type of data produced by transformer {@link ContinuousQueryWithTransformer#getRemoteTransformerFactory()}.
+ * @see ContinuousQueryWithTransformer
+ * @see ContinuousQueryWithTransformer#setLocalListener(EventListener)
+ */
+ public interface EventListener<T> {
+ /**
+ * Called after one or more entries have been updated.
+ *
+ * @param events The entries just updated that transformed with remote transformer of {@link ContinuousQueryWithTransformer}.
+ */
+ void onUpdated(Iterable<? extends T> events);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a83f3038/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyImpl.java
index 7f71c74..a834022 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyImpl.java
@@ -36,6 +36,8 @@ import javax.cache.Cache;
import javax.cache.CacheException;
import javax.cache.configuration.CacheEntryListenerConfiguration;
import javax.cache.configuration.Configuration;
+import javax.cache.configuration.Factory;
+import javax.cache.event.CacheEntryUpdatedListener;
import javax.cache.expiry.ExpiryPolicy;
import javax.cache.integration.CompletionListener;
import javax.cache.processor.EntryProcessor;
@@ -46,11 +48,15 @@ import org.apache.ignite.IgniteCacheRestartingException;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.cache.CacheEntry;
+import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
import org.apache.ignite.cache.CacheEntryProcessor;
import org.apache.ignite.cache.CacheManager;
import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.cache.query.AbstractContinuousQuery;
import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cache.query.ContinuousQueryWithTransformer;
+import org.apache.ignite.cache.query.ContinuousQueryWithTransformer.EventListener;
import org.apache.ignite.cache.query.FieldsQueryCursor;
import org.apache.ignite.cache.query.Query;
import org.apache.ignite.cache.query.QueryCursor;
@@ -62,6 +68,7 @@ import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.cache.query.SqlQuery;
import org.apache.ignite.cache.query.TextQuery;
import org.apache.ignite.cluster.ClusterGroup;
+import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.AsyncSupportAdapter;
import org.apache.ignite.internal.IgniteEx;
@@ -92,6 +99,7 @@ import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.lang.IgniteProductVersion;
import org.apache.ignite.mxbean.CacheMetricsMXBean;
import org.apache.ignite.plugin.security.SecurityPermission;
import org.jetbrains.annotations.NotNull;
@@ -106,6 +114,12 @@ public class IgniteCacheProxyImpl<K, V> extends AsyncSupportAdapter<IgniteCache<
/** */
private static final long serialVersionUID = 0L;
+ /**
+ * Ignite version that introduce {@link ContinuousQueryWithTransformer} feature.
+ */
+ private static final IgniteProductVersion CONT_QRY_WITH_TRANSFORMER_SINCE =
+ IgniteProductVersion.fromString("2.5.0");
+
/** Context. */
private volatile GridCacheContext<K, V> ctx;
@@ -498,22 +512,66 @@ public class IgniteCacheProxyImpl<K, V> extends AsyncSupportAdapter<IgniteCache<
* @return Initial iteration cursor.
*/
@SuppressWarnings("unchecked")
- private QueryCursor<Cache.Entry<K, V>> queryContinuous(ContinuousQuery qry, boolean loc, boolean keepBinary) {
- if (qry.getInitialQuery() instanceof ContinuousQuery)
+ private QueryCursor<Cache.Entry<K, V>> queryContinuous(AbstractContinuousQuery qry, boolean loc, boolean keepBinary) {
+ assert qry instanceof ContinuousQuery || qry instanceof ContinuousQueryWithTransformer;
+
+ if (qry.getInitialQuery() instanceof ContinuousQuery ||
+ qry.getInitialQuery() instanceof ContinuousQueryWithTransformer) {
throw new IgniteException("Initial predicate for continuous query can't be an instance of another " +
"continuous query. Use SCAN or SQL query for initial iteration.");
+ }
+
+ CacheEntryUpdatedListener locLsnr = null;
+
+ EventListener locTransLsnr = null;
+
+ CacheEntryEventSerializableFilter rmtFilter = null;
+
+ Factory<? extends IgniteClosure> rmtTransFactory = null;
+
+ if (qry instanceof ContinuousQuery) {
+ ContinuousQuery<K, V> qry0 = (ContinuousQuery<K, V>)qry;
+
+ if (qry0.getLocalListener() == null)
+ throw new IgniteException("Mandatory local listener is not set for the query: " + qry);
+
+ if (qry0.getRemoteFilter() != null && qry0.getRemoteFilterFactory() != null)
+ throw new IgniteException("Should be used either RemoterFilter or RemoteFilterFactory.");
- if (qry.getLocalListener() == null)
- throw new IgniteException("Mandatory local listener is not set for the query: " + qry);
+ locLsnr = qry0.getLocalListener();
- if (qry.getRemoteFilter() != null && qry.getRemoteFilterFactory() != null)
- throw new IgniteException("Should be used either RemoterFilter or RemoteFilterFactory.");
+ rmtFilter = qry0.getRemoteFilter();
+ }
+ else {
+ ContinuousQueryWithTransformer<K, V, ?> qry0 = (ContinuousQueryWithTransformer<K, V, ?>)qry;
+
+ if (qry0.getLocalListener() == null)
+ throw new IgniteException("Mandatory local transformed event listener is not set for the query: " + qry);
+
+ if (qry0.getRemoteTransformerFactory() == null)
+ throw new IgniteException("Mandatory RemoteTransformerFactory is not set for the query: " + qry);
+
+ Collection<ClusterNode> nodes = context().grid().cluster().nodes();
+
+ for (ClusterNode node : nodes) {
+ if (node.version().compareTo(CONT_QRY_WITH_TRANSFORMER_SINCE) < 0) {
+ throw new IgniteException("Can't start ContinuousQueryWithTransformer, " +
+ "because some nodes in cluster doesn't support this feature: " + node);
+ }
+ }
+
+ locTransLsnr = qry0.getLocalListener();
+
+ rmtTransFactory = qry0.getRemoteTransformerFactory();
+ }
try {
final UUID routineId = ctx.continuousQueries().executeQuery(
- qry.getLocalListener(),
- qry.getRemoteFilter(),
+ locLsnr,
+ locTransLsnr,
+ rmtFilter,
qry.getRemoteFilterFactory(),
+ rmtTransFactory,
qry.getPageSize(),
qry.getTimeInterval(),
qry.isAutoUnsubscribe(),
@@ -596,8 +654,8 @@ public class IgniteCacheProxyImpl<K, V> extends AsyncSupportAdapter<IgniteCache<
boolean keepBinary = opCtxCall != null && opCtxCall.isKeepBinary();
- if (qry instanceof ContinuousQuery)
- return (QueryCursor<R>)queryContinuous((ContinuousQuery<K, V>)qry, qry.isLocal(), keepBinary);
+ if (qry instanceof ContinuousQuery || qry instanceof ContinuousQueryWithTransformer)
+ return (QueryCursor<R>)queryContinuous((AbstractContinuousQuery)qry, qry.isLocal(), keepBinary);
if (qry instanceof SqlQuery)
return (QueryCursor<R>)ctx.kernalContext().query().querySql(ctx, (SqlQuery)qry, keepBinary);
@@ -688,8 +746,8 @@ public class IgniteCacheProxyImpl<K, V> extends AsyncSupportAdapter<IgniteCache<
*/
private void validate(Query qry) {
if (!QueryUtils.isEnabled(ctx.config()) && !(qry instanceof ScanQuery) &&
- !(qry instanceof ContinuousQuery) && !(qry instanceof SpiQuery) && !(qry instanceof SqlQuery) &&
- !(qry instanceof SqlFieldsQuery))
+ !(qry instanceof ContinuousQuery) && !(qry instanceof ContinuousQueryWithTransformer) &&
+ !(qry instanceof SpiQuery) && !(qry instanceof SqlQuery) && !(qry instanceof SqlFieldsQuery))
throw new CacheException("Indexing is disabled for cache: " + ctx.cache().name() +
". Use setIndexedTypes or setTypeMetadata methods on CacheConfiguration to enable.");
http://git-wip-us.apache.org/repos/asf/ignite/blob/a83f3038/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java
index 7e3f0b5..88005d0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java
@@ -28,7 +28,6 @@ import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheDeployable;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -282,9 +281,8 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
* @throws IgniteCheckedException In case of error.
*/
void prepareMarshal(GridCacheContext cctx) throws IgniteCheckedException {
- assert key != null;
-
- key.prepareMarshal(cctx.cacheObjectContext());
+ if (key != null)
+ key.prepareMarshal(cctx.cacheObjectContext());
if (newVal != null)
newVal.prepareMarshal(cctx.cacheObjectContext());
@@ -300,7 +298,8 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
*/
void unmarshal(GridCacheContext cctx, @Nullable ClassLoader ldr) throws IgniteCheckedException {
if (!isFiltered()) {
- key.finishUnmarshal(cctx.cacheObjectContext(), ldr);
+ if (key != null)
+ key.finishUnmarshal(cctx.cacheObjectContext(), ldr);
if (newVal != null)
newVal.finishUnmarshal(cctx.cacheObjectContext(), ldr);
http://git-wip-us.apache.org/repos/asf/ignite/blob/a83f3038/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
----------------------------------------------------------------------
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 59b2a68..f0cd7ca 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
@@ -24,6 +24,7 @@ import java.io.ObjectOutput;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -39,6 +40,7 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
+import org.apache.ignite.cache.query.ContinuousQueryWithTransformer.EventListener;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.CacheQueryExecutedEvent;
import org.apache.ignite.events.CacheQueryReadEvent;
@@ -69,6 +71,7 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteAsyncCallback;
import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.thread.IgniteStripedThreadPoolExecutor;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -91,6 +94,19 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
static final int LSNR_MAX_BUF_SIZE =
IgniteSystemProperties.getInteger("IGNITE_CONTINUOUS_QUERY_LISTENER_MAX_BUFFER_SIZE", 10_000);
+ /**
+ * Transformer implementation for processing received remote events.
+ * They are already transformed so we simply return transformed value for event.
+ */
+ private transient IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, ?> returnValTrans =
+ new IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, Object>() {
+ @Override public Object apply(CacheEntryEvent<? extends K, ? extends V> evt) {
+ assert evt.getKey() == null;
+
+ return evt.getValue();
+ }
+ };
+
/** Cache name. */
private String cacheName;
@@ -161,7 +177,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
private transient boolean ignoreClsNotFound;
/** */
- private transient boolean asyncCb;
+ transient boolean asyncCb;
/** */
private transient UUID nodeId;
@@ -196,14 +212,13 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
public CacheContinuousQueryHandler(
String cacheName,
Object topic,
- CacheEntryUpdatedListener<K, V> locLsnr,
- CacheEntryEventSerializableFilter<K, V> rmtFilter,
+ @Nullable CacheEntryUpdatedListener<K, V> locLsnr,
+ @Nullable CacheEntryEventSerializableFilter<K, V> rmtFilter,
boolean oldValRequired,
boolean sync,
boolean ignoreExpired,
boolean ignoreClsNotFound) {
assert topic != null;
- assert locLsnr != null;
this.cacheName = cacheName;
this.topic = topic;
@@ -505,14 +520,14 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
if (asyncCb) {
ctx.asyncCallbackPool().execute(new Runnable() {
@Override public void run() {
- locLsnr.onUpdated(evts);
+ notifyLocalListener(evts, getTransformer());
}
}, part);
}
else
skipCtx.addProcessClosure(new Runnable() {
@Override public void run() {
- locLsnr.onUpdated(evts);
+ notifyLocalListener(evts, getTransformer());
}
});
}
@@ -584,6 +599,20 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
/**
+ * @return Cache entry event transformer.
+ */
+ @Nullable protected IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, ?> getTransformer() {
+ return null;
+ }
+
+ /**
+ * @return Local listener of transformed events.
+ */
+ @Nullable protected EventListener<?> localTransformedEventListener() {
+ return null;
+ }
+
+ /**
* @param cctx Context.
* @param nodeId ID of the node that started routine.
* @param entry Entry.
@@ -752,8 +781,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
}
- if (!entries0.isEmpty())
- locLsnr.onUpdated(entries0);
+ notifyLocalListener(entries0, returnValTrans);
}
/**
@@ -825,26 +853,31 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
if (cctx == null)
return;
- final CacheContinuousQueryEntry entry = evt.entry();
+ CacheContinuousQueryEntry entry = evt.entry();
+
+ IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, ?> trans = getTransformer();
if (loc) {
if (!locCache) {
Collection<CacheEntryEvent<? extends K, ? extends V>> evts = handleEvent(ctx, entry);
- if (!evts.isEmpty())
- locLsnr.onUpdated(evts);
+ notifyLocalListener(evts, trans);
if (!internal && !skipPrimaryCheck)
sendBackupAcknowledge(ackBuf.onAcknowledged(entry), routineId, ctx);
}
else {
if (!entry.isFiltered())
- locLsnr.onUpdated(F.<CacheEntryEvent<? extends K, ? extends V>>asList(evt));
+ notifyLocalListener(F.<CacheEntryEvent<? extends K, ? extends V>>asList(evt), trans);
}
}
else {
- if (!entry.isFiltered())
+ if (!entry.isFiltered()) {
+ if (trans != null)
+ entry = transformToEntry(trans, evt);
+
prepareEntry(cctx, nodeId, entry);
+ }
Object entryOrList = handleEntry(cctx, entry);
@@ -889,6 +922,28 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
/**
+ * Notifies local listener.
+ *
+ * @param evts Events.
+ * @param trans Transformer
+ */
+ private void notifyLocalListener(Collection<CacheEntryEvent<? extends K, ? extends V>> evts,
+ @Nullable IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, ?> trans) {
+ EventListener locTransLsnr = localTransformedEventListener();
+
+ assert (locLsnr != null && locTransLsnr == null) || (locLsnr == null && locTransLsnr != null);
+
+ if (F.isEmpty(evts))
+ return;
+
+ if (locLsnr != null)
+ locLsnr.onUpdated(evts);
+
+ if (locTransLsnr != null)
+ locTransLsnr.onUpdated(transform(trans, evts));
+ }
+
+ /**
* @return Task name.
*/
private String taskName() {
@@ -1258,4 +1313,72 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
}
+ /**
+ * @param trans Transformer.
+ * @param evts Source events.
+ * @return Collection of transformed values.
+ */
+ private Iterable transform(final IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, ?> trans,
+ Collection<CacheEntryEvent<? extends K, ? extends V>> evts) {
+ final Iterator<CacheEntryEvent<? extends K, ? extends V>> iter = evts.iterator();
+
+ return new Iterable() {
+ @NotNull @Override public Iterator iterator() {
+ return new Iterator() {
+ @Override public boolean hasNext() {
+ return iter.hasNext();
+ }
+
+ @Override public Object next() {
+ return transform(trans, iter.next());
+ }
+ };
+ }
+ };
+ }
+
+ /**
+ * Transform event data with {@link #getTransformer()} if exists.
+ *
+ * @param trans Transformer.
+ * @param evt Event to transform.
+ * @return Entry contains only transformed data if transformer exists. Unchanged event if transformer is not set.
+ * @see #getTransformer()
+ */
+ private CacheContinuousQueryEntry transformToEntry(IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, ?> trans,
+ CacheContinuousQueryEvent<? extends K, ? extends V> evt) {
+ Object transVal = transform(trans, evt);
+
+ return new CacheContinuousQueryEntry(evt.entry().cacheId(),
+ evt.entry().eventType(),
+ null,
+ transVal == null ? null : cacheContext(ctx).toCacheObject(transVal),
+ null,
+ evt.entry().isKeepBinary(),
+ evt.entry().partition(),
+ evt.entry().updateCounter(),
+ evt.entry().topologyVersion(),
+ evt.entry().flags());
+ }
+
+ /**
+ * @param trans Transformer.
+ * @param evt Event.
+ * @return Transformed value.
+ */
+ private Object transform(IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, ?> trans,
+ CacheEntryEvent<? extends K, ? extends V> evt) {
+ assert trans != null;
+
+ Object transVal = null;
+
+ try {
+ transVal = trans.apply(evt);
+ }
+ catch (Exception e) {
+ U.error(log, e);
+ }
+
+ return transVal;
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a83f3038/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV2.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV2.java
index e48d22e..86c1ae1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV2.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV2.java
@@ -41,7 +41,7 @@ public class CacheContinuousQueryHandlerV2<K, V> extends CacheContinuousQueryHan
private static final long serialVersionUID = 0L;
/** Remote filter factory. */
- private Factory<? extends CacheEntryEventFilter> rmtFilterFactory;
+ Factory<? extends CacheEntryEventFilter> rmtFilterFactory;
/** Deployable object for filter factory. */
private CacheContinuousQueryDeployableObject rmtFilterFactoryDep;
@@ -74,8 +74,8 @@ public class CacheContinuousQueryHandlerV2<K, V> extends CacheContinuousQueryHan
public CacheContinuousQueryHandlerV2(
String cacheName,
Object topic,
- CacheEntryUpdatedListener<K, V> locLsnr,
- Factory<? extends CacheEntryEventFilter<K, V>> rmtFilterFactory,
+ @Nullable CacheEntryUpdatedListener<K, V> locLsnr,
+ @Nullable Factory<? extends CacheEntryEventFilter<K, V>> rmtFilterFactory,
boolean oldValRequired,
boolean sync,
boolean ignoreExpired,
@@ -89,9 +89,6 @@ public class CacheContinuousQueryHandlerV2<K, V> extends CacheContinuousQueryHan
sync,
ignoreExpired,
ignoreClsNotFound);
-
- assert rmtFilterFactory != null;
-
this.rmtFilterFactory = rmtFilterFactory;
if (types != null) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/a83f3038/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV3.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV3.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV3.java
new file mode 100644
index 0000000..14d5605
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandlerV3.java
@@ -0,0 +1,185 @@
+/*
+ * 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.cache.query.continuous;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.UUID;
+import javax.cache.configuration.Factory;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryEventFilter;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.query.ContinuousQueryWithTransformer;
+import org.apache.ignite.cache.query.ContinuousQueryWithTransformer.EventListener;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteAsyncCallback;
+import org.apache.ignite.lang.IgniteClosure;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Continuous query handler V3 version.
+ * Contains {@link Factory} for remote transformer and {@link EventListener}.
+ *
+ * @see ContinuousQueryWithTransformer
+ */
+public class CacheContinuousQueryHandlerV3<K, V> extends CacheContinuousQueryHandlerV2<K, V> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Remote transformer. */
+ private Factory<? extends IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, ?>> rmtTransFactory;
+
+ /** Deployable object for transformer. */
+ private CacheContinuousQueryDeployableObject rmtTransFactoryDep;
+
+ /** Remote transformer. */
+ private transient IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, ?> rmtTrans;
+
+ /** Local listener for transformed events. */
+ private transient EventListener<?> locTransLsnr;
+
+ /**
+ * Empty constructor.
+ */
+ public CacheContinuousQueryHandlerV3() {
+ super();
+ }
+
+ /**
+ * @param cacheName Cache name.
+ * @param topic Topic.
+ * @param locTransLsnr Local listener of transformed events
+ * @param rmtFilterFactory Remote filter factory.
+ * @param rmtTransFactory Remote transformer factory.
+ * @param oldValRequired OldValRequired flag.
+ * @param sync Sync flag.
+ * @param ignoreExpired IgnoreExpired flag.
+ * @param ignoreClsNotFound IgnoreClassNotFoundException flag.
+ */
+ public CacheContinuousQueryHandlerV3(
+ String cacheName,
+ Object topic,
+ EventListener<?> locTransLsnr,
+ @Nullable Factory<? extends CacheEntryEventFilter<K, V>> rmtFilterFactory,
+ Factory<? extends IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, ?>> rmtTransFactory,
+ boolean oldValRequired,
+ boolean sync,
+ boolean ignoreExpired,
+ boolean ignoreClsNotFound) {
+ super(
+ cacheName,
+ topic,
+ null,
+ rmtFilterFactory,
+ oldValRequired,
+ sync,
+ ignoreExpired,
+ ignoreClsNotFound,
+ null);
+
+ assert locTransLsnr != null;
+ assert rmtTransFactory != null;
+
+ this.locTransLsnr = locTransLsnr;
+ this.rmtTransFactory = rmtTransFactory;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, ?> getTransformer() {
+ if (rmtTrans == null && rmtTransFactory != null)
+ rmtTrans = rmtTransFactory.create();
+
+ return rmtTrans;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected EventListener<?> localTransformedEventListener() {
+ return locTransLsnr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public CacheEntryEventFilter<K, V> getEventFilter() {
+ if (rmtFilterFactory == null)
+ return null;
+
+ return super.getEventFilter();
+ }
+
+ /** {@inheritDoc} */
+ @Override public RegisterStatus register(UUID nodeId, UUID routineId,
+ GridKernalContext ctx) throws IgniteCheckedException {
+ final IgniteClosure trans = getTransformer();
+
+ if (trans != null)
+ ctx.resource().injectGeneric(trans);
+
+ if (locTransLsnr != null) {
+ ctx.resource().injectGeneric(locTransLsnr);
+
+ asyncCb = U.hasAnnotation(locTransLsnr, IgniteAsyncCallback.class);
+ }
+
+ return super.register(nodeId, routineId, ctx);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void p2pMarshal(GridKernalContext ctx) throws IgniteCheckedException {
+ super.p2pMarshal(ctx);
+
+ if (rmtTransFactory != null && !U.isGrid(rmtTransFactory.getClass()))
+ rmtTransFactoryDep = new CacheContinuousQueryDeployableObject(rmtTransFactory, ctx);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void p2pUnmarshal(UUID nodeId, GridKernalContext ctx) throws IgniteCheckedException {
+ super.p2pUnmarshal(nodeId, ctx);
+
+ if (rmtTransFactoryDep != null)
+ rmtTransFactory = rmtTransFactoryDep.unmarshal(nodeId, ctx);
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ super.writeExternal(out);
+
+ boolean b = rmtTransFactoryDep != null;
+
+ out.writeBoolean(b);
+
+ if (b)
+ out.writeObject(rmtTransFactoryDep);
+ else
+ out.writeObject(rmtTransFactory);
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ super.readExternal(in);
+
+ boolean b = in.readBoolean();
+
+ if (b)
+ rmtTransFactoryDep = (CacheContinuousQueryDeployableObject)in.readObject();
+ else
+ rmtTransFactory = (Factory<? extends IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, ?>>)in.readObject();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a83f3038/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
index 628111b..1e131ef 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
@@ -49,6 +49,7 @@ import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.query.CacheQueryEntryEvent;
import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cache.query.ContinuousQueryWithTransformer.EventListener;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -57,8 +58,8 @@ import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheManagerAdapter;
import org.apache.ignite.internal.processors.cache.IgniteCacheProxy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicAbstractUpdateFuture;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.continuous.GridContinuousHandler;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.CI2;
@@ -66,6 +67,7 @@ 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.lang.IgniteAsyncCallback;
+import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteOutClosure;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.plugin.security.SecurityPermission;
@@ -434,6 +436,7 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
/**
* @param locLsnr Local listener.
* @param rmtFilter Remote filter.
+ * @param rmtFilterFactory Remote filter factory
* @param bufSize Buffer size.
* @param timeInterval Time interval.
* @param autoUnsubscribe Auto unsubscribe flag.
@@ -441,9 +444,11 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
* @return Continuous routine ID.
* @throws IgniteCheckedException In case of error.
*/
- public UUID executeQuery(final CacheEntryUpdatedListener locLsnr,
+ public UUID executeQuery(@Nullable final CacheEntryUpdatedListener locLsnr,
+ @Nullable final EventListener locTransLsnr,
@Nullable final CacheEntryEventSerializableFilter rmtFilter,
@Nullable final Factory<? extends CacheEntryEventFilter> rmtFilterFactory,
+ @Nullable final Factory<? extends IgniteClosure> rmtTransFactory,
int bufSize,
long timeInterval,
boolean autoUnsubscribe,
@@ -453,12 +458,30 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
{
IgniteOutClosure<CacheContinuousQueryHandler> clsr;
- if (rmtFilterFactory != null)
+ if (rmtTransFactory != null) {
+ clsr = new IgniteOutClosure<CacheContinuousQueryHandler>() {
+ @Override public CacheContinuousQueryHandler apply() {
+ assert locTransLsnr != null;
+
+ return new CacheContinuousQueryHandlerV3(
+ cctx.name(),
+ TOPIC_CACHE.topic(topicPrefix, cctx.localNodeId(), seq.getAndIncrement()),
+ locTransLsnr,
+ rmtFilterFactory,
+ rmtTransFactory,
+ true,
+ false,
+ !includeExpired,
+ false);
+ }
+ };
+ }
+ else if (rmtFilterFactory != null) {
clsr = new IgniteOutClosure<CacheContinuousQueryHandler>() {
@Override public CacheContinuousQueryHandler apply() {
- CacheContinuousQueryHandler hnd;
+ assert locLsnr != null;
- hnd = new CacheContinuousQueryHandlerV2(
+ return new CacheContinuousQueryHandlerV2(
cctx.name(),
TOPIC_CACHE.topic(topicPrefix, cctx.localNodeId(), seq.getAndIncrement()),
locLsnr,
@@ -468,13 +491,15 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
!includeExpired,
false,
null);
-
- return hnd;
}
};
- else
+ }
+ else {
clsr = new IgniteOutClosure<CacheContinuousQueryHandler>() {
@Override public CacheContinuousQueryHandler apply() {
+ assert locLsnr != null;
+ assert locTransLsnr == null;
+
return new CacheContinuousQueryHandler(
cctx.name(),
TOPIC_CACHE.topic(topicPrefix, cctx.localNodeId(), seq.getAndIncrement()),
@@ -486,6 +511,7 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
false);
}
};
+ }
return executeQuery0(
locLsnr,
@@ -676,6 +702,8 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
}
if (notifyExisting) {
+ assert locLsnr != null : "Local listener can't be null if notification for existing entries are enabled";
+
final Iterator<CacheDataRow> it = cctx.offheap().cacheIterator(cctx.cacheId(),
true,
true,
http://git-wip-us.apache.org/repos/asf/ignite/blob/a83f3038/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTest.java
index 142ff35..9a815de 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTest.java
@@ -35,6 +35,7 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
import javax.cache.Cache;
import javax.cache.configuration.Factory;
+import javax.cache.configuration.FactoryBuilder;
import javax.cache.event.CacheEntryEvent;
import javax.cache.event.CacheEntryEventFilter;
import javax.cache.event.CacheEntryListenerException;
@@ -51,8 +52,11 @@ import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.affinity.Affinity;
+import org.apache.ignite.cache.query.AbstractContinuousQuery;
import org.apache.ignite.cache.query.CacheQueryEntryEvent;
import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cache.query.ContinuousQueryWithTransformer;
+import org.apache.ignite.cache.query.ContinuousQueryWithTransformer.EventListener;
import org.apache.ignite.cache.query.QueryCursor;
import org.apache.ignite.cache.store.CacheStore;
import org.apache.ignite.cache.store.CacheStoreAdapter;
@@ -62,6 +66,7 @@ import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.PA;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
@@ -464,7 +469,7 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
IgniteCache<QueryTestKey, QueryTestValue> cache = grid(getClientIndex()).createCache(ccfg);
try {
- ContinuousQuery<QueryTestKey, QueryTestValue> qry = new ContinuousQuery<>();
+ AbstractContinuousQuery<QueryTestKey, QueryTestValue> qry = createQuery();
final List<CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> evts =
new CopyOnWriteArrayList<>();
@@ -472,13 +477,20 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
if (noOpFilterFactory() != null)
qry.setRemoteFilterFactory(noOpFilterFactory());
- qry.setLocalListener(new CacheEntryUpdatedListener<QueryTestKey, QueryTestValue>() {
- @Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey,
- ? extends QueryTestValue>> events) throws CacheEntryListenerException {
- for (CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> e : events)
- evts.add(e);
- }
- });
+ if (qry instanceof ContinuousQuery) {
+ ((ContinuousQuery<QueryTestKey, QueryTestValue>)qry).setLocalListener(new CacheEntryUpdatedListener<QueryTestKey, QueryTestValue>() {
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> events) throws CacheEntryListenerException {
+ for (CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> e : events)
+ evts.add(e);
+ }
+ });
+ }
+ else if (qry instanceof ContinuousQueryWithTransformer)
+ initQueryWithTransformer(
+ (ContinuousQueryWithTransformer<QueryTestKey, QueryTestValue, CacheEntryEvent>)qry, evts);
+ else
+ fail("Unknown query type");
QueryTestKey key = new QueryTestKey(1);
@@ -595,7 +607,7 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
IgniteCache<QueryTestKey, QueryTestValue> cache = grid(getClientIndex()).createCache(ccfg);
try {
- ContinuousQuery<QueryTestKey, QueryTestValue> qry = new ContinuousQuery<>();
+ AbstractContinuousQuery<QueryTestKey, QueryTestValue> qry = createQuery();
final List<CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> evts =
new CopyOnWriteArrayList<>();
@@ -603,13 +615,20 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
if (noOpFilterFactory() != null)
qry.setRemoteFilterFactory(noOpFilterFactory());
- qry.setLocalListener(new CacheEntryUpdatedListener<QueryTestKey, QueryTestValue>() {
- @Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey,
- ? extends QueryTestValue>> events) throws CacheEntryListenerException {
- for (CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> e : events)
- evts.add(e);
- }
- });
+ if (qry instanceof ContinuousQuery) {
+ ((ContinuousQuery<QueryTestKey, QueryTestValue>)qry).setLocalListener(new CacheEntryUpdatedListener<QueryTestKey, QueryTestValue>() {
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> events) throws CacheEntryListenerException {
+ for (CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> e : events)
+ evts.add(e);
+ }
+ });
+ }
+ else if (qry instanceof ContinuousQueryWithTransformer)
+ initQueryWithTransformer(
+ (ContinuousQueryWithTransformer<QueryTestKey, QueryTestValue, CacheEntryEvent>)qry, evts);
+ else
+ fail("Unknown query type");
Map<QueryTestKey, QueryTestValue> map = new TreeMap<>();
@@ -834,16 +853,23 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
Collection<QueryCursor<?>> curs = new ArrayList<>();
if (deploy == CLIENT) {
- ContinuousQuery<Object, Object> qry = new ContinuousQuery<>();
+ AbstractContinuousQuery<Object, Object> qry = createQuery();
final BlockingQueue<CacheEntryEvent<?, ?>> evtsQueue = new ArrayBlockingQueue<>(50_000);
- qry.setLocalListener(new CacheEntryUpdatedListener<Object, Object>() {
- @Override public void onUpdated(Iterable<CacheEntryEvent<?, ?>> evts) {
- for (CacheEntryEvent<?, ?> evt : evts)
- evtsQueue.add(evt);
- }
- });
+ if (qry instanceof ContinuousQuery) {
+ ((ContinuousQuery<Object, Object>)qry).setLocalListener(new CacheEntryUpdatedListener<Object, Object>() {
+ @Override public void onUpdated(Iterable<CacheEntryEvent<?, ?>> evts) {
+ for (CacheEntryEvent<?, ?> evt : evts)
+ evtsQueue.add(evt);
+ }
+ });
+ }
+ else if (qry instanceof ContinuousQueryWithTransformer)
+ initQueryWithTransformer(
+ (ContinuousQueryWithTransformer<Object, Object, CacheEntryEvent>)qry, evtsQueue);
+ else
+ fail("Unknown query type");
evtsQueues.add(evtsQueue);
@@ -852,16 +878,23 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
curs.add(cur);
}
else if (deploy == SERVER) {
- ContinuousQuery<Object, Object> qry = new ContinuousQuery<>();
+ AbstractContinuousQuery<Object, Object> qry = createQuery();
final BlockingQueue<CacheEntryEvent<?, ?>> evtsQueue = new ArrayBlockingQueue<>(50_000);
- qry.setLocalListener(new CacheEntryUpdatedListener<Object, Object>() {
- @Override public void onUpdated(Iterable<CacheEntryEvent<?, ?>> evts) {
- for (CacheEntryEvent<?, ?> evt : evts)
- evtsQueue.add(evt);
- }
- });
+ if (qry instanceof ContinuousQuery) {
+ ((ContinuousQuery<Object, Object>)qry).setLocalListener(new CacheEntryUpdatedListener<Object, Object>() {
+ @Override public void onUpdated(Iterable<CacheEntryEvent<?, ?>> evts) {
+ for (CacheEntryEvent<?, ?> evt : evts)
+ evtsQueue.add(evt);
+ }
+ });
+ }
+ else if (qry instanceof ContinuousQueryWithTransformer)
+ initQueryWithTransformer(
+ (ContinuousQueryWithTransformer<Object, Object, CacheEntryEvent>)qry, evtsQueue);
+ else
+ fail("Unknown query type");
evtsQueues.add(evtsQueue);
@@ -871,16 +904,23 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
}
else {
for (int i = 0; i <= getServerNodeCount(); i++) {
- ContinuousQuery<Object, Object> qry = new ContinuousQuery<>();
+ AbstractContinuousQuery<Object, Object> qry = createQuery();
final BlockingQueue<CacheEntryEvent<?, ?>> evtsQueue = new ArrayBlockingQueue<>(50_000);
- qry.setLocalListener(new CacheEntryUpdatedListener<Object, Object>() {
- @Override public void onUpdated(Iterable<CacheEntryEvent<?, ?>> evts) {
- for (CacheEntryEvent<?, ?> evt : evts)
- evtsQueue.add(evt);
- }
- });
+ if (qry instanceof ContinuousQuery) {
+ ((ContinuousQuery<Object, Object>)qry).setLocalListener(new CacheEntryUpdatedListener<Object, Object>() {
+ @Override public void onUpdated(Iterable<CacheEntryEvent<?, ?>> evts) {
+ for (CacheEntryEvent<?, ?> evt : evts)
+ evtsQueue.add(evt);
+ }
+ });
+ }
+ else if (qry instanceof ContinuousQueryWithTransformer)
+ initQueryWithTransformer(
+ (ContinuousQueryWithTransformer<Object, Object, CacheEntryEvent>)qry, evtsQueue);
+ else
+ fail("Unknown query type");
evtsQueues.add(evtsQueue);
@@ -1417,6 +1457,15 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
}
/**
+ * @param <K> Key type.
+ * @param <V> Value type.
+ * @return New instance of continuous query.
+ */
+ protected <K, V> AbstractContinuousQuery<K, V> createQuery() {
+ return new ContinuousQuery<>();
+ }
+
+ /**
*
*/
private static class TestStoreFactory implements Factory<CacheStore<Object, Object>> {
@@ -1586,4 +1635,37 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
protected enum ContinuousDeploy {
CLIENT, SERVER, ALL
}
+
+ /**
+ * Initialize continuous query with transformer.
+ * Query will accumulate all events in accumulator.
+ *
+ * @param qry Continuous query.
+ * @param acc Accumulator for events.
+ * @param <K> Key type.
+ * @param <V> Value type.
+ */
+ private <K, V> void initQueryWithTransformer(
+ ContinuousQueryWithTransformer<K, V, CacheEntryEvent> qry,
+ Collection<CacheEntryEvent<? extends K, ? extends V>> acc) {
+
+ IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, CacheEntryEvent> transformer =
+ new IgniteClosure<CacheEntryEvent<? extends K, ? extends V>, CacheEntryEvent>() {
+ @Override public CacheEntryEvent apply(CacheEntryEvent<? extends K, ? extends V> event) {
+ return event;
+ }
+ };
+
+ ContinuousQueryWithTransformer<K, V, CacheEntryEvent> qry0 =
+ (ContinuousQueryWithTransformer<K, V, CacheEntryEvent>)qry;
+
+ qry0.setRemoteTransformerFactory(FactoryBuilder.factoryOf(transformer));
+
+ qry0.setLocalListener(new EventListener<CacheEntryEvent>() {
+ @Override public void onUpdated(Iterable<? extends CacheEntryEvent> events) {
+ for (CacheEntryEvent e : events)
+ acc.add(e);
+ }
+ });
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a83f3038/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerClientSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerClientSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerClientSelfTest.java
new file mode 100644
index 0000000..8d8b4c1
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerClientSelfTest.java
@@ -0,0 +1,40 @@
+/*
+ * 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.cache.query.continuous;
+
+import org.apache.ignite.Ignite;
+
+/**
+ */
+public class CacheContinuousWithTransformerClientSelfTest extends CacheContinuousWithTransformerReplicatedSelfTest {
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ client = true;
+
+ startGrid("client");
+
+ client = false;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected Ignite gridToRunQuery() throws Exception {
+ return grid("client");
+ }
+}