You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2014/12/31 13:53:24 UTC
[1/2] incubator-ignite git commit: # ignite-43
Repository: incubator-ignite
Updated Branches:
refs/heads/ignite-43 dc3faefe4 -> f9f023159
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java
new file mode 100644
index 0000000..e97c0a6
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java
@@ -0,0 +1,24 @@
+/* @java.file.header */
+
+/* _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.gridgain.grid.cache.*;
+
+import static org.gridgain.grid.cache.GridCacheMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheEntryListenerAtomicReplicatedTest extends IgniteCacheEntryListenerAtomicTest {
+ /** {@inheritDoc} */
+ @Override protected GridCacheMode cacheMode() {
+ return REPLICATED;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicTest.java
new file mode 100644
index 0000000..3f96744
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicTest.java
@@ -0,0 +1,47 @@
+/* @java.file.header */
+
+/* _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.gridgain.grid.cache.*;
+
+import static org.gridgain.grid.cache.GridCacheAtomicWriteOrderMode.*;
+import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
+import static org.gridgain.grid.cache.GridCacheDistributionMode.*;
+import static org.gridgain.grid.cache.GridCacheMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheEntryListenerAtomicTest extends IgniteCacheEntryListenerAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected int gridCount() {
+ return 3;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheMode cacheMode() {
+ return PARTITIONED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheAtomicityMode atomicityMode() {
+ return ATOMIC;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheAtomicWriteOrderMode atomicWriteOrderMode() {
+ return PRIMARY;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheDistributionMode distributionMode() {
+ return PARTITIONED_ONLY;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerTxLocalTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerTxLocalTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerTxLocalTest.java
new file mode 100644
index 0000000..86ed67d
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerTxLocalTest.java
@@ -0,0 +1,41 @@
+/* @java.file.header */
+
+/* _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.gridgain.grid.cache.*;
+
+import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
+import static org.gridgain.grid.cache.GridCacheDistributionMode.*;
+import static org.gridgain.grid.cache.GridCacheMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheEntryListenerTxLocalTest extends IgniteCacheEntryListenerAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected int gridCount() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheMode cacheMode() {
+ return LOCAL;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheAtomicityMode atomicityMode() {
+ return TRANSACTIONAL;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheDistributionMode distributionMode() {
+ return PARTITIONED_ONLY;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerTxReplicatedTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerTxReplicatedTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerTxReplicatedTest.java
new file mode 100644
index 0000000..ff8a100
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerTxReplicatedTest.java
@@ -0,0 +1,24 @@
+/* @java.file.header */
+
+/* _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.gridgain.grid.cache.*;
+
+import static org.gridgain.grid.cache.GridCacheMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheEntryListenerTxReplicatedTest extends IgniteCacheEntryListenerTxTest {
+ /** {@inheritDoc} */
+ @Override protected GridCacheMode cacheMode() {
+ return REPLICATED;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerTxTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerTxTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerTxTest.java
new file mode 100644
index 0000000..b725289
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerTxTest.java
@@ -0,0 +1,41 @@
+/* @java.file.header */
+
+/* _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.gridgain.grid.cache.*;
+
+import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
+import static org.gridgain.grid.cache.GridCacheDistributionMode.*;
+import static org.gridgain.grid.cache.GridCacheMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheEntryListenerTxTest extends IgniteCacheEntryListenerAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected int gridCount() {
+ return 3;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheMode cacheMode() {
+ return PARTITIONED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheAtomicityMode atomicityMode() {
+ return TRANSACTIONAL;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheDistributionMode distributionMode() {
+ return PARTITIONED_ONLY;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/test/java/org/gridgain/testsuites/bamboo/GridDataGridTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/testsuites/bamboo/GridDataGridTestSuite.java b/modules/core/src/test/java/org/gridgain/testsuites/bamboo/GridDataGridTestSuite.java
index d9a5e22..efe49fb 100644
--- a/modules/core/src/test/java/org/gridgain/testsuites/bamboo/GridDataGridTestSuite.java
+++ b/modules/core/src/test/java/org/gridgain/testsuites/bamboo/GridDataGridTestSuite.java
@@ -38,6 +38,13 @@ public class GridDataGridTestSuite extends TestSuite {
public static TestSuite suite() throws Exception {
TestSuite suite = new TestSuite("IgniteCache Test Suite");
+ suite.addTestSuite(IgniteCacheEntryListenerAtomicTest.class);
+ suite.addTestSuite(IgniteCacheEntryListenerAtomicReplicatedTest.class);
+ suite.addTestSuite(IgniteCacheEntryListenerAtomicLocalTest.class);
+ suite.addTestSuite(IgniteCacheEntryListenerTxTest.class);
+ suite.addTestSuite(IgniteCacheEntryListenerTxReplicatedTest.class);
+ suite.addTestSuite(IgniteCacheEntryListenerTxLocalTest.class);
+
suite.addTest(IgniteCacheExpiryPolicyTestSuite.suite());
suite.addTestSuite(IgniteCacheAtomicInvokeTest.class);
Re: [1/2] incubator-ignite git commit: # ignite-43
Posted by Semyon Boikov <sb...@gridgain.com>.
Sorry, it was me, I forgot to change file template in IDE, now fixed.
Semyon
On Wed, Jan 14, 2015 at 12:11 AM, Branko Čibej <br...@apache.org> wrote:
> On 06.01.2015 13:31, Branko Čibej wrote:
> > On 31.12.2014 13:53, sboikov@apache.org wrote:
> >> Repository: incubator-ignite
> >> Updated Branches:
> >> refs/heads/ignite-43 dc3faefe4 -> f9f023159
> >>
> >>
> >>
> http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java
> >> ----------------------------------------------------------------------
> >> diff --git
> a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java
> b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java
> >> new file mode 100644
> >> index 0000000..e97c0a6
> >> --- /dev/null
> >> +++
> b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java
> >> @@ -0,0 +1,24 @@
> >> +/* @java.file.header */
> >> +
> >> +/* _________ _____ __________________ _____
> >> + * __ ____/___________(_)______ /__ ____/______ ____(_)_______
> >> + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
> >> + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
> >> + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
> >> + */
> >
> > IMO it would be an amazingly good idea to change the file headers as you
> > import them into the ASF repo. You're only creating extra work for
> > yourselves delaying this task.
> >
> > -- Brane
> >
>
> Guys, you're still committing files with invalid headers. Please stop.
> You can't even create a nightly snapshot with wrong headers all over the
> source, and definitely not a release.
>
> -- Brane
>
Re: [1/2] incubator-ignite git commit: # ignite-43
Posted by Branko Čibej <br...@apache.org>.
On 06.01.2015 13:31, Branko Čibej wrote:
> On 31.12.2014 13:53, sboikov@apache.org wrote:
>> Repository: incubator-ignite
>> Updated Branches:
>> refs/heads/ignite-43 dc3faefe4 -> f9f023159
>>
>>
>> http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java
>> ----------------------------------------------------------------------
>> diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java
>> new file mode 100644
>> index 0000000..e97c0a6
>> --- /dev/null
>> +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java
>> @@ -0,0 +1,24 @@
>> +/* @java.file.header */
>> +
>> +/* _________ _____ __________________ _____
>> + * __ ____/___________(_)______ /__ ____/______ ____(_)_______
>> + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
>> + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
>> + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
>> + */
>
> IMO it would be an amazingly good idea to change the file headers as you
> import them into the ASF repo. You're only creating extra work for
> yourselves delaying this task.
>
> -- Brane
>
Guys, you're still committing files with invalid headers. Please stop.
You can't even create a nightly snapshot with wrong headers all over the
source, and definitely not a release.
-- Brane
Re: [1/2] incubator-ignite git commit: # ignite-43
Posted by Branko Čibej <br...@apache.org>.
On 31.12.2014 13:53, sboikov@apache.org wrote:
> Repository: incubator-ignite
> Updated Branches:
> refs/heads/ignite-43 dc3faefe4 -> f9f023159
>
>
> http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java
> ----------------------------------------------------------------------
> diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java
> new file mode 100644
> index 0000000..e97c0a6
> --- /dev/null
> +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicReplicatedTest.java
> @@ -0,0 +1,24 @@
> +/* @java.file.header */
> +
> +/* _________ _____ __________________ _____
> + * __ ____/___________(_)______ /__ ____/______ ____(_)_______
> + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
> + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
> + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
> + */
IMO it would be an amazingly good idea to change the file headers as you
import them into the ASF repo. You're only creating extra work for
yourselves delaying this task.
-- Brane
[2/2] incubator-ignite git commit: # ignite-43
Posted by sb...@apache.org.
# ignite-43
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/f9f02315
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/f9f02315
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/f9f02315
Branch: refs/heads/ignite-43
Commit: f9f02315992b046106b8fad38b14c37887e6103c
Parents: dc3faef
Author: sboikov <sb...@gridgain.com>
Authored: Wed Dec 31 12:03:37 2014 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Wed Dec 31 15:53:08 2014 +0300
----------------------------------------------------------------------
.../apache/ignite/cache/CacheEntryEvent.java | 29 +-
.../processors/cache/IgniteCacheProxy.java | 323 +----------
.../grid/cache/GridCacheConfiguration.java | 1 +
.../cache/query/GridCacheContinuousQuery.java | 10 +-
.../grid/kernal/GridEventConsumeHandler.java | 4 +-
.../processors/cache/GridCacheMapEntry.java | 49 +-
.../processors/cache/GridCacheProcessor.java | 2 +-
.../GridCacheDataStructuresManager.java | 2 +-
.../GridCacheContinuousQueryAdapter.java | 39 +-
.../GridCacheContinuousQueryEntry.java | 28 +-
.../GridCacheContinuousQueryHandler.java | 23 +-
.../GridCacheContinuousQueryHandlerV2.java | 12 +-
.../GridCacheContinuousQueryManager.java | 498 ++++++++++++++++-
.../service/GridServiceProcessor.java | 4 +-
.../IgniteCacheEntryListenerAbstractTest.java | 535 +++++++++++++++++--
...IgniteCacheEntryListenerAtomicLocalTest.java | 41 ++
...eCacheEntryListenerAtomicReplicatedTest.java | 24 +
.../IgniteCacheEntryListenerAtomicTest.java | 47 ++
.../IgniteCacheEntryListenerTxLocalTest.java | 41 ++
...gniteCacheEntryListenerTxReplicatedTest.java | 24 +
.../cache/IgniteCacheEntryListenerTxTest.java | 41 ++
.../bamboo/GridDataGridTestSuite.java | 7 +
22 files changed, 1341 insertions(+), 443 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/apache/ignite/cache/CacheEntryEvent.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/CacheEntryEvent.java b/modules/core/src/main/java/org/apache/ignite/cache/CacheEntryEvent.java
index b3a4f52..b480ca4 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/CacheEntryEvent.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/CacheEntryEvent.java
@@ -10,7 +10,7 @@
package org.apache.ignite.cache;
import org.apache.ignite.*;
-import org.apache.ignite.events.*;
+import org.gridgain.grid.cache.query.*;
import javax.cache.event.*;
@@ -19,52 +19,49 @@ import javax.cache.event.*;
*/
public class CacheEntryEvent<K, V> extends javax.cache.event.CacheEntryEvent<K, V> {
/** */
- private final IgniteCacheEvent evt;
+ private final GridCacheContinuousQueryEntry<K, V> e;
/**
* @param src Cache.
* @param type Event type.
- * @param evt Ignite event.
+ * @param e Ignite event.
*/
- public CacheEntryEvent(IgniteCache src, EventType type, IgniteCacheEvent evt) {
+ public CacheEntryEvent(IgniteCache src, EventType type, GridCacheContinuousQueryEntry<K, V> e) {
super(src, type);
- this.evt = evt;
+ this.e = e;
}
/** {@inheritDoc} */
- @SuppressWarnings("unchecked")
@Override public V getOldValue() {
- return (V)evt.oldValue();
+ return e.getOldValue();
}
/** {@inheritDoc} */
@Override public boolean isOldValueAvailable() {
- return evt.hasOldValue();
+ return e.getOldValue() != null;
}
/** {@inheritDoc} */
@Override public K getKey() {
- return evt.key();
+ return e.getKey();
}
/** {@inheritDoc} */
- @SuppressWarnings("unchecked")
@Override public V getValue() {
- return (V)evt.newValue();
+ return e.getValue();
}
/** {@inheritDoc} */
- @SuppressWarnings("unchecked")
@Override public <T> T unwrap(Class<T> cls) {
- if (cls.equals(IgniteCacheEvent.class))
- return (T)evt;
-
throw new IllegalArgumentException();
}
/** {@inheritDoc} */
@Override public String toString() {
- return "CacheEntryEvent [evtType=" + getEventType() + ", evt=" + evt + ']';
+ return "CacheEntryEvent [evtType=" + getEventType() +
+ ", key=" + getKey() +
+ ", val=" + getValue() +
+ ", oldVal=" + getOldValue() + ']';
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
index 47d0722..b526f15 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
@@ -11,11 +11,8 @@ package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.*;
import org.apache.ignite.cache.*;
-import org.apache.ignite.cache.CacheEntryEvent;
import org.apache.ignite.cache.query.*;
-import org.apache.ignite.events.*;
import org.apache.ignite.lang.*;
-import org.apache.ignite.resources.*;
import org.gridgain.grid.cache.*;
import org.gridgain.grid.kernal.*;
import org.gridgain.grid.kernal.processors.cache.*;
@@ -26,7 +23,6 @@ import org.jetbrains.annotations.*;
import javax.cache.*;
import javax.cache.configuration.*;
-import javax.cache.event.*;
import javax.cache.expiry.*;
import javax.cache.integration.*;
import javax.cache.processor.*;
@@ -34,8 +30,6 @@ import java.io.*;
import java.util.*;
import java.util.concurrent.locks.*;
-import static org.apache.ignite.events.IgniteEventType.*;
-
/**
* Cache proxy.
*/
@@ -771,325 +765,36 @@ public class IgniteCacheProxy<K, V> extends IgniteAsyncSupportAdapter implements
}
/** {@inheritDoc} */
- @Override public void registerCacheEntryListener(CacheEntryListenerConfiguration lsnrCfg) {
+ @Override public void registerCacheEntryListener(CacheEntryListenerConfiguration<K, V> lsnrCfg) {
GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
try {
- A.notNull(lsnrCfg, "lsnrCfg");
-
- Factory<CacheEntryListener<? super K, ? super V>> factory = lsnrCfg.getCacheEntryListenerFactory();
-
- A.notNull(factory, "cacheEntryListenerFactory");
-
- CacheEntryListener lsnr = factory.create();
-
- A.notNull(lsnr, "lsnr");
-
- EventCallback cb = new EventCallback(lsnr);
-
- Set<Integer> types = new HashSet<>();
-
- if (cb.create() || cb.update())
- types.add(EVT_CACHE_OBJECT_PUT);
-
- if (cb.remove())
- types.add(EVT_CACHE_OBJECT_REMOVED);
-
- if (cb.expire())
- types.add(EVT_CACHE_OBJECT_EXPIRED);
-
- if (types.isEmpty())
- throw new IllegalArgumentException();
-
- int[] types0 = new int[types.size()];
-
- int i = 0;
-
- for (Integer type : types)
- types0[i++] = type;
-
- EventFilter fltr = new EventFilter(cb.create(),
- cb.update(),
- lsnrCfg.getCacheEntryEventFilterFactory(),
- ignite(),
- ctx.name());
-
- IgniteFuture<UUID> fut = ctx.kernalContext().continuous().startRoutine(
- new GridEventConsumeHandler(cb, fltr, types0),
- 1,
- 0,
- true,
- null);
-
- try {
- fut.get();
- }
- catch (IgniteCheckedException e) {
- throw new IgniteException(e);
- }
+ ctx.continuousQueries().registerCacheEntryListener(lsnrCfg);
+ }
+ catch (IgniteCheckedException e) {
+ throw cacheException(e);
}
finally {
gate.leave(prev);
}
}
- /**
- *
- */
- static class EventFilter implements IgnitePredicate<IgniteEvent>, Externalizable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** */
- private boolean update;
-
- /** */
- private boolean create;
-
- /** */
- private Factory<CacheEntryEventFilter> fltrFactory;
-
- /** */
- private CacheEntryEventFilter fltr;
-
- /** */
- @IgniteInstanceResource
- private Ignite ignite;
-
- /** */
- private IgniteCache cache;
-
- /** */
- private String cacheName;
-
- /**
- *
- */
- public EventFilter() {
- // No-op.
- }
-
- /**
- * @param create {@code True} if listens for create event.
- * @param update {@code True} if listens for create event.
- * @param fltrFactory Filter factory.
- * @param ignite Ignite instance.
- * @param cacheName Cache name.
- */
- EventFilter(
- boolean create,
- boolean update,
- Factory<CacheEntryEventFilter> fltrFactory,
- Ignite ignite,
- @Nullable String cacheName) {
- this.update = update;
- this.create = create;
- this.fltrFactory = fltrFactory;
- this.ignite = ignite;
- this.cacheName = cacheName;
-
- if (fltrFactory != null)
- fltr = fltrFactory.create();
-
- cache = ignite.jcache(cacheName);
-
- assert cache != null : cacheName;
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings("unchecked")
- @Override public boolean apply(IgniteEvent evt) {
- assert evt instanceof IgniteCacheEvent : evt;
-
- IgniteCacheEvent cacheEvt = (IgniteCacheEvent)evt;
-
- EventType evtType;
-
- switch (cacheEvt.type()) {
- case EVT_CACHE_OBJECT_REMOVED: {
- evtType = EventType.REMOVED;
-
- break;
- }
-
- case EVT_CACHE_OBJECT_PUT: {
- assert update || create;
-
- if (cacheEvt.hasOldValue()) {
- if (!update)
- return false;
-
- evtType = EventType.UPDATED;
- }
- else {
- if (!create)
- return false;
-
- evtType = EventType.CREATED;
- }
-
- break;
- }
-
- case EVT_CACHE_OBJECT_EXPIRED: {
- evtType = EventType.EXPIRED;
-
- break;
- }
-
- default:
- assert false : cacheEvt;
-
- throw new IgniteException("Unexpected event: " + cacheEvt);
- }
-
- if (cache == null) {
- cache = ignite.jcache(cacheName);
-
- assert cache != null : cacheName;
- }
-
- return fltr == null || fltr.evaluate(new CacheEntryEvent(cache, evtType, cacheEvt));
- }
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- out.writeBoolean(create);
-
- out.writeBoolean(update);
-
- U.writeString(out, cacheName);
-
- out.writeObject(fltrFactory);
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings("unchecked")
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- create = in.readBoolean();
-
- update = in.readBoolean();
-
- cacheName = U.readString(in);
-
- fltrFactory = (Factory<CacheEntryEventFilter>)in.readObject();
-
- if (fltrFactory != null)
- fltr = fltrFactory.create();
- }
- }
-
- /**
- *
- */
- class EventCallback implements IgniteBiPredicate<UUID, IgniteEvent> {
- /** */
- private final CacheEntryCreatedListener createLsnr;
-
- /** */
- private final CacheEntryUpdatedListener updateLsnr;
-
- /** */
- private final CacheEntryRemovedListener rmvLsnr;
-
- /** */
- private final CacheEntryExpiredListener expireLsnr;
-
- /**
- * @param lsnr Listener.
- */
- EventCallback(CacheEntryListener lsnr) {
- createLsnr = lsnr instanceof CacheEntryCreatedListener ? (CacheEntryCreatedListener)lsnr : null;
- updateLsnr = lsnr instanceof CacheEntryUpdatedListener ? (CacheEntryUpdatedListener)lsnr : null;
- rmvLsnr = lsnr instanceof CacheEntryRemovedListener ? (CacheEntryRemovedListener)lsnr : null;
- expireLsnr = lsnr instanceof CacheEntryExpiredListener ? (CacheEntryExpiredListener)lsnr : null;
- }
-
- /**
- * @return {@code True} if listens for create event.
- */
- boolean create() {
- return createLsnr != null;
- }
-
- /**
- * @return {@code True} if listens for update event.
- */
- boolean update() {
- return updateLsnr != null;
- }
+ /** {@inheritDoc} */
+ @Override public void deregisterCacheEntryListener(CacheEntryListenerConfiguration lsnrCfg) {
+ GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
- /**
- * @return {@code True} if listens for remove event.
- */
- boolean remove() {
- return rmvLsnr != null;
+ try {
+ ctx.continuousQueries().deregisterCacheEntryListener(lsnrCfg);
}
-
- /**
- * @return {@code True} if listens for expire event.
- */
- boolean expire() {
- return expireLsnr != null;
+ catch (IgniteCheckedException e) {
+ throw cacheException(e);
}
-
- /** {@inheritDoc} */
- @SuppressWarnings("unchecked")
- @Override public boolean apply(UUID uuid, IgniteEvent evt) {
- assert evt instanceof IgniteCacheEvent : evt;
-
- IgniteCacheEvent cacheEvt = (IgniteCacheEvent)evt;
-
- switch (cacheEvt.type()) {
- case EVT_CACHE_OBJECT_REMOVED: {
- assert rmvLsnr != null;
-
- CacheEntryEvent evt0 = new CacheEntryEvent(IgniteCacheProxy.this, EventType.REMOVED, cacheEvt);
-
- rmvLsnr.onRemoved(Collections.singleton(evt0));
-
- break;
- }
-
- case EVT_CACHE_OBJECT_PUT: {
- if (cacheEvt.hasOldValue()) {
- assert updateLsnr != null;
-
- CacheEntryEvent evt0 = new CacheEntryEvent(IgniteCacheProxy.this, EventType.UPDATED, cacheEvt);
-
- updateLsnr.onUpdated(Collections.singleton(evt0));
- }
- else {
- assert createLsnr != null;
-
- CacheEntryEvent evt0 = new CacheEntryEvent(IgniteCacheProxy.this, EventType.CREATED, cacheEvt);
-
- createLsnr.onCreated(Collections.singleton(evt0));
- }
-
- break;
- }
-
- case EVT_CACHE_OBJECT_EXPIRED: {
- assert expireLsnr != null;
-
- CacheEntryEvent evt0 = new CacheEntryEvent(IgniteCacheProxy.this, EventType.EXPIRED, cacheEvt);
-
- expireLsnr.onExpired(Collections.singleton(evt0));
-
- break;
- }
- }
-
- return false;
+ finally {
+ gate.leave(prev);
}
}
/** {@inheritDoc} */
- @Override public void deregisterCacheEntryListener(CacheEntryListenerConfiguration lsnrCfg) {
- }
-
- /** {@inheritDoc} */
@Override public Iterator<Cache.Entry<K, V>> iterator() {
// TODO IGNITE-1.
throw new UnsupportedOperationException();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/gridgain/grid/cache/GridCacheConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/cache/GridCacheConfiguration.java b/modules/core/src/main/java/org/gridgain/grid/cache/GridCacheConfiguration.java
index 5a9a675..149bee9 100644
--- a/modules/core/src/main/java/org/gridgain/grid/cache/GridCacheConfiguration.java
+++ b/modules/core/src/main/java/org/gridgain/grid/cache/GridCacheConfiguration.java
@@ -362,6 +362,7 @@ public class GridCacheConfiguration extends MutableConfiguration {
interceptor = cc.getInterceptor();
invalidate = cc.isInvalidate();
keepPortableInStore = cc.isKeepPortableInStore();
+ listenerConfigurations = cc.listenerConfigurations;
offHeapMaxMem = cc.getOffHeapMaxMemory();
maxConcurrentAsyncOps = cc.getMaxConcurrentAsyncOperations();
maxQryIterCnt = cc.getMaximumQueryIteratorCount();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/gridgain/grid/cache/query/GridCacheContinuousQuery.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/cache/query/GridCacheContinuousQuery.java b/modules/core/src/main/java/org/gridgain/grid/cache/query/GridCacheContinuousQuery.java
index a2675aa..60db85a 100644
--- a/modules/core/src/main/java/org/gridgain/grid/cache/query/GridCacheContinuousQuery.java
+++ b/modules/core/src/main/java/org/gridgain/grid/cache/query/GridCacheContinuousQuery.java
@@ -136,13 +136,13 @@ public interface GridCacheContinuousQuery<K, V> extends AutoCloseable {
* can get deadlocks.
*
* @param cb Local callback.
- * @deprecated Deprecated in favor of {@link #localCallback(org.apache.ignite.lang.IgniteBiPredicate)} method.
+ * @deprecated Deprecated in favor of {@link #localCallback(IgniteBiPredicate)} method.
*/
@Deprecated
public void callback(@Nullable IgniteBiPredicate<UUID, Collection<Map.Entry<K, V>>> cb);
/**
- * Gets local callback. See {@link #callback(org.apache.ignite.lang.IgniteBiPredicate)} for more information.
+ * Gets local callback. See {@link #callback(IgniteBiPredicate)} for more information.
*
* @return Local callback.
* @deprecated Deprecated in favor of {@link #localCallback()} method.
@@ -167,7 +167,7 @@ public interface GridCacheContinuousQuery<K, V> extends AutoCloseable {
public void filter(@Nullable IgniteBiPredicate<K, V> filter);
/**
- * Gets key-value filter. See {@link #filter(org.apache.ignite.lang.IgniteBiPredicate)} for more information.
+ * Gets key-value filter. See {@link #filter(IgniteBiPredicate)} for more information.
*
* @return Key-value filter.
* @deprecated Deprecated in favor of {@link #remoteFilter()} method.
@@ -197,7 +197,7 @@ public interface GridCacheContinuousQuery<K, V> extends AutoCloseable {
public void localCallback(IgniteBiPredicate<UUID, Collection<GridCacheContinuousQueryEntry<K, V>>> locCb);
/**
- * Gets local callback. See {@link #callback(org.apache.ignite.lang.IgniteBiPredicate)} for more information.
+ * Gets local callback. See {@link #callback(IgniteBiPredicate)} for more information.
*
* @return Local callback.
*/
@@ -218,7 +218,7 @@ public interface GridCacheContinuousQuery<K, V> extends AutoCloseable {
public void remoteFilter(@Nullable IgnitePredicate<GridCacheContinuousQueryEntry<K, V>> filter);
/**
- * Gets key-value filter. See {@link #filter(org.apache.ignite.lang.IgniteBiPredicate)} for more information.
+ * Gets key-value filter. See {@link #filter(IgniteBiPredicate)} for more information.
*
* @return Key-value filter.
*/
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/gridgain/grid/kernal/GridEventConsumeHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/GridEventConsumeHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/GridEventConsumeHandler.java
index 49fbb81..a48f6e1 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/GridEventConsumeHandler.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/GridEventConsumeHandler.java
@@ -31,7 +31,7 @@ import static org.apache.ignite.events.IgniteEventType.*;
/**
* Continuous routine handler for remote event listening.
*/
-public class GridEventConsumeHandler implements GridContinuousHandler {
+class GridEventConsumeHandler implements GridContinuousHandler {
/** */
private static final long serialVersionUID = 0L;
@@ -76,7 +76,7 @@ public class GridEventConsumeHandler implements GridContinuousHandler {
* @param filter Filter.
* @param types Types.
*/
- public GridEventConsumeHandler(@Nullable IgniteBiPredicate<UUID, IgniteEvent> cb, @Nullable IgnitePredicate<IgniteEvent> filter,
+ GridEventConsumeHandler(@Nullable IgniteBiPredicate<UUID, IgniteEvent> cb, @Nullable IgnitePredicate<IgniteEvent> filter,
@Nullable int[] types) {
this.cb = cb == null ? DFLT_CALLBACK : cb;
this.filter = filter;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
index bb493cc..b19e44f 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
@@ -775,9 +775,23 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
ret = old;
}
- if (evt && expired && cctx.events().isRecordable(EVT_CACHE_OBJECT_EXPIRED)) {
- cctx.events().addEvent(partition(), key, tx, owner, EVT_CACHE_OBJECT_EXPIRED, null, false, expiredVal,
- expiredVal != null || hasOldBytes, subjId, null, taskName);
+ if (evt && expired) {
+ if (cctx.events().isRecordable(EVT_CACHE_OBJECT_EXPIRED)) {
+ cctx.events().addEvent(partition(),
+ key,
+ tx,
+ owner,
+ EVT_CACHE_OBJECT_EXPIRED,
+ null,
+ false,
+ expiredVal,
+ expiredVal != null || hasOldBytes,
+ subjId,
+ null,
+ taskName);
+ }
+
+ cctx.continuousQueries().onEntryExpired(this, key, expiredVal, null);
// No more notifications.
evt = false;
@@ -1147,7 +1161,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
if (mode == GridCacheMode.LOCAL || mode == GridCacheMode.REPLICATED ||
(tx != null && tx.local() && !isNear()))
- cctx.continuousQueries().onEntryUpdate(this, key, val, valueBytesUnlocked(), old, oldBytes);
+ cctx.continuousQueries().onEntryUpdate(this, key, val, valueBytesUnlocked(), old, oldBytes, false);
cctx.dataStructures().onEntryUpdated(key, false);
}
@@ -1309,7 +1323,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
if (mode == GridCacheMode.LOCAL || mode == GridCacheMode.REPLICATED ||
(tx != null && tx.local() && !isNear()))
- cctx.continuousQueries().onEntryUpdate(this, key, null, null, old, oldBytes);
+ cctx.continuousQueries().onEntryUpdate(this, key, null, null, old, oldBytes, false);
cctx.dataStructures().onEntryUpdated(key, true);
}
@@ -1580,7 +1594,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
if (metrics)
cctx.cache().metrics0().onWrite();
- cctx.continuousQueries().onEntryUpdate(this, key, val, valueBytesUnlocked(), old, oldBytes);
+ cctx.continuousQueries().onEntryUpdate(this, key, val, valueBytesUnlocked(), old, oldBytes, false);
cctx.dataStructures().onEntryUpdated(key, op == DELETE);
@@ -2038,7 +2052,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
cctx.cache().metrics0().onWrite();
if (primary || cctx.isReplicated())
- cctx.continuousQueries().onEntryUpdate(this, key, val, valueBytesUnlocked(), old, oldBytes);
+ cctx.continuousQueries().onEntryUpdate(this, key, val, valueBytesUnlocked(), old, oldBytes, false);
cctx.dataStructures().onEntryUpdated(key, op == DELETE);
@@ -2959,8 +2973,14 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
/** {@inheritDoc} */
@SuppressWarnings({"RedundantTypeArguments"})
- @Override public boolean initialValue(V val, byte[] valBytes, GridCacheVersion ver, long ttl, long expireTime,
- boolean preload, long topVer, GridDrType drType) throws IgniteCheckedException, GridCacheEntryRemovedException {
+ @Override public boolean initialValue(V val,
+ byte[] valBytes,
+ GridCacheVersion ver,
+ long ttl,
+ long expireTime,
+ boolean preload,
+ long topVer,
+ GridDrType drType) throws IgniteCheckedException, GridCacheEntryRemovedException {
if (cctx.isUnmarshalValues() && valBytes != null && val == null && isNewLocked())
val = cctx.marshaller().<V>unmarshal(valBytes, cctx.deploy().globalLoader());
@@ -2996,8 +3016,15 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
drReplicate(drType, val, valBytes, ver);
if (!skipQryNtf) {
- if (cctx.affinity().primary(cctx.localNode(), key, topVer) || cctx.isReplicated())
- cctx.continuousQueries().onEntryUpdate(this, key, val, valueBytesUnlocked(), null, null);
+ if (cctx.affinity().primary(cctx.localNode(), key, topVer) || cctx.isReplicated()) {
+ cctx.continuousQueries().onEntryUpdate(this,
+ key,
+ val,
+ valueBytesUnlocked(),
+ null,
+ null,
+ preload);
+ }
cctx.dataStructures().onEntryUpdated(key, false);
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProcessor.java
index 8e9cea0..46755c6 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProcessor.java
@@ -43,6 +43,7 @@ import org.gridgain.grid.util.typedef.*;
import org.gridgain.grid.util.typedef.internal.*;
import org.jetbrains.annotations.*;
+import javax.cache.configuration.*;
import javax.management.*;
import java.util.*;
@@ -811,7 +812,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
cacheCtx.cache().start();
-
if (log.isInfoEnabled())
log.info("Started cache [name=" + cfg.getName() + ", mode=" + cfg.getCacheMode() + ']');
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheDataStructuresManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheDataStructuresManager.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheDataStructuresManager.java
index 35b10b7..db072d6 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheDataStructuresManager.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheDataStructuresManager.java
@@ -679,7 +679,7 @@ public final class GridCacheDataStructuresManager<K, V> extends GridCacheManager
}
});
- queueQry.execute(cctx.isLocal() || cctx.isReplicated() ? cctx.grid().forLocal() : null, true);
+ queueQry.execute(cctx.isLocal() || cctx.isReplicated() ? cctx.grid().forLocal() : null, true, false);
}
GridCacheQueueProxy queue = queuesMap.get(header.id());
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryAdapter.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryAdapter.java
index 9b88858..c00c961 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryAdapter.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryAdapter.java
@@ -12,7 +12,6 @@ package org.gridgain.grid.kernal.processors.cache.query.continuous;
import org.apache.ignite.*;
import org.apache.ignite.cluster.*;
import org.apache.ignite.lang.*;
-import org.gridgain.grid.*;
import org.gridgain.grid.cache.*;
import org.gridgain.grid.cache.query.*;
import org.gridgain.grid.cache.query.GridCacheContinuousQueryEntry;
@@ -24,6 +23,7 @@ import org.gridgain.grid.util.typedef.internal.*;
import org.gridgain.grid.util.*;
import org.jetbrains.annotations.*;
+import javax.cache.event.*;
import java.util.*;
import java.util.concurrent.locks.*;
@@ -213,12 +213,12 @@ public class GridCacheContinuousQueryAdapter<K, V> implements GridCacheContinuou
/** {@inheritDoc} */
@Override public void execute() throws IgniteCheckedException {
- execute(null, false);
+ execute(null, false, false);
}
/** {@inheritDoc} */
@Override public void execute(@Nullable ClusterGroup prj) throws IgniteCheckedException {
- execute(prj, false);
+ execute(prj, false, false);
}
/**
@@ -226,9 +226,10 @@ public class GridCacheContinuousQueryAdapter<K, V> implements GridCacheContinuou
*
* @param prj Grid projection.
* @param internal If {@code true} then query notified about internal entries updates.
+ * @param entryLsnr {@code True} if query created for {@link CacheEntryListener}.
* @throws IgniteCheckedException If failed.
*/
- public void execute(@Nullable ClusterGroup prj, boolean internal) throws IgniteCheckedException {
+ public void execute(@Nullable ClusterGroup prj, boolean internal, boolean entryLsnr) throws IgniteCheckedException {
if (locCb == null)
throw new IllegalStateException("Mandatory local callback is not set for the query: " + this);
@@ -271,12 +272,32 @@ public class GridCacheContinuousQueryAdapter<K, V> implements GridCacheContinuou
guard.block();
- GridContinuousHandler hnd = ctx.kernalContext().security().enabled() ?
- new GridCacheContinuousQueryHandlerV2<>(ctx.name(), topic, locCb, rmtFilter, prjPred, internal,
- ctx.kernalContext().job().currentTaskNameHash()) :
- new GridCacheContinuousQueryHandler<>(ctx.name(), topic, locCb, rmtFilter, prjPred, internal);
+ GridContinuousHandler hnd;
+
+ if (ctx.kernalContext().security().enabled()) {
+ hnd = new GridCacheContinuousQueryHandlerV2<>(ctx.name(),
+ topic,
+ locCb,
+ rmtFilter,
+ prjPred,
+ internal,
+ entryLsnr,
+ ctx.kernalContext().job().currentTaskNameHash());
+ }
+ else {
+ hnd = new GridCacheContinuousQueryHandler<>(ctx.name(),
+ topic,
+ locCb,
+ rmtFilter,
+ prjPred,
+ internal,
+ entryLsnr);
+ }
- routineId = ctx.kernalContext().continuous().startRoutine(hnd, bufSize, timeInterval, autoUnsubscribe,
+ routineId = ctx.kernalContext().continuous().startRoutine(hnd,
+ bufSize,
+ timeInterval,
+ autoUnsubscribe,
prj.predicate()).get();
}
finally {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryEntry.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryEntry.java
index 991573b..3c5265c 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryEntry.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryEntry.java
@@ -77,6 +77,9 @@ public class GridCacheContinuousQueryEntry<K, V> implements GridCacheEntry<K, V>
@GridToStringExclude
private GridDeploymentInfo depInfo;
+ /** */
+ private boolean expired;
+
/**
* Required by {@link Externalizable}.
*/
@@ -85,7 +88,7 @@ public class GridCacheContinuousQueryEntry<K, V> implements GridCacheEntry<K, V>
impl = null;
}
- /**
+ /*
* @param ctx Cache context.
* @param impl Cache entry.
* @param key Key.
@@ -93,9 +96,16 @@ public class GridCacheContinuousQueryEntry<K, V> implements GridCacheEntry<K, V>
* @param newValBytes Value bytes.
* @param oldVal Old value.
* @param oldValBytes Old value bytes.
+ * @param expired {@code True} if created for expired entry.
*/
- GridCacheContinuousQueryEntry(GridCacheContext<K, V> ctx, GridCacheEntry<K, V> impl, K key, @Nullable V newVal,
- @Nullable GridCacheValueBytes newValBytes, @Nullable V oldVal, @Nullable GridCacheValueBytes oldValBytes) {
+ GridCacheContinuousQueryEntry(GridCacheContext<K, V> ctx,
+ GridCacheEntry<K, V> impl,
+ K key,
+ @Nullable V newVal,
+ @Nullable GridCacheValueBytes newValBytes,
+ @Nullable V oldVal,
+ @Nullable GridCacheValueBytes oldValBytes,
+ boolean expired) {
assert ctx != null;
assert impl != null;
assert key != null;
@@ -107,6 +117,14 @@ public class GridCacheContinuousQueryEntry<K, V> implements GridCacheEntry<K, V>
this.newValBytes = newValBytes;
this.oldVal = oldVal;
this.oldValBytes = oldValBytes;
+ this.expired = expired;
+ }
+
+ /**
+ * @return {@code True} if entry expired.
+ */
+ public boolean expired() {
+ return expired;
}
/**
@@ -710,6 +728,8 @@ public class GridCacheContinuousQueryEntry<K, V> implements GridCacheEntry<K, V>
out.writeObject(newVal);
out.writeObject(oldVal);
}
+
+ out.writeBoolean(expired);
}
/** {@inheritDoc} */
@@ -734,6 +754,8 @@ public class GridCacheContinuousQueryEntry<K, V> implements GridCacheEntry<K, V>
newVal = (V)in.readObject();
oldVal = (V)in.readObject();
}
+
+ expired = in.readBoolean();
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java
index 9d810c0..e61b2a2 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java
@@ -24,6 +24,7 @@ import org.gridgain.grid.util.typedef.*;
import org.gridgain.grid.util.typedef.internal.*;
import org.jetbrains.annotations.*;
+import javax.cache.event.*;
import java.io.*;
import java.util.*;
@@ -61,6 +62,9 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
/** Internal flag. */
private boolean internal;
+ /** Entry listener flag. */
+ private boolean entryLsnr;
+
/**
* Required by {@link Externalizable}.
*/
@@ -75,11 +79,15 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
* @param filter Filter.
* @param prjPred Projection predicate.
* @param internal If {@code true} then query is notified about internal entries updates.
+ * @param entryLsnr {@code True} if query created for {@link CacheEntryListener}.
*/
- GridCacheContinuousQueryHandler(@Nullable String cacheName, Object topic,
+ GridCacheContinuousQueryHandler(@Nullable String cacheName,
+ Object topic,
IgniteBiPredicate<UUID, Collection<org.gridgain.grid.cache.query.GridCacheContinuousQueryEntry<K, V>>> cb,
@Nullable IgnitePredicate<org.gridgain.grid.cache.query.GridCacheContinuousQueryEntry<K, V>> filter,
- @Nullable IgnitePredicate<GridCacheEntry<K, V>> prjPred, boolean internal) {
+ @Nullable IgnitePredicate<GridCacheEntry<K, V>> prjPred,
+ boolean internal,
+ boolean entryLsnr) {
assert topic != null;
assert cb != null;
@@ -89,6 +97,7 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
this.filter = filter;
this.prjPred = prjPred;
this.internal = internal;
+ this.entryLsnr = entryLsnr;
}
/** {@inheritDoc} */
@@ -183,7 +192,7 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
}
}
- if (recordEvt) {
+ if (!entryLsnr && recordEvt) {
ctx.event().record(new IgniteCacheQueryReadEvent<>(
ctx.discovery().localNode(),
"Continuous query executed.",
@@ -241,12 +250,12 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
}
};
- return manager(ctx).registerListener(nodeId, routineId, lsnr, internal);
+ return manager(ctx).registerListener(routineId, lsnr, internal, entryLsnr);
}
/** {@inheritDoc} */
@Override public void onListenerRegistered(UUID routineId, GridKernalContext ctx) {
- manager(ctx).iterate(internal, routineId);
+ manager(ctx).iterate(internal, routineId, entryLsnr);
}
/** {@inheritDoc} */
@@ -371,6 +380,8 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
out.writeObject(prjPred);
out.writeBoolean(internal);
+
+ out.writeBoolean(entryLsnr);
}
/** {@inheritDoc} */
@@ -394,6 +405,8 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
prjPred = (IgnitePredicate<GridCacheEntry<K, V>>)in.readObject();
internal = in.readBoolean();
+
+ entryLsnr = in.readBoolean();
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandlerV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandlerV2.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandlerV2.java
index 6a185fd..63209ec 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandlerV2.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandlerV2.java
@@ -14,6 +14,7 @@ import org.gridgain.grid.cache.*;
import org.gridgain.grid.cache.query.GridCacheContinuousQueryEntry;
import org.jetbrains.annotations.*;
+import javax.cache.event.*;
import java.io.*;
import java.util.*;
@@ -41,13 +42,18 @@ public class GridCacheContinuousQueryHandlerV2<K, V> extends GridCacheContinuous
* @param filter Filter.
* @param prjPred Projection predicate.
* @param internal If {@code true} then query is notified about internal entries updates.
+ * @param entryLsnr {@code True} if query created for {@link CacheEntryListener}.
* @param taskHash Task hash.
*/
- public GridCacheContinuousQueryHandlerV2(@Nullable String cacheName, Object topic,
+ public GridCacheContinuousQueryHandlerV2(@Nullable String cacheName,
+ Object topic,
IgniteBiPredicate<UUID, Collection<GridCacheContinuousQueryEntry<K, V>>> cb,
@Nullable IgnitePredicate<GridCacheContinuousQueryEntry<K, V>> filter,
- @Nullable IgnitePredicate<GridCacheEntry<K, V>> prjPred, boolean internal, int taskHash) {
- super(cacheName, topic, cb, filter, prjPred, internal);
+ @Nullable IgnitePredicate<GridCacheEntry<K, V>> prjPred,
+ boolean internal,
+ boolean entryLsnr,
+ int taskHash) {
+ super(cacheName, topic, cb, filter, prjPred, internal, entryLsnr);
this.taskHash = taskHash;
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java
index b67d93c..8c1b70e 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java
@@ -11,18 +11,25 @@ package org.gridgain.grid.kernal.processors.cache.query.continuous;
import org.apache.ignite.*;
import org.apache.ignite.cluster.*;
+import org.apache.ignite.internal.processors.cache.*;
import org.apache.ignite.lang.*;
+import org.apache.ignite.resources.*;
import org.gridgain.grid.cache.*;
import org.gridgain.grid.cache.query.*;
import org.gridgain.grid.kernal.processors.cache.*;
import org.gridgain.grid.util.typedef.*;
+import org.gridgain.grid.util.typedef.internal.*;
import org.jdk8.backport.*;
import org.jetbrains.annotations.*;
+import javax.cache.configuration.*;
+import javax.cache.event.*;
+import java.io.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
+import static javax.cache.event.EventType.*;
import static org.apache.ignite.events.IgniteEventType.*;
import static org.gridgain.grid.kernal.GridTopic.*;
@@ -48,6 +55,10 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
/** Query sequence number for message topic. */
private final AtomicLong seq = new AtomicLong();
+ /** Continues queries created for cache event listeners. */
+ private final ConcurrentMap<CacheEntryListenerConfiguration, GridCacheContinuousQuery<K, V>> lsnrQrys =
+ new ConcurrentHashMap8<>();
+
/** {@inheritDoc} */
@Override protected void start0() throws IgniteCheckedException {
// Append cache name to the topic.
@@ -55,6 +66,7 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
}
/** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
@Override protected void onKernalStart0() throws IgniteCheckedException {
if (intLsnrCnt.get() > 0 || lsnrCnt.get() > 0) {
Collection<ClusterNode> nodes = cctx.discovery().cacheNodes(cctx.name(), -1);
@@ -65,6 +77,30 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
"for versions below 6.2.0");
}
}
+
+ Iterable<CacheEntryListenerConfiguration<K, V>> lsnrCfgs = cctx.config().getCacheEntryListenerConfigurations();
+
+ if (lsnrCfgs != null) {
+ IgniteCacheProxy<K, V> cache = cctx.kernalContext().cache().jcache(cctx.name());
+
+ for (CacheEntryListenerConfiguration<K, V> cfg : lsnrCfgs)
+ cache.registerCacheEntryListener(cfg);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void onKernalStop0(boolean cancel) {
+ super.onKernalStop0(cancel);
+
+ for (CacheEntryListenerConfiguration lsnrCfg : lsnrQrys.keySet()) {
+ try {
+ deregisterCacheEntryListener(lsnrCfg);
+ }
+ catch (IgniteCheckedException e) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to remove cache entry listener: " + e);
+ }
+ }
}
/**
@@ -84,10 +120,16 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
* @param newBytes New value bytes.
* @param oldVal Old value.
* @param oldBytes Old value bytes.
+ * @param preload {@code True} if entry is updated during preloading.
* @throws IgniteCheckedException In case of error.
*/
- public void onEntryUpdate(GridCacheEntryEx<K, V> e, K key, @Nullable V newVal,
- @Nullable GridCacheValueBytes newBytes, V oldVal, @Nullable GridCacheValueBytes oldBytes) throws IgniteCheckedException {
+ public void onEntryUpdate(GridCacheEntryEx<K, V> e,
+ K key,
+ @Nullable V newVal,
+ @Nullable GridCacheValueBytes newBytes,
+ V oldVal,
+ @Nullable GridCacheValueBytes oldBytes,
+ boolean preload) throws IgniteCheckedException {
assert e != null;
assert key != null;
@@ -104,25 +146,145 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
oldVal = cctx.unwrapTemporary(oldVal);
GridCacheContinuousQueryEntry<K, V> e0 = new GridCacheContinuousQueryEntry<>(
- cctx, e.wrap(false), key, newVal, newBytes, oldVal, oldBytes);
+ cctx,
+ e.wrap(false),
+ key,
+ newVal,
+ newBytes,
+ oldVal,
+ oldBytes,
+ false);
e0.initValue(cctx.marshaller(), cctx.deploy().globalLoader());
boolean recordEvt = !e.isInternal() && cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ);
- for (ListenerInfo<K, V> lsnr : lsnrCol.values())
+ for (ListenerInfo<K, V> lsnr : lsnrCol.values()) {
+ if (preload && lsnr.entryListener())
+ continue;
+
lsnr.onEntryUpdate(e0, recordEvt);
+ }
+ }
+
+ /**
+ * @param e Entry.
+ * @param key Key.
+ * @param oldVal Old value.
+ * @param oldBytes Old value bytes.
+ */
+ public void onEntryExpired(GridCacheEntryEx<K, V> e,
+ K key,
+ V oldVal,
+ @Nullable GridCacheValueBytes oldBytes) {
+ if (e.isInternal())
+ return;
+
+ ConcurrentMap<UUID, ListenerInfo<K, V>> lsnrCol = lsnrs;
+
+ if (F.isEmpty(lsnrCol))
+ return;
+
+ GridCacheContinuousQueryEntry<K, V> e0 = new GridCacheContinuousQueryEntry<>(
+ cctx,
+ e.wrap(false),
+ key,
+ null,
+ null,
+ oldVal,
+ oldBytes,
+ true);
+
+ for (ListenerInfo<K, V> lsnr : lsnrCol.values()) {
+ if (!lsnr.entryListener())
+ continue;
+
+ lsnr.onEntryUpdate(e0, false);
+ }
+ }
+
+ /**
+ * @param lsnrCfg Listener configuration.
+ * @throws IgniteCheckedException If failed.
+ */
+ public void registerCacheEntryListener(CacheEntryListenerConfiguration<K, V> lsnrCfg)
+ throws IgniteCheckedException {
+ GridCacheContinuousQueryAdapter<K, V> qry = null;
+
+ try {
+ A.notNull(lsnrCfg, "lsnrCfg");
+
+ Factory<CacheEntryListener<? super K, ? super V>> factory = lsnrCfg.getCacheEntryListenerFactory();
+
+ A.notNull(factory, "cacheEntryListenerFactory");
+
+ CacheEntryListener lsnr = factory.create();
+
+ A.notNull(lsnr, "lsnr");
+
+ IgniteCacheProxy<K, V> cache= cctx.kernalContext().cache().jcache(cctx.name());
+
+ EntryListenerCallback cb = new EntryListenerCallback(cache, lsnr);
+
+ if (!(cb.create() || cb.update() || cb.remove() || cb.expire()))
+ throw new IllegalArgumentException("Listener must implement one of CacheEntryListener sub-interfaces.");
+
+ qry = (GridCacheContinuousQueryAdapter<K, V>)cctx.cache().queries().createContinuousQuery();
+
+ GridCacheContinuousQuery<K, V> old = lsnrQrys.putIfAbsent(lsnrCfg, qry);
+
+ if (old != null)
+ throw new IllegalArgumentException("Listener is already registered for configuration: " + lsnrCfg);
+
+ qry.autoUnsubscribe(true);
+
+ qry.bufferSize(1);
+
+ qry.localCallback(cb);
+
+ EntryListenerFilter<K, V> fltr = new EntryListenerFilter<>(cb.create(),
+ cb.update(),
+ cb.remove(),
+ cb.expire(),
+ lsnrCfg.getCacheEntryEventFilterFactory(),
+ cctx.kernalContext().grid(),
+ cctx.name());
+
+ qry.remoteFilter(fltr);
+
+ qry.execute(null, false, true);
+ }
+ catch (IgniteCheckedException e) {
+ lsnrQrys.remove(lsnrCfg, qry); // Remove query if failed to execute it.
+
+ throw e;
+ }
+ }
+
+ /**
+ * @param lsnrCfg Listener configuration.
+ * @throws IgniteCheckedException If failed.
+ */
+ public void deregisterCacheEntryListener(CacheEntryListenerConfiguration lsnrCfg) throws IgniteCheckedException {
+ A.notNull(lsnrCfg, "lsnrCfg");
+
+ GridCacheContinuousQuery<K, V> qry = lsnrQrys.remove(lsnrCfg);
+
+ if (qry != null)
+ qry.close();
}
/**
- * @param nodeId Node ID.
* @param lsnrId Listener ID.
* @param lsnr Listener.
* @param internal Internal flag.
+ * @param entryLsnr {@code True} if query created for {@link CacheEntryListener}.
* @return Whether listener was actually registered.
*/
- boolean registerListener(UUID nodeId, UUID lsnrId, GridCacheContinuousQueryListener<K, V> lsnr, boolean internal) {
- ListenerInfo<K, V> info = new ListenerInfo<>(lsnr);
+ boolean registerListener(UUID lsnrId, GridCacheContinuousQueryListener<K, V> lsnr,
+ boolean internal,
+ boolean entryLsnr) {
+ ListenerInfo<K, V> info = new ListenerInfo<>(lsnr, entryLsnr);
boolean added;
@@ -164,25 +326,36 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
* Iterates through existing data.
*
* @param internal Internal flag.
+ * @param entryLsnr {@code True} if query created for {@link CacheEntryListener}.
* @param id Listener ID.
*/
- void iterate(boolean internal, UUID id) {
+ void iterate(boolean internal, UUID id, boolean entryLsnr) {
ListenerInfo<K, V> info = internal ? intLsnrs.get(id) : lsnrs.get(id);
assert info != null;
- Set<GridCacheEntry<K, V>> entries;
-
- if (cctx.isReplicated())
- entries = internal ? cctx.cache().entrySetx() :
- cctx.cache().entrySet();
- else
- entries = internal ? cctx.cache().primaryEntrySetx() :
- cctx.cache().primaryEntrySet();
-
- for (GridCacheEntry<K, V> e : entries) {
- info.onIterate(new GridCacheContinuousQueryEntry<>(cctx, e, e.getKey(), e.getValue(), null, null, null),
- !internal && cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ));
+ if (!entryLsnr) {
+ Set<GridCacheEntry<K, V>> entries;
+
+ if (cctx.isReplicated())
+ entries = internal ? cctx.cache().entrySetx() :
+ cctx.cache().entrySet();
+ else
+ entries = internal ? cctx.cache().primaryEntrySetx() :
+ cctx.cache().primaryEntrySet();
+
+ for (GridCacheEntry<K, V> e : entries) {
+ GridCacheContinuousQueryEntry<K, V> qryEntry = new GridCacheContinuousQueryEntry<>(cctx,
+ e,
+ e.getKey(),
+ e.getValue(),
+ null,
+ null,
+ null,
+ false);
+
+ info.onIterate(qryEntry, !internal && cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ));
+ }
}
info.flushPending();
@@ -198,11 +371,16 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
/** Pending entries. */
private Collection<PendingEntry<K, V>> pending = new LinkedList<>();
+ /** */
+ private boolean entryLsnr;
+
/**
* @param lsnr Listener.
+ * @param entryLsnr {@code True} if listener created for {@link CacheEntryListener}.
*/
- private ListenerInfo(GridCacheContinuousQueryListener<K, V> lsnr) {
+ private ListenerInfo(GridCacheContinuousQueryListener<K, V> lsnr, boolean entryLsnr) {
this.lsnr = lsnr;
+ this.entryLsnr = entryLsnr;
}
/**
@@ -247,6 +425,13 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
for (PendingEntry<K, V> e : pending0)
lsnr.onEntryUpdate(e.entry, e.recordEvt);
}
+
+ /**
+ * @return {@code True} if listener created for {@link CacheEntryListener}.
+ */
+ boolean entryListener() {
+ return entryLsnr;
+ }
}
/**
@@ -268,4 +453,275 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
this.recordEvt = recordEvt;
}
}
+
+ /**
+ *
+ */
+ static class EntryListenerFilter<K1, V1> implements
+ IgnitePredicate<org.gridgain.grid.cache.query.GridCacheContinuousQueryEntry<K1, V1>>, Externalizable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private boolean create;
+
+ /** */
+ private boolean update;
+
+ /** */
+ private boolean rmv;
+
+ /** */
+ private boolean expire;
+
+ /** */
+ private Factory<CacheEntryEventFilter<? super K1, ? super V1>> fltrFactory;
+
+ /** */
+ private CacheEntryEventFilter fltr;
+
+ /** */
+ @IgniteInstanceResource
+ private Ignite ignite;
+
+ /** */
+ private IgniteCache cache;
+
+ /** */
+ private String cacheName;
+
+ /**
+ *
+ */
+ public EntryListenerFilter() {
+ // No-op.
+ }
+
+ /**
+ * @param create {@code True} if listens for create events.
+ * @param update {@code True} if listens for create events.
+ * @param rmv {@code True} if listens for remove events.
+ * @param expire {@code True} if listens for expire events.
+ * @param fltrFactory Filter factory.
+ * @param ignite Ignite instance.
+ * @param cacheName Cache name.
+ */
+ EntryListenerFilter(
+ boolean create,
+ boolean update,
+ boolean rmv,
+ boolean expire,
+ Factory<CacheEntryEventFilter<? super K1, ? super V1>> fltrFactory,
+ Ignite ignite,
+ @Nullable String cacheName) {
+ this.create = create;
+ this.update = update;
+ this.rmv = rmv;
+ this.expire = expire;
+ this.fltrFactory = fltrFactory;
+ this.ignite = ignite;
+ this.cacheName = cacheName;
+
+ if (fltrFactory != null)
+ fltr = fltrFactory.create();
+
+ cache = ignite.jcache(cacheName);
+
+ assert cache != null : cacheName;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public boolean apply(org.gridgain.grid.cache.query.GridCacheContinuousQueryEntry<K1, V1> entry) {
+ try {
+ EventType evtType;
+
+ if (entry.getValue() == null) {
+ if (((GridCacheContinuousQueryEntry)entry).expired()) { // Expire.
+ if (!expire)
+ return false;
+
+ evtType = EXPIRED;
+ }
+ else { // Remove.
+ if (!rmv)
+ return false;
+
+ evtType = REMOVED;
+ }
+ }
+ else {
+ if (entry.getOldValue() != null) { // Update.
+ if (!update)
+ return false;
+
+ evtType = UPDATED;
+ }
+ else { // Create.
+ if (!create)
+ return false;
+
+ evtType = CREATED;
+ }
+ }
+
+ if (cache == null) {
+ cache = ignite.jcache(cacheName);
+
+ assert cache != null : cacheName;
+ }
+
+ return fltr == null || fltr.evaluate(new org.apache.ignite.cache.CacheEntryEvent(cache, evtType, entry));
+ }
+ catch (CacheEntryListenerException e) {
+ LT.warn(ignite.log(), e, "Cache entry event filter error: " + e);
+
+ return false;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeBoolean(create);
+
+ out.writeBoolean(update);
+
+ out.writeBoolean(rmv);
+
+ out.writeBoolean(expire);
+
+ U.writeString(out, cacheName);
+
+ out.writeObject(fltrFactory);
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ create = in.readBoolean();
+
+ update = in.readBoolean();
+
+ rmv = in.readBoolean();
+
+ expire = in.readBoolean();
+
+ cacheName = U.readString(in);
+
+ fltrFactory = (Factory<CacheEntryEventFilter<? super K1, ? super V1>>)in.readObject();
+
+ if (fltrFactory != null)
+ fltr = fltrFactory.create();
+ }
+ }
+
+ /**
+ *
+ */
+ private class EntryListenerCallback implements
+ IgniteBiPredicate<UUID, Collection<org.gridgain.grid.cache.query.GridCacheContinuousQueryEntry<K, V>>> {
+ /** */
+ private final IgniteCacheProxy<K, V> cache;
+
+ /** */
+ private final CacheEntryCreatedListener createLsnr;
+
+ /** */
+ private final CacheEntryUpdatedListener updateLsnr;
+
+ /** */
+ private final CacheEntryRemovedListener rmvLsnr;
+
+ /** */
+ private final CacheEntryExpiredListener expireLsnr;
+
+ /**
+ * @param cache Cache to be used as event source.
+ * @param lsnr Listener.
+ */
+ EntryListenerCallback(IgniteCacheProxy<K, V> cache, CacheEntryListener lsnr) {
+ this.cache = cache;
+
+ createLsnr = lsnr instanceof CacheEntryCreatedListener ? (CacheEntryCreatedListener)lsnr : null;
+ updateLsnr = lsnr instanceof CacheEntryUpdatedListener ? (CacheEntryUpdatedListener)lsnr : null;
+ rmvLsnr = lsnr instanceof CacheEntryRemovedListener ? (CacheEntryRemovedListener)lsnr : null;
+ expireLsnr = lsnr instanceof CacheEntryExpiredListener ? (CacheEntryExpiredListener)lsnr : null;
+ }
+
+ /**
+ * @return {@code True} if listens for create event.
+ */
+ boolean create() {
+ return createLsnr != null;
+ }
+
+ /**
+ * @return {@code True} if listens for update event.
+ */
+ boolean update() {
+ return updateLsnr != null;
+ }
+
+ /**
+ * @return {@code True} if listens for remove event.
+ */
+ boolean remove() {
+ return rmvLsnr != null;
+ }
+
+ /**
+ * @return {@code True} if listens for expire event.
+ */
+ boolean expire() {
+ return expireLsnr != null;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public boolean apply(UUID uuid, Collection<org.gridgain.grid.cache.query.GridCacheContinuousQueryEntry<K, V>> entries) {
+ for (org.gridgain.grid.cache.query.GridCacheContinuousQueryEntry entry : entries) {
+ try {
+ if (entry.getValue() == null) { // Remove.
+ if (((GridCacheContinuousQueryEntry)entry).expired()) { // Expire.
+ assert expireLsnr != null;
+
+ org.apache.ignite.cache.CacheEntryEvent evt0 =
+ new org.apache.ignite.cache.CacheEntryEvent(cache, EXPIRED, entry);
+
+ expireLsnr.onExpired(Collections.singleton(evt0));
+ }
+ else {
+ assert rmvLsnr != null;
+
+ org.apache.ignite.cache.CacheEntryEvent evt0 =
+ new org.apache.ignite.cache.CacheEntryEvent(cache, REMOVED, entry);
+
+ rmvLsnr.onRemoved(Collections.singleton(evt0));
+ }
+ }
+ else if (entry.getOldValue() != null) { // Update.
+ assert updateLsnr != null;
+
+ org.apache.ignite.cache.CacheEntryEvent evt0 =
+ new org.apache.ignite.cache.CacheEntryEvent(cache, UPDATED, entry);
+
+ updateLsnr.onUpdated(Collections.singleton(evt0));
+ }
+ else { // Create.
+ assert createLsnr != null;
+
+ org.apache.ignite.cache.CacheEntryEvent evt0 =
+ new org.apache.ignite.cache.CacheEntryEvent(cache, CREATED, entry);
+
+ createLsnr.onCreated(Collections.singleton(evt0));
+ }
+ }
+ catch (CacheEntryListenerException e) {
+ LT.warn(log, e, "Cache entry listener error: " + e);
+ }
+ }
+
+ return true;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java
index b8b5998..d8dfc97 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java
@@ -124,13 +124,13 @@ public class GridServiceProcessor extends GridProcessorAdapter {
cfgQry.localCallback(new DeploymentListener());
- cfgQry.execute(ctx.grid().forLocal(), true);
+ cfgQry.execute(ctx.grid().forLocal(), true, false);
assignQry = (GridCacheContinuousQueryAdapter<Object, Object>)cache.queries().createContinuousQuery();
assignQry.localCallback(new AssignmentListener());
- assignQry.execute(ctx.grid().forLocal(), true);
+ assignQry.execute(ctx.grid().forLocal(), true, false);
}
finally {
if (ctx.deploy().enabled())
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java
index 949b5c6..6e3221a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java
@@ -10,88 +10,197 @@
package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.*;
-import org.apache.ignite.configuration.*;
-import org.apache.ignite.events.*;
import org.gridgain.grid.cache.*;
+import org.gridgain.grid.util.typedef.internal.*;
+import org.gridgain.testframework.*;
+import org.jetbrains.annotations.*;
import javax.cache.configuration.*;
import javax.cache.event.*;
-import static org.gridgain.grid.cache.GridCacheAtomicWriteOrderMode.*;
-import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
-import static org.gridgain.grid.cache.GridCacheDistributionMode.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+import static javax.cache.event.EventType.*;
import static org.gridgain.grid.cache.GridCacheMode.*;
/**
*
*/
-public class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAbstractTest {
- @Override protected int gridCount() {
- return 3;
- }
+public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAbstractTest {
+ /** */
+ private static volatile List<CacheEntryEvent<? extends Integer, ? extends Integer>> evts;
- @Override protected GridCacheMode cacheMode() {
- return PARTITIONED;
- }
+ /** */
+ private static volatile CountDownLatch evtsLatch;
- @Override protected GridCacheAtomicityMode atomicityMode() {
- return ATOMIC;
- }
+ /** */
+ private Integer lastKey = 0;
- @Override protected GridCacheDistributionMode distributionMode() {
- return PARTITIONED_ONLY;
- }
+ /** */
+ private CacheEntryListenerConfiguration lsnrCfg;
- @Override protected GridCacheAtomicWriteOrderMode atomicWriteOrderMode() {
- return PRIMARY;
- }
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override protected GridCacheConfiguration cacheConfiguration(String gridName) throws Exception {
+ GridCacheConfiguration cfg = super.cacheConfiguration(gridName);
- @Override
- protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
- IgniteConfiguration cfg = super.getConfiguration(gridName);
-
- cfg.setIncludeEventTypes(IgniteEventType.EVT_CACHE_OBJECT_PUT);
+ if (lsnrCfg != null)
+ cfg.addCacheEntryListenerConfiguration(lsnrCfg);
return cfg;
}
- @Override
- protected GridCacheConfiguration cacheConfiguration(String gridName) throws Exception {
- GridCacheConfiguration ccfg = super.cacheConfiguration(gridName);
+ /**
+ * @throws Exception If failed.
+ */
+ public void testEvents() throws Exception {
+ CacheEntryCreatedListener<Integer, Integer> createLsnr = new CacheEntryCreatedListener<Integer, Integer>() {
+ @Override public void onCreated(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
+ for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts)
+ onEvent(evt);
+ }
+ };
+
+ CacheEntryUpdatedListener<Integer, Integer> updateLsnr = new CacheEntryUpdatedListener<Integer, Integer>() {
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
+ for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts)
+ onEvent(evt);
+ }
+ };
+
+ CacheEntryRemovedListener<Integer, Integer> rmvLsnr = new CacheEntryRemovedListener<Integer, Integer>() {
+ @Override public void onRemoved(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
+ for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts)
+ onEvent(evt);
+ }
+ };
+
+ IgniteCache<Integer, Integer> cache = jcache();
+
+ Map<Integer, Integer> vals = new HashMap<>();
+
+ for (int i = 0; i < 100; i++)
+ vals.put(i + 1_000_000, i);
+
+ cache.putAll(vals); // Put some data in cache to make sure events are not generated for existing entries.
+
+ for (Integer key : keys()) {
+ log.info("Check create event [key=" + key + ']');
+
+ checkEvents(cache, createLsnr, key, true, false, false);
+
+ log.info("Check update event [key=" + key + ']');
+
+ checkEvents(cache, updateLsnr, key, false, true, false);
+
+ log.info("Check remove event [key=" + key + ']');
+
+ checkEvents(cache, rmvLsnr, key, false, false, true);
+
+ log.info("Check create/update events [key=" + key + ']');
+
+ checkEvents(cache, new CreateUpdateListener(), key, true, true, false);
+
+ log.info("Check create/update/remove events [key=" + key + ']');
+
+ checkEvents(cache, new CreateUpdateRemoveListener(), key, true, true, true);
+ }
+
+ CacheEntryListenerConfiguration<Integer, Integer> lsnrCfg = new MutableCacheEntryListenerConfiguration<>(
+ new Factory<CacheEntryListener<Integer, Integer>>() {
+ @Override public CacheEntryListener<Integer, Integer> create() {
+ return new CreateUpdateRemoveListener();
+ }
+ },
+ new TestFilterFactory(),
+ true,
+ false
+ );
+
+ cache.registerCacheEntryListener(lsnrCfg);
+
+ log.info("Check filter.");
+
+ checkFilter(cache, vals);
+
+ cache.deregisterCacheEntryListener(lsnrCfg);
- //ccfg.setBackups(1);
+ cache.putAll(vals);
- return ccfg;
+ checkListenerOnStart(vals);
}
- public void testEvent() throws Exception {
- Ignite ignite = ignite(0);
+ /**
+ * @param vals Values in cache.
+ * @throws Exception If failed.
+ */
+ @SuppressWarnings("unchecked")
+ private void checkListenerOnStart(Map<Integer, Integer> vals) throws Exception {
+ lsnrCfg = new MutableCacheEntryListenerConfiguration<>(
+ new Factory<CacheEntryListener<Integer, Integer>>() {
+ @Override public CacheEntryListener<Integer, Integer> create() {
+ return new CreateUpdateRemoveListener();
+ }
+ },
+ null,
+ true,
+ false
+ );
- /*
- ignite.events().remoteListen(new IgniteBiPredicate<UUID, IgniteEvent>() {
- @Override public boolean apply(UUID uuid, IgniteEvent e) {
- IgniteCacheEvent evt0 = (IgniteCacheEvent)e;
+ Ignite grid = startGrid(gridCount());
- System.out.println("Event: " + uuid + " " + evt0.eventNode() + " " + e);
+ IgniteCache<Integer, Integer> cache = grid.jcache(null);
- return false;
- }
- }, null, IgniteEventType.EVT_CACHE_OBJECT_PUT);
- */
+ Integer key = Integer.MAX_VALUE;
- IgniteCache<Integer, Integer> cache = jcache();
+ log.info("Check create/update/remove events for listener in configuration [key=" + key + ']');
+
+ checkEvents(cache, lsnrCfg, key, true, true, true);
- final CacheEntryCreatedListener<Integer, Integer> lsnr = new CacheEntryCreatedListener<Integer, Integer>() {
- @Override public void onCreated(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) throws CacheEntryListenerException {
- for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts) {
- System.out.println("Event: " + evt.getEventType() + " " + evt.getKey() + " " + evt.getOldValue() + " " + evt.getValue());
+ stopGrid(gridCount());
+
+ lsnrCfg = new MutableCacheEntryListenerConfiguration<>(
+ new Factory<CacheEntryListener<Integer, Integer>>() {
+ @Override public CacheEntryListener<Integer, Integer> create() {
+ return new CreateUpdateRemoveListener();
}
- }
- };
+ },
+ new TestFilterFactory(),
+ true,
+ false
+ );
- CacheEntryListenerConfiguration<Integer, Integer> lsnrCfg = new MutableCacheEntryListenerConfiguration(
- new Factory<CacheEntryListener>() {
- @Override public CacheEntryListener create() {
+ grid = startGrid(gridCount());
+
+ cache = grid.jcache(null);
+
+ log.info("Check filter for listener in configuration.");
+
+ checkFilter(cache, vals);
+
+ stopGrid(gridCount());
+ }
+
+ /**
+ * @param cache Cache.
+ * @param lsnr Listener.
+ * @param key Key.
+ * @param create {@code True} if listens for create events.
+ * @param update {@code True} if listens for update events.
+ * @param rmv {@code True} if listens for remove events.
+ * @throws Exception If failed.
+ */
+ private void checkEvents(
+ final IgniteCache<Integer, Integer> cache,
+ final CacheEntryListener<Integer, Integer> lsnr,
+ Integer key,
+ boolean create,
+ boolean update,
+ boolean rmv) throws Exception {
+ CacheEntryListenerConfiguration<Integer, Integer> lsnrCfg = new MutableCacheEntryListenerConfiguration<>(
+ new Factory<CacheEntryListener<Integer, Integer>>() {
+ @Override public CacheEntryListener<Integer, Integer> create() {
return lsnr;
}
},
@@ -102,9 +211,325 @@ public class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAbstractTes
cache.registerCacheEntryListener(lsnrCfg);
- ignite(1).cache(null).put(1, 1);
- ignite(0).cache(null).put(1, 2);
+ checkEvents(cache, lsnrCfg, key, create, update, rmv);
+ }
+
+ /**
+ * @param cache Cache.
+ * @param vals Values in cache.
+ * @throws Exception If failed.
+ */
+ private void checkFilter(IgniteCache<Integer, Integer> cache, Map<Integer, Integer> vals) throws Exception {
+ evts = new ArrayList<>();
+
+ final int expEvts = (vals.size() / 2) * 3; // Remove, create and update for half of modified entries.
+
+ evtsLatch = new CountDownLatch(expEvts);
+
+ cache.removeAll(vals.keySet());
+
+ cache.putAll(vals);
+
+ Map<Integer, Integer> newVals = new HashMap<>();
+
+ for (Integer key : vals.keySet())
+ newVals.put(key, -1);
+
+ cache.putAll(newVals);
+
+ evtsLatch.await(5000, TimeUnit.MILLISECONDS);
+
+ assertEquals(expEvts, evts.size());
+
+ Iterator<CacheEntryEvent<? extends Integer, ? extends Integer>> iter = evts.iterator();
+
+ for (Integer key : vals.keySet()) {
+ if (key % 2 == 0) {
+ CacheEntryEvent<? extends Integer, ? extends Integer> evt = iter.next();
+
+ assertTrue(evt.getKey() % 2 == 0);
+ assertTrue(vals.keySet().contains(evt.getKey()));
+ assertEquals(REMOVED, evt.getEventType());
+ assertNull(evt.getValue());
+ assertEquals(vals.get(evt.getKey()), evt.getOldValue());
+
+ iter.remove();
+ }
+ }
+
+ for (Integer key : vals.keySet()) {
+ if (key % 2 == 0) {
+ CacheEntryEvent<? extends Integer, ? extends Integer> evt = iter.next();
+
+ assertTrue(evt.getKey() % 2 == 0);
+ assertTrue(vals.keySet().contains(evt.getKey()));
+ assertEquals(CREATED, evt.getEventType());
+ assertEquals(vals.get(evt.getKey()), evt.getValue());
+ assertNull(evt.getOldValue());
+
+ iter.remove();
+ }
+ }
+
+ for (Integer key : vals.keySet()) {
+ if (key % 2 == 0) {
+ CacheEntryEvent<? extends Integer, ? extends Integer> evt = iter.next();
+
+ assertTrue(evt.getKey() % 2 == 0);
+ assertTrue(vals.keySet().contains(evt.getKey()));
+ assertEquals(UPDATED, evt.getEventType());
+ assertEquals(-1, (int) evt.getValue());
+ assertEquals(vals.get(evt.getKey()), evt.getOldValue());
+
+ iter.remove();
+ }
+ }
+ }
+
+ /**
+ * @param cache Cache.
+ * @param lsnrCfg Listener configuration.
+ * @param key Key.
+ * @param create {@code True} if listens for create events.
+ * @param update {@code True} if listens for update events.
+ * @param rmv {@code True} if listens for remove events.
+ * @throws Exception If failed.
+ */
+ private void checkEvents(
+ final IgniteCache<Integer, Integer> cache,
+ final CacheEntryListenerConfiguration<Integer, Integer> lsnrCfg,
+ Integer key,
+ boolean create,
+ boolean update,
+ boolean rmv) throws Exception {
+ GridTestUtils.assertThrows(log, new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ cache.registerCacheEntryListener(lsnrCfg);
+
+ return null;
+ }
+ }, IllegalArgumentException.class, null);
+
+ final int UPDATES = 10;
+
+ int expEvts = 0;
+
+ if (create)
+ expEvts += 2;
+
+ if (update)
+ expEvts += (UPDATES + 1);
+
+ if (rmv)
+ expEvts += 2;
+
+ evts = new ArrayList<>();
+
+ evtsLatch = new CountDownLatch(expEvts);
+
+ cache.put(key, 0);
+
+ for (int i = 0; i < UPDATES; i++)
+ cache.put(key, i + 1);
+
+ assertFalse(cache.putIfAbsent(key, -1));
+
+ assertFalse(cache.remove(key, -1));
+
+ assertTrue(cache.remove(key));
+
+ IgniteCache<Integer, Integer> cache1 = cache;
+
+ if (gridCount() > 1)
+ cache1 = jcache(1); // Do updates from another node.
+
+ cache1.put(key, 1);
+
+ cache1.put(key, 2);
+
+ assertTrue(cache1.remove(key));
+
+ evtsLatch.await(5000, TimeUnit.MILLISECONDS);
+
+ assertEquals(expEvts, evts.size());
+
+ Iterator<CacheEntryEvent<? extends Integer, ? extends Integer>> iter = evts.iterator();
+
+ if (create)
+ checkEvent(iter, key, CREATED, 0, null);
+
+ if (update) {
+ for (int i = 0; i < UPDATES; i++)
+ checkEvent(iter, key, UPDATED, i + 1, i);
+ }
+
+ if (rmv)
+ checkEvent(iter, key, REMOVED, null, UPDATES);
+
+ if (create)
+ checkEvent(iter, key, CREATED, 1, null);
+
+ if (update)
+ checkEvent(iter, key, UPDATED, 2, 1);
+
+ if (rmv)
+ checkEvent(iter, key, REMOVED, null, 2);
+
+ assertEquals(0, evts.size());
+
+ log.info("Remove listener. ");
+
+ cache.deregisterCacheEntryListener(lsnrCfg);
+
+ cache.put(key, 1);
+
+ cache.put(key, 2);
+
+ assertTrue(cache.remove(key));
+
+ U.sleep(500); // Sleep some time to ensure listener was really removed.
+
+ assertEquals(0, evts.size());
+
+ cache.registerCacheEntryListener(lsnrCfg);
+
+ cache.deregisterCacheEntryListener(lsnrCfg);
+ }
+
+ /**
+ * @param iter Received events iterator.
+ * @param expKey Expected key.
+ * @param expType Expected type.
+ * @param expVal Expected value.
+ * @param expOld Expected old value.
+ */
+ private void checkEvent(Iterator<CacheEntryEvent<? extends Integer, ? extends Integer>> iter,
+ Integer expKey,
+ EventType expType,
+ @Nullable Integer expVal,
+ @Nullable Integer expOld) {
+ assertTrue(iter.hasNext());
+
+ CacheEntryEvent<? extends Integer, ? extends Integer> evt = iter.next();
+
+ iter.remove();
+
+ assertTrue(evt.getSource() instanceof IgniteCacheProxy);
+
+ assertEquals(expKey, evt.getKey());
+
+ assertEquals(expType, evt.getEventType());
+
+ assertEquals(expVal, evt.getValue());
+
+ assertEquals(expOld, evt.getOldValue());
+
+ if (expOld == null)
+ assertFalse(evt.isOldValueAvailable());
+ else
+ assertTrue(evt.isOldValueAvailable());
+ }
+
+ /**
+ * @return Test keys.
+ * @throws Exception If failed.
+ */
+ protected Collection<Integer> keys() throws Exception {
+ GridCache<Integer, Object> cache = cache(0);
+
+ ArrayList<Integer> keys = new ArrayList<>();
+
+ keys.add(primaryKeys(cache, 1, lastKey).get(0));
+
+ if (gridCount() > 1) {
+ keys.add(backupKeys(cache, 1, lastKey).get(0));
+
+ if (cache.configuration().getCacheMode() != REPLICATED)
+ keys.add(nearKeys(cache, 1, lastKey).get(0));
+ }
+
+ lastKey = Collections.max(keys) + 1;
+
+ return keys;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ evts = null;
+
+ evtsLatch = null;
+ }
+
+ /**
+ * @param evt Event.
+ */
+ private static void onEvent(CacheEntryEvent<? extends Integer, ? extends Integer> evt) {
+ //System.out.println("Received event [evt=" + evt + ", thread=" + Thread.currentThread().getName() + ']');
+
+ assert evt != null;
+ assert evt.getSource() != null : evt;
+ assert evt.getEventType() != null : evt;
+ assert evt.getKey() != null : evt;
+
+ evts.add(evt);
+
+ evtsLatch.countDown();
+ }
+
+ /**
+ *
+ */
+ static class TestFilterFactory implements Factory<CacheEntryEventFilter<Integer, Integer>> {
+ /** {@inheritDoc} */
+ @Override public CacheEntryEventFilter<Integer, Integer> create() {
+ return new TestFilter();
+ }
+ }
+
+ /**
+ *
+ */
+ static class TestFilter implements CacheEntryEventFilter<Integer, Integer> {
+ /** {@inheritDoc} */
+ @Override public boolean evaluate(CacheEntryEvent<? extends Integer, ? extends Integer> evt) {
+ assert evt != null;
+ assert evt.getSource() != null : evt;
+ assert evt.getEventType() != null : evt;
+ assert evt.getKey() != null : evt;
+
+ return evt.getKey() % 2 == 0;
+ }
+ }
+
+ /**
+ *
+ */
+ static class CreateUpdateListener implements CacheEntryCreatedListener<Integer, Integer>,
+ CacheEntryUpdatedListener<Integer, Integer> {
+ /** {@inheritDoc} */
+ @Override public void onCreated(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
+ for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts)
+ onEvent(evt);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
+ for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts)
+ onEvent(evt);
+ }
+ }
- Thread.sleep(2000);
+ /**
+ *
+ */
+ static class CreateUpdateRemoveListener extends CreateUpdateListener
+ implements CacheEntryRemovedListener<Integer, Integer> {
+ /** {@inheritDoc} */
+ @Override public void onRemoved(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
+ for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts)
+ onEvent(evt);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9f02315/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicLocalTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicLocalTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicLocalTest.java
new file mode 100644
index 0000000..5c7ec68
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAtomicLocalTest.java
@@ -0,0 +1,41 @@
+/* @java.file.header */
+
+/* _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.gridgain.grid.cache.*;
+
+import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
+import static org.gridgain.grid.cache.GridCacheDistributionMode.*;
+import static org.gridgain.grid.cache.GridCacheMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheEntryListenerAtomicLocalTest extends IgniteCacheEntryListenerAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected int gridCount() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheMode cacheMode() {
+ return LOCAL;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheAtomicityMode atomicityMode() {
+ return ATOMIC;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheDistributionMode distributionMode() {
+ return PARTITIONED_ONLY;
+ }
+}