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/30 13:11:48 UTC
[03/18] incubator-ignite git commit: # ignite-44
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.
*/