ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [10/18] incubator-ignite git commit: # ignite-43 skip update if EntryProcessor does not modify entry
Date Mon, 19 Jan 2015 12:33:07 GMT
# ignite-43 skip update if EntryProcessor does not modify entry


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/fc0b5e8b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/fc0b5e8b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/fc0b5e8b

Branch: refs/heads/ignite-1
Commit: fc0b5e8b34a927eb9cf59efa5db16069c55980c9
Parents: 8f8f6ef
Author: sboikov <sboikov@gridgain.com>
Authored: Wed Jan 14 12:50:28 2015 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Wed Jan 14 12:50:28 2015 +0300

----------------------------------------------------------------------
 .../processors/cache/IgniteCacheProxy.java      | 25 ++++++++-
 .../processors/cache/CacheInvokeEntry.java      | 14 +++++
 .../processors/cache/GridCacheMapEntry.java     | 15 ++++++
 .../GridCacheContinuousQueryManager.java        |  4 +-
 .../cache/transactions/IgniteTxAdapter.java     | 22 +++++++-
 .../IgniteCacheEntryListenerAbstractTest.java   | 57 +++++++++++++++++++-
 6 files changed, 130 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fc0b5e8b/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 ac015e8..31264ba 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
@@ -850,7 +850,30 @@ public class IgniteCacheProxy<K, V> extends IgniteAsyncSupportAdapter
implements
     /** {@inheritDoc} */
     @Override public Iterator<Cache.Entry<K, V>> iterator() {
         // TODO IGNITE-1.
-        throw new UnsupportedOperationException();
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
+
+        try {
+            return F.iterator(delegate, new C1<GridCacheEntry<K, V>, Entry<K,
V>>() {
+                @Override public Entry<K, V> apply(final GridCacheEntry<K, V>
e) {
+                    return new Entry<K, V>() {
+                        @Override public K getKey() {
+                            return e.getKey();
+                        }
+
+                        @Override public V getValue() {
+                            return e.getValue();
+                        }
+
+                        @Override public <T> T unwrap(Class<T> clazz) {
+                            throw new IllegalArgumentException();
+                        }
+                    };
+                }
+            }, false);
+        }
+        finally {
+            gate.leave(prev);
+        }
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fc0b5e8b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/CacheInvokeEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/CacheInvokeEntry.java
b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/CacheInvokeEntry.java
index c9ca244..ab7dfc4 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/CacheInvokeEntry.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/CacheInvokeEntry.java
@@ -26,6 +26,9 @@ public class CacheInvokeEntry<K, V> implements MutableEntry<K,
V> {
     @GridToStringInclude
     private V val;
 
+    /** */
+    private boolean modified;
+
     /**
      * @param key Key.
      * @param val Value.
@@ -43,6 +46,8 @@ public class CacheInvokeEntry<K, V> implements MutableEntry<K,
V> {
     /** {@inheritDoc} */
     @Override public void remove() {
         val = null;
+
+        modified = true;
     }
 
     /** {@inheritDoc} */
@@ -51,6 +56,8 @@ public class CacheInvokeEntry<K, V> implements MutableEntry<K,
V> {
             throw new NullPointerException();
 
         this.val = val;
+
+        modified = true;
     }
 
     /** {@inheritDoc} */
@@ -68,6 +75,13 @@ public class CacheInvokeEntry<K, V> implements MutableEntry<K,
V> {
         throw new IllegalArgumentException();
     }
 
+    /**
+     * @return {@code True} if {@link #setValue} or {@link #remove was called}.
+     */
+    public boolean modified() {
+        return modified;
+    }
+
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(CacheInvokeEntry.class, this);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fc0b5e8b/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 b80f729..8c90c62 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
@@ -1479,6 +1479,9 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K,
V>
 
                     invokeRes = new CacheInvokeResult<>(e);
                 }
+
+                if (!entry.modified())
+                    return new GridTuple3<>(false, null, invokeRes);
             }
             else
                 updated = (V)writeObj;
@@ -1832,6 +1835,18 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K,
V>
 
                     valBytes = oldBytes.getIfMarshaled();
                 }
