ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [03/50] [abbrv] incubator-ignite git commit: # ignite-44
Date Mon, 19 Jan 2015 14:20:40 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/928aa3d4/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalEx.java
b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalEx.java
index 63b5a79..8a485b6 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalEx.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalEx.java
@@ -16,6 +16,7 @@ import org.gridgain.grid.kernal.processors.cache.*;
 import org.gridgain.grid.kernal.processors.cache.dr.*;
 import org.jetbrains.annotations.*;
 
+import javax.cache.processor.*;
 import java.util.*;
 
 /**
@@ -53,6 +54,7 @@ public interface IgniteTxLocalEx<K, V> extends IgniteTxEx<K, V>
{
     @Nullable public IgniteTxEntry<K, V> groupLockEntry();
 
     /**
+     * @param cacheCtx Cache context.
      * @param keys Keys to get.
      * @param cached Cached entry if this method is called from entry wrapper.
      *      Cached entry is passed if and only if there is only one key in collection of
keys.
@@ -68,6 +70,7 @@ public interface IgniteTxLocalEx<K, V> extends IgniteTxEx<K, V>
{
         IgnitePredicate<GridCacheEntry<K, V>>[] filter);
 
     /**
+     * @param cacheCtx Cache context.
      * @param map Map to put.
      * @param retval Flag indicating whether a value should be returned.
      * @param cached Cached entry, if any. Will be provided only if map has size 1.
@@ -84,17 +87,20 @@ public interface IgniteTxLocalEx<K, V> extends IgniteTxEx<K,
V> {
         IgnitePredicate<GridCacheEntry<K, V>>[] filter);
 
     /**
-     * @param map Map to put.
+     * @param cacheCtx Cache context.
+     * @param map Entry processors map.
+     * @param invokeArgs Optional arguments for entry processor.
+     * @param retval Flag indicating whether a value should be returned.
      * @return Transform operation future.
      */
-    public IgniteFuture<GridCacheReturn<V>> transformAllAsync(
+    public <T> IgniteFuture<GridCacheReturn<Map<K, EntryProcessorResult<T>>>>
invokeAsync(
         GridCacheContext<K, V> cacheCtx,
-        @Nullable Map<? extends K, ? extends IgniteClosure<V, V>> map,
         boolean retval,
-        @Nullable GridCacheEntryEx<K, V> cached,
-        long ttl);
+        Map<? extends K, EntryProcessor> map,
+        Object... invokeArgs);
 
     /**
+     * @param cacheCtx Cache context.
      * @param keys Keys to remove.
      * @param retval Flag indicating whether a value should be returned.
      * @param cached Cached entry, if any. Will be provided only if size of keys collection
is 1.
@@ -109,6 +115,7 @@ public interface IgniteTxLocalEx<K, V> extends IgniteTxEx<K,
V> {
         IgnitePredicate<GridCacheEntry<K, V>>[] filter);
 
     /**
+     * @param cacheCtx Cache context.
      * @param drMap DR map to put.
      * @return Future for DR put operation.
      */
@@ -117,6 +124,7 @@ public interface IgniteTxLocalEx<K, V> extends IgniteTxEx<K,
V> {
         Map<? extends K, GridCacheDrInfo<V>> drMap);
 
     /**
+     * @param cacheCtx Cache context.
      * @param drMap DR map.
      * @return Future for asynchronous remove.
      */