+
+                if (!entry.modified()) {
+                    return new GridCacheUpdateAtomicResult<>(false,
+                        retval ? old : null,
+                        null,
+                        invokeRes,
+                        -1L,
+                        -1L,
+                        null,
+                        null,
+                        false);
+                }
             }
             else
                 updated = (V)writeObj;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fc0b5e8b/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 56c7020..e180adc 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
@@ -584,10 +584,10 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
 
                 return fltr.evaluate(new org.apache.ignite.cache.CacheEntryEvent(cache, evtType,
entry));
             }
-            catch (CacheEntryListenerException e) {
+            catch (Exception e) {
                 LT.warn(ignite.log(), e, "Cache entry event filter error: " + e);
 
-                return false;
+                return true;
             }
         }
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fc0b5e8b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxAdapter.java
b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxAdapter.java
index bfd9359..f6ec594 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxAdapter.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxAdapter.java
@@ -23,6 +23,7 @@ import org.gridgain.grid.util.lang.*;
 import org.gridgain.grid.util.tostring.*;
 import org.jetbrains.annotations.*;
 
+import javax.cache.processor.*;
 import java.io.*;
 import java.util.*;
 import java.util.concurrent.atomic.*;
@@ -1199,9 +1200,26 @@ public abstract class IgniteTxAdapter<K, V> extends GridMetadataAwareAdapter
                         CU.<K, V>empty(),
                         null);
 
-                val = txEntry.applyEntryProcessors(val);
+                boolean modified = false;
 
-                GridCacheOperation op = val == null ? DELETE : UPDATE;
+                for (T2<EntryProcessor<K, V, ?>, Object[]> t : txEntry.entryProcessors())
{
+                    CacheInvokeEntry<K, V> invokeEntry = new CacheInvokeEntry<>(txEntry.key(),
val);
+
+                    try {
+                        EntryProcessor processor = t.get1();
+
+                        processor.process(invokeEntry, t.get2());
+
+                        val = invokeEntry.getValue();
+                    }
+                    catch (Exception ignore) {
+                        // No-op.
+                    }
+
+                    modified |= invokeEntry.modified();
+                }
+
+                GridCacheOperation op = modified ? (val == null ? DELETE : UPDATE) : NOOP;
 
                 return F.t(op, (V)cacheCtx.<V>unwrapTemporary(val), null);
             }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fc0b5e8b/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 f63ab9d..718e847 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
@@ -22,6 +22,7 @@ import org.jetbrains.annotations.*;
 import javax.cache.configuration.*;
 import javax.cache.event.*;
 import javax.cache.expiry.*;
+import javax.cache.processor.*;
 
 import java.util.*;
 import java.util.concurrent.*;
@@ -667,8 +668,15 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
 
         cache.put(key, 0);
 
-        for (int i = 0; i < UPDATES; i++)
-            cache.put(key, i + 1);
+        for (int i = 0; i < UPDATES; i++) {
+            if (i % 2 == 0)
+                cache.put(key, i + 1);
+            else
+                cache.invoke(key, new SetValueProcessor(i + 1));
+        }
+
+        // Invoke processor does not update value, should not trigger event.
+        assertEquals(String.valueOf(UPDATES), cache.invoke(key, new ToStringProcessor()));
 
         assertFalse(cache.putIfAbsent(key, -1));
 
@@ -941,4 +949,49 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
                 onEvent(evt);
         }
     }
+
+    /**
+     *
+     */
+    protected static class ToStringProcessor implements EntryProcessor<Integer, Integer,
String> {
+        /** {@inheritDoc} */
+        @Override public String process(MutableEntry<Integer, Integer> e, Object...
arguments)
+            throws EntryProcessorException {
+            return String.valueOf(e.getValue());
+        }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(ToStringProcessor.class, this);
+        }
+    }
+
+    /**
+     *
+     */
+    protected static class SetValueProcessor implements EntryProcessor<Integer, Integer,
String> {
+        /** */
+        private Integer val;
+
+        /**
+         * @param val Value to set.
+         */
+        public SetValueProcessor(Integer val) {
+            this.val = val;
+        }
+
+        /** {@inheritDoc} */
+        @Override public String process(MutableEntry<Integer, Integer> e, Object...
arguments)
+            throws EntryProcessorException {
+            e.setValue(val);
+
+            return null;
+        }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(SetValueProcessor.class, this);
+        }
+    }
+
 }


Mime
View raw message