@@ -127,6 +135,7 @@ public interface IgniteTxLocalEx<K, V> extends IgniteTxEx<K,
V> {
     /**
      * Performs keys locking for affinity-based group lock transactions.
      *
+     * @param cacheCtx Cache context.
      * @param keys Keys to lock.
      * @return Lock future.
      */
@@ -147,6 +156,7 @@ public interface IgniteTxLocalEx<K, V> extends IgniteTxEx<K,
V> {
     public boolean finish(boolean commit) throws IgniteCheckedException;
 
     /**
+     * @param cacheCtx  Cache context.
      * @param async if {@code True}, then loading will happen in a separate thread.
      * @param keys Keys.
      * @param c Closure.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/928aa3d4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheInvokeAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheInvokeAbstractTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheInvokeAbstractTest.java
index 1da627b..8731306 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheInvokeAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheInvokeAbstractTest.java
@@ -10,7 +10,9 @@
 package org.apache.ignite.internal.processors.cache;
 
 import org.apache.ignite.*;
+import org.apache.ignite.transactions.*;
 import org.gridgain.grid.cache.*;
+import org.gridgain.grid.util.typedef.internal.*;
 import org.gridgain.testframework.*;
 import org.jetbrains.annotations.*;
 
@@ -19,6 +21,9 @@ import javax.cache.processor.*;
 import java.util.*;
 import java.util.concurrent.*;
 
+import static org.apache.ignite.transactions.IgniteTxConcurrency.*;
+import static org.apache.ignite.transactions.IgniteTxIsolation.*;
+import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
 import static org.gridgain.grid.cache.GridCacheMode.*;
 
 /**
@@ -34,39 +39,86 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
     public void testInvoke() throws Exception {
         // TODO IGNITE41 test with forceTransformBackups.
 
+        invoke(null);
+
+        if (atomicityMode() == TRANSACTIONAL) {
+            invoke(PESSIMISTIC);
+
+            invoke(OPTIMISTIC);
+        }
+    }
+
+    /**
+     * @param txMode Not null transaction concurrency mode if explicit transaction should
be started.
+     * @throws Exception If failed.
+     */
+    private void invoke(@Nullable IgniteTxConcurrency txMode) throws Exception {
         final IgniteCache<Integer, Integer> cache = jcache();
 
         IncrementProcessor incProcessor = new IncrementProcessor();
 
         for (final Integer key : keys()) {
-            log.info("Test invoke [key=" + key + ']');
+            log.info("Test invoke [key=" + key + ", txMode=" + txMode + ']');
 
             cache.remove(key);
 
+            IgniteTx tx = startTx(txMode);
+
             Integer res = cache.invoke(key, incProcessor);
 
+            if (tx != null)
+                tx.commit();
+
             assertEquals(-1, (int)res);
 
             checkValue(key, 1);
 
+            tx = startTx(txMode);
+
             res = cache.invoke(key, incProcessor);
 
+            if (tx != null)
+                tx.commit();
+
             assertEquals(1, (int)res);
 
             checkValue(key, 2);
 
+            tx = startTx(txMode);
+
             res = cache.invoke(key, incProcessor);
 
+            if (tx != null)
+                tx.commit();
+
             assertEquals(2, (int)res);
 
             checkValue(key, 3);
 
+            tx = startTx(txMode);
+
             res = cache.invoke(key, new ArgumentsSumProcessor(), 10, 20, 30);
 
+            if (tx != null)
+                tx.commit();
+
             assertEquals(3, (int)res);
 
             checkValue(key, 63);
 
+            tx = startTx(txMode);
+
+            String strRes = cache.invoke(key, new ToStringProcessor());
+
+            if (tx != null)
+                tx.commit();
+
+            assertEquals("63", strRes);
+
+            checkValue(key, 63);
+
+            tx = startTx(txMode);
+
             GridTestUtils.assertThrows(log, new Callable<Void>() {
                 @Override public Void call() throws Exception {
                     cache.invoke(key, new ExceptionProcessor(63));
@@ -75,10 +127,18 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
                 }
             }, EntryProcessorException.class, "Test processor exception.");
 
+            if (tx != null)
+                tx.commit();
+
             checkValue(key, 63);
 
+            tx = startTx(txMode);
+
             assertNull(cache.invoke(key, new RemoveProcessor(63)));
 
+            if (tx != null)
+                tx.commit();
+
             checkValue(key, null);
         }
     }
@@ -87,13 +147,27 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
      * @throws Exception If failed.
      */
     public void testInvokeAll() throws Exception {
+        invokeAll(null);
+
+        if (atomicityMode() == TRANSACTIONAL) {
+            invoke(PESSIMISTIC);
+
+            invoke(OPTIMISTIC);
+        }
+    }
+
+    /**
+     * @param txMode Not null transaction concurrency mode if explicit transaction should
be started.
+     * @throws Exception If failed.
+     */
+    private void invokeAll(@Nullable IgniteTxConcurrency txMode) throws Exception {
         IgniteCache<Integer, Integer> cache = jcache();
 
-        invokeAll(cache, new HashSet<>(primaryKeys(cache, 3, 0)));
+        invokeAll(cache, new HashSet<>(primaryKeys(cache, 3, 0)), txMode);
 
-        invokeAll(cache, new HashSet<>(backupKeys(cache, 3, 0)));
+        invokeAll(cache, new HashSet<>(backupKeys(cache, 3, 0)), txMode);
 
-        invokeAll(cache, new HashSet<>(nearKeys(cache, 3, 0)));
+        invokeAll(cache, new HashSet<>(nearKeys(cache, 3, 0)), txMode);
 
         Set<Integer> keys = new HashSet<>();
 
@@ -101,29 +175,38 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
         keys.addAll(primaryKeys(jcache(1), 3, 0));
         keys.addAll(primaryKeys(jcache(2), 3, 0));
 
-        invokeAll(cache, keys);
+        invokeAll(cache, keys, txMode);
 
         keys = new HashSet<>();
 
         for (int i = 0; i < 1000; i++)
             keys.add(i);
 
-        invokeAll(cache, keys);
+        invokeAll(cache, keys, txMode);
     }
 
     /**
      * @param cache Cache.
      * @param keys Keys.
+     * @param txMode Not null transaction concurrency mode if explicit transaction should
be started.
+     * @throws Exception If failed.
      */
-    private void invokeAll(IgniteCache<Integer, Integer> cache, Set<Integer>
keys) {
+    private void invokeAll(IgniteCache<Integer, Integer> cache,
+        Set<Integer> keys,
+        @Nullable IgniteTxConcurrency txMode) throws Exception {
         cache.removeAll(keys);
 
-        log.info("Test invokeAll [keys=" + keys + ']');
+        log.info("Test invokeAll [keys=" + keys + ", txMode=" + txMode + ']');
 
         IncrementProcessor incProcessor = new IncrementProcessor();
 
+        IgniteTx tx = startTx(txMode);
+
         Map<Integer, EntryProcessorResult<Integer>> resMap = cache.invokeAll(keys,
incProcessor);
 
+        if (tx != null)
+            tx.commit();
+
         Map<Object, Object> exp = new HashMap<>();
 
         for (Integer key : keys)
@@ -134,8 +217,13 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
         for (Integer key : keys)
             checkValue(key, 1);
 
+        tx = startTx(txMode);
+
         resMap = cache.invokeAll(keys, incProcessor);
 
+        if (tx != null)
+            tx.commit();
+
         exp = new HashMap<>();
 
         for (Integer key : keys)
@@ -146,8 +234,13 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
         for (Integer key : keys)
             checkValue(key, 2);
 
+        tx = startTx(txMode);
+
         resMap = cache.invokeAll(keys, new ArgumentsSumProcessor(), 10, 20, 30);
 
+        if (tx != null)
+            tx.commit();
+
         for (Integer key : keys)
             exp.put(key, 3);
 
@@ -156,8 +249,13 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
         for (Integer key : keys)
             checkValue(key, 62);
 
+        tx = startTx(txMode);
+
         resMap = cache.invokeAll(keys, new ExceptionProcessor(null));
 
+        if (tx != null)
+            tx.commit();
+
         for (Integer key : keys) {
             final EntryProcessorResult<Integer> res = resMap.get(key);
 
@@ -175,8 +273,13 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
         for (Integer key : keys)
             checkValue(key, 62);
 
+        tx = startTx(txMode);
+
         resMap = cache.invokeAll(keys, new RemoveProcessor(null));
 
+        if (tx != null)
+            tx.commit();
+
         for (Integer key : keys) {
             final EntryProcessorResult<Integer> res = resMap.get(key);
 
@@ -237,7 +340,7 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
      * @return Test keys.
      * @throws Exception If failed.
      */
-    private Collection<Integer> keys() throws Exception {
+    protected Collection<Integer> keys() throws Exception {
         GridCache<Integer, Object> cache = cache(0);
 
         ArrayList<Integer> keys = new ArrayList<>();
@@ -257,6 +360,14 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
     }
 
     /**
+     * @param txMode Transaction concurrency mode.
+     * @return Transaction.
+     */
+    @Nullable private IgniteTx startTx(@Nullable IgniteTxConcurrency txMode) {
+        return txMode == null ? null : ignite(0).transactions().txStart(txMode, REPEATABLE_READ);
+    }
+
+    /**
      *
      */
     private static class ArgumentsSumProcessor implements EntryProcessor<Integer, Integer,
Integer> {
@@ -284,10 +395,27 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
     /**
      *
      */
-    private static class IncrementProcessor implements EntryProcessor<Integer, Integer,
Integer> {
+    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 IncrementProcessor implements EntryProcessor<Integer, Integer,
Integer> {
         /** {@inheritDoc} */
         @Override public Integer process(MutableEntry<Integer, Integer> e,
             Object... arguments) throws EntryProcessorException {
+            System.out.println(Thread.currentThread() + " compute, old=" + e.getValue());
             if (e.exists()) {
                 Integer val = e.getValue();
 
@@ -307,6 +435,11 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
                 return -1;
             }
         }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(IncrementProcessor.class, this);
+        }
     }
 
     /**
@@ -337,6 +470,11 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
 
             return null;
         }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(RemoveProcessor.class, this);
+        }
     }
 
     /**
@@ -363,5 +501,10 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
 
             throw new EntryProcessorException("Test processor exception.");
         }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(ExceptionProcessor.class, this);
+        }
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/928aa3d4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheTxInvokeTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheTxInvokeTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheTxInvokeTest.java
new file mode 100644
index 0000000..aaa2a21
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheTxInvokeTest.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 IgniteCacheTxInvokeTest extends IgniteCacheInvokeAbstractTest {
+    /** {@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/928aa3d4/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java
b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java
index 2ffc327..fd3751c 100644
--- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java
+++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java
@@ -25,6 +25,7 @@ import org.gridgain.testframework.*;
 import org.jetbrains.annotations.*;
 
 import javax.cache.expiry.*;
+import javax.cache.processor.*;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.*;
@@ -42,17 +43,29 @@ import static org.gridgain.testframework.GridTestUtils.*;
  * Full API cache test.
  */
 public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstractSelfTest
{
-    /** Increment closure for transform operations. */
-    public static final IgniteClosure<Integer,Integer> INCR_CLOS = new IgniteClosure<Integer,
Integer>() {
-        @Override public Integer apply(Integer old) {
-            return old == null ? 1 : old + 1;
+    /** Increment processor for invoke operations. */
+    public static final EntryProcessor<String, Integer, String> INCR_PROCESSOR = new
EntryProcessor<String, Integer, String>() {
+        @Override public String process(MutableEntry<String, Integer> e, Object...
args) {
+            assertNotNull(e.getKey());
+
+            Integer old = e.getValue();
+
+            e.setValue(old == null ? 1 : old + 1);
+
+            return String.valueOf(old);
         }
     };
 
-    /** Remove closure for transform operations. */
-    public static final IgniteClosure<Integer,Integer> RMV_CLOS = new IgniteClosure<Integer,
Integer>() {
-        @Override public Integer apply(Integer e) {
-            return null;
+    /** Increment processor for invoke operations. */
+    public static final EntryProcessor<String, Integer, String> RMV_PROCESSOR = new
EntryProcessor<String, Integer, String>() {
+        @Override public String process(MutableEntry<String, Integer> e, Object...
args) {
+            assertNotNull(e.getKey());
+
+            Integer old = e.getValue();
+
+            e.remove();
+
+            return String.valueOf(old);
         }
     };
 
@@ -772,18 +785,17 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
      * @throws Exception If failed.
      */
     private void checkTransform(IgniteTxConcurrency concurrency, IgniteTxIsolation isolation)
throws Exception {
-        GridCacheProjection<String, Integer> cache = cache();
+        IgniteCache<String, Integer> cache = jcache();
 
         cache.put("key2", 1);
         cache.put("key3", 3);
 
-
-        IgniteTx tx = txEnabled() ? cache.txStart(concurrency, isolation) : null;
+        IgniteTx tx = txEnabled() ? ignite(0).transactions().txStart(concurrency, isolation)
: null;
 
         try {
-            cache.transform("key1", INCR_CLOS);
-            cache.transform("key2", INCR_CLOS);
-            cache.transform("key3", RMV_CLOS);
+            assertEquals("null", cache.invoke("key1", INCR_PROCESSOR));
+            assertEquals("1", cache.invoke("key2", INCR_PROCESSOR));
+            assertEquals("3", cache.invoke("key3", RMV_PROCESSOR));
 
             if (tx != null)
                 tx.commit();
@@ -809,9 +821,9 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
         cache.put("key2", 1);
         cache.put("key3", 3);
 
-        cache.transform("key1", INCR_CLOS);
-        cache.transform("key2", INCR_CLOS);
-        cache.transform("key3", RMV_CLOS);
+        assertEquals("null", cache.invoke("key1", INCR_PROCESSOR));
+        assertEquals("1", cache.invoke("key2", INCR_PROCESSOR));
+        assertEquals("3", cache.invoke("key3", RMV_PROCESSOR));
 
         assertEquals((Integer)1, cache.get("key1"));
         assertEquals((Integer)2, cache.get("key2"));
@@ -856,25 +868,32 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
      */
     private void checkTransformAll(IgniteTxConcurrency concurrency, IgniteTxIsolation isolation)
         throws Exception {
-        GridCacheProjection<String, Integer> cache = cache();
+        final IgniteCache<String, Integer> cache = jcache();
 
         cache.put("key2", 1);
         cache.put("key3", 3);
 
         if (txEnabled()) {
-            CU.inTx(cache, concurrency, isolation, new CIX1<GridCacheProjection<String,
Integer>>() {
-                @Override
-                public void applyx(GridCacheProjection<String, Integer> c) throws IgniteCheckedException
{
-                    c.transformAll(F.asSet("key1", "key2", "key3"), INCR_CLOS);
-                }
-            });
+            Map<String, EntryProcessorResult<String>> res;
+
+            try (IgniteTx tx = ignite(0).transactions().txStart(concurrency, isolation))
{
+                res = cache.invokeAll(F.asSet("key1", "key2", "key3"), INCR_PROCESSOR);
+
+                tx.commit();
+            }
 
             assertEquals((Integer)1, cache.get("key1"));
             assertEquals((Integer)2, cache.get("key2"));
             assertEquals((Integer)4, cache.get("key3"));
+
+            assertEquals("null", res.get("key1").get());
+            assertEquals("1", res.get("key2").get());
+            assertEquals("3", res.get("key3").get());
+
+            assertEquals(3, res.size());
         }
 
-        cache.transformAll(F.asSet("key1", "key2", "key3"), RMV_CLOS);
+        Map<String, EntryProcessorResult<String>> res = cache.invokeAll(F.asSet("key1",
"key2", "key3"), RMV_PROCESSOR);
 
         for (int i = 0; i < gridCount(); i++) {
             assertNull(cache(i).peek("key1"));
@@ -882,15 +901,27 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
             assertNull(cache(i).peek("key3"));
         }
 
+        assertEquals("1", res.get("key1").get());
+        assertEquals("2", res.get("key2").get());
+        assertEquals("4", res.get("key3").get());
+
+        assertEquals(3, res.size());
+
         cache.remove("key1");
         cache.put("key2", 1);
         cache.put("key3", 3);
 
-        cache.transformAll(F.asMap("key1", INCR_CLOS, "key2", INCR_CLOS, "key3", INCR_CLOS));
+        res = cache.invokeAll(F.asSet("key1", "key2", "key3"), INCR_PROCESSOR);
 
         assertEquals((Integer)1, cache.get("key1"));
         assertEquals((Integer)2, cache.get("key2"));
         assertEquals((Integer)4, cache.get("key3"));
+
+        assertEquals("null", res.get("key1").get());
+        assertEquals("1", res.get("key2").get());
+        assertEquals("3", res.get("key3").get());
+
+        assertEquals(3, res.size());
     }
 
     /**
@@ -915,8 +946,8 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
         {
             Map<String, IgniteClosure<Integer, Integer>> tm = new HashMap<>(2);
 
-            tm.put("key1", INCR_CLOS);
-            tm.put(null, INCR_CLOS);
+            tm.put("key1", INCR_PROCESSOR);
+            tm.put(null, INCR_PROCESSOR);
 
             // WARN: F.asMap() doesn't work here, because it will throw NPE.
 
@@ -926,7 +957,7 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
         {
             Map<String, IgniteClosure<Integer, Integer>> tm = new HashMap<>(2);
 
-            tm.put("key1", INCR_CLOS);
+            tm.put("key1", INCR_PROCESSOR);
             tm.put("key2", null);
 
             // WARN: F.asMap() doesn't work here, because it will throw NPE.
@@ -934,7 +965,7 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
             cache.transformAll(tm);
         }
 
-        cache.transformAll(null, INCR_CLOS); // This should be no-op.
+        cache.transformAll(null, INCR_PROCESSOR); // This should be no-op.
 
         {
             Set<String> ts = new HashSet<>(3);
@@ -944,7 +975,7 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
 
             // WARN: F.asSet() doesn't work here, because it will throw NPE.
 
-            cache.transformAll(ts, INCR_CLOS);
+            cache.transformAll(ts, INCR_PROCESSOR);
         }
     }
 
@@ -991,9 +1022,9 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
             if (startVal)
                 cache.put("key", 2);
 
-            cache.transform("key", INCR_CLOS);
-            cache.transform("key", INCR_CLOS);
-            cache.transform("key", INCR_CLOS);
+            cache.transform("key", INCR_PROCESSOR);
+            cache.transform("key", INCR_PROCESSOR);
+            cache.transform("key", INCR_PROCESSOR);
 
             if (tx != null)
                 tx.commit();
@@ -1041,9 +1072,9 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
         try {
             cache.remove("key");
 
-            cache.transform("key", INCR_CLOS);
-            cache.transform("key", INCR_CLOS);
-            cache.transform("key", INCR_CLOS);
+            cache.transform("key", INCR_PROCESSOR);
+            cache.transform("key", INCR_PROCESSOR);
+            cache.transform("key", INCR_PROCESSOR);
 
             if (tx != null)
                 tx.commit();
@@ -1110,7 +1141,7 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
             if (put)
                 cache.put("key", 1);
 
-            cache.transform("key", INCR_CLOS);
+            cache.transform("key", INCR_PROCESSOR);
 
             assertEquals((Integer)2, cache.get("key"));
 
@@ -1190,7 +1221,7 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
 
         assertEquals((Integer)1, entry.getValue());
 
-        entry.transform(INCR_CLOS);
+        entry.transform(INCR_PROCESSOR);
 
         assertEquals((Integer)2, entry.getValue());
     }
@@ -1204,9 +1235,9 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
         cache.put("key2", 1);
         cache.put("key3", 3);
 
-        IgniteFuture<?> fut0 = cache.transformAsync("key1", INCR_CLOS);
-        IgniteFuture<?> fut1 = cache.transformAsync("key2", INCR_CLOS);
-        IgniteFuture<?> fut2 = cache.transformAsync("key3", RMV_CLOS);
+        IgniteFuture<?> fut0 = cache.transformAsync("key1", INCR_PROCESSOR);
+        IgniteFuture<?> fut1 = cache.transformAsync("key2", INCR_PROCESSOR);
+        IgniteFuture<?> fut2 = cache.transformAsync("key3", RMV_PROCESSOR);
 
         fut0.get();
         fut1.get();

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/928aa3d4/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractSelfTest.java
b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractSelfTest.java
index 314df0b..b4cd689 100644
--- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractSelfTest.java
@@ -348,6 +348,21 @@ public abstract class GridCacheAbstractSelfTest extends GridCommonAbstractTest
{
     }
 
     /**
+     * @return Default cache instance.
+     */
+    protected IgniteCache<String, Integer> jcache() {
+        return jcache(0);
+    }
+
+    /**
+     * @param idx Index of grid.
+     * @return Default cache.
+     */
+    protected IgniteCache<String, Integer> jcache(int idx) {
+        return ignite(idx).jcache(null);
+    }
+
+    /**
      * @param idx Index of grid.
      * @return Cache context.
      */


Mime
View raw message