You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by av...@apache.org on 2016/01/29 13:46:34 UTC
[1/3] ignite git commit: 2224
Repository: ignite
Updated Branches:
refs/heads/ignite-2224 [created] b28f63e46
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java
index 035f1b0..7286fb3 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java
@@ -34,6 +34,7 @@ import javax.cache.processor.EntryProcessorResult;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCompute;
+import org.apache.ignite.cache.CacheEntry;
import org.apache.ignite.cache.CacheEntryProcessor;
import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CachePeekMode;
@@ -129,7 +130,7 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K, V> {
}
/** {@inheritDoc} */
- @Override public void loadCache(@Nullable IgniteBiPredicate<K, V> p, @Nullable Object... args)
+ @Override public void loadCache(@Nullable IgniteBiPredicate<K, V> p, @Nullable Object... args)
throws CacheException {
throw new UnsupportedOperationException("Method should be supported.");
}
@@ -224,11 +225,21 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K, V> {
}
/** {@inheritDoc} */
+ @Override public CacheEntry<K, V> getEntry(K key) {
+ return compute.call(new GetEntryTask<K, V>(cacheName, isAsync, key));
+ }
+
+ /** {@inheritDoc} */
@Override public Map<K, V> getAll(Set<? extends K> keys) {
return compute.call(new GetAllTask<K, V>(cacheName, isAsync, keys));
}
/** {@inheritDoc} */
+ @Override public Collection<CacheEntry<K, V>> getEntries(Set<? extends K> keys) {
+ return compute.call(new GetEntriesTask<K, V>(cacheName, isAsync, keys));
+ }
+
+ /** {@inheritDoc} */
@Override public Map<K, V> getAllOutTx(Set<? extends K> keys) {
return compute.call(new GetAllOutTxTask<K, V>(cacheName, isAsync, keys));
}
@@ -710,6 +721,29 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K, V> {
/**
*
*/
+ private static class GetEntryTask<K, V> extends CacheTaskAdapter<K, V, CacheEntry<K, V>> {
+ /** Key. */
+ private final K key;
+
+ /**
+ * @param cacheName Cache name.
+ * @param async Async.
+ * @param key Key.
+ */
+ public GetEntryTask(String cacheName, boolean async, K key) {
+ super(cacheName, async);
+ this.key = key;
+ }
+
+ /** {@inheritDoc} */
+ @Override public CacheEntry<K, V> call() throws Exception {
+ return cache().getEntry(key);
+ }
+ }
+
+ /**
+ *
+ */
private static class RemoveAllTask<K, V> extends CacheTaskAdapter<K, V, Void> {
/**
* @param cacheName Cache name.
@@ -973,6 +1007,29 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K, V> {
/**
*
*/
+ private static class GetEntriesTask<K, V> extends CacheTaskAdapter<K, V, Collection<CacheEntry<K, V>> > {
+ /** Keys. */
+ private final Set<? extends K> keys;
+
+ /**
+ * @param cacheName Cache name.
+ * @param async Async.
+ * @param keys Keys.
+ */
+ public GetEntriesTask(String cacheName, boolean async, Set<? extends K> keys) {
+ super(cacheName, async);
+ this.keys = keys;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Collection<CacheEntry<K, V>> call() throws Exception {
+ return cache().getEntries(keys);
+ }
+ }
+
+ /**
+ *
+ */
private static class GetAllOutTxTask<K, V> extends CacheTaskAdapter<K, V, Map<K, V>> {
/** Keys. */
private final Set<? extends K> keys;
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
index 68e52df..45679dd 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
@@ -19,6 +19,12 @@ package org.apache.ignite.testsuites;
import junit.framework.TestSuite;
import org.apache.ignite.cache.store.jdbc.CacheJdbcStoreSessionListenerSelfTest;
+import org.apache.ignite.internal.processors.cache.CacheGetEntryOptimisticReadCommittedSeltTest;
+import org.apache.ignite.internal.processors.cache.CacheGetEntryOptimisticRepeatableReadSeltTest;
+import org.apache.ignite.internal.processors.cache.CacheGetEntryOptimisticSerializableSeltTest;
+import org.apache.ignite.internal.processors.cache.CacheGetEntryPessimisticReadCommittedSeltTest;
+import org.apache.ignite.internal.processors.cache.CacheGetEntryPessimisticRepeatableReadSeltTest;
+import org.apache.ignite.internal.processors.cache.CacheGetEntryPessimisticSerializableSeltTest;
import org.apache.ignite.internal.processors.cache.IgniteCacheGetCustomCollectionsSelfTest;
import org.apache.ignite.internal.processors.GridCacheTxLoadFromStoreOnLockSelfTest;
import org.apache.ignite.internal.processors.cache.CacheClientStoreSelfTest;
@@ -241,6 +247,12 @@ public class IgniteCacheTestSuite4 extends TestSuite {
suite.addTestSuite(IgniteSystemCacheOnClientTest.class);
suite.addTestSuite(CacheRemoveAllSelfTest.class);
+ suite.addTestSuite(CacheGetEntryOptimisticReadCommittedSeltTest.class);
+ suite.addTestSuite(CacheGetEntryOptimisticRepeatableReadSeltTest.class);
+ suite.addTestSuite(CacheGetEntryOptimisticSerializableSeltTest.class);
+ suite.addTestSuite(CacheGetEntryPessimisticReadCommittedSeltTest.class);
+ suite.addTestSuite(CacheGetEntryPessimisticRepeatableReadSeltTest.class);
+ suite.addTestSuite(CacheGetEntryPessimisticSerializableSeltTest.class);
suite.addTestSuite(CacheStopAndDestroySelfTest.class);
[2/3] ignite git commit: 2224
Posted by av...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
index 6130ead..fed3e33 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
@@ -477,7 +477,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
/** {@inheritDoc} */
@SuppressWarnings("unchecked")
- @Override @Nullable public V get(K key, boolean deserializeBinary) throws IgniteCheckedException {
+ @Override @Nullable public V get(K key, boolean deserializeBinary, boolean needVer) throws IgniteCheckedException {
String taskName = ctx.kernalContext().job().currentTaskName();
Map<K, V> m = getAllInternal(Collections.singleton(key),
@@ -485,7 +485,8 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
ctx.readThrough(),
taskName,
deserializeBinary,
- false);
+ false,
+ needVer);
assert m.isEmpty() || m.size() == 1 : m.size();
@@ -494,7 +495,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
/** {@inheritDoc} */
@SuppressWarnings("unchecked")
- @Override public final Map<K, V> getAll(Collection<? extends K> keys, boolean deserializeBinary)
+ @Override public final Map<K, V> getAll(Collection<? extends K> keys, boolean deserializeBinary, boolean needVer)
throws IgniteCheckedException {
A.notNull(keys, "keys");
@@ -505,7 +506,8 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
ctx.readThrough(),
taskName,
deserializeBinary,
- false);
+ false,
+ needVer);
}
@@ -519,7 +521,8 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
final String taskName,
final boolean deserializeBinary,
final boolean skipVals,
- boolean canRemap
+ boolean canRemap,
+ final boolean needVer
) {
A.notNull(keys, "keys");
@@ -528,7 +531,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
return asyncOp(new Callable<Map<K, V>>() {
@Override public Map<K, V> call() throws Exception {
- return getAllInternal(keys, swapOrOffheap, storeEnabled, taskName, deserializeBinary, skipVals);
+ return getAllInternal(keys, swapOrOffheap, storeEnabled, taskName, deserializeBinary, skipVals, needVer);
}
});
}
@@ -542,6 +545,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
* @param taskName Task name.
* @param deserializeBinary Deserialize binary .
* @param skipVals Skip value flag.
+ * @param needVer Need version.
* @return Key-value map.
* @throws IgniteCheckedException If failed.
*/
@@ -551,7 +555,8 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
boolean storeEnabled,
String taskName,
boolean deserializeBinary,
- boolean skipVals
+ boolean skipVals,
+ boolean needVer
) throws IgniteCheckedException {
ctx.checkSecurity(SecurityPermission.CACHE_READ);
@@ -584,24 +589,65 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
entry = swapOrOffheap ? entryEx(cacheKey) : peekEx(cacheKey);
if (entry != null) {
- CacheObject v = entry.innerGet(null,
- /*swap*/swapOrOffheap,
- /*read-through*/false,
- /*fail-fast*/false,
- /*unmarshal*/true,
- /**update-metrics*/true,
- /**event*/!skipVals,
- /**temporary*/false,
- subjId,
- null,
- taskName,
- expiry,
- !deserializeBinary);
+ CacheObject v ;
+ GridCacheVersion ver;
- if (v != null)
- ctx.addResult(vals, cacheKey, v, skipVals, false, deserializeBinary, true);
- else
- success = false;
+ if (needVer) {
+ T2<CacheObject, GridCacheVersion> res = entry.innerGetVersioned(
+ null,
+ /*swap*/swapOrOffheap,
+ /*unmarshal*/true,
+ /**update-metrics*/false,
+ /*event*/!skipVals,
+ subjId,
+ null,
+ taskName,
+ expiry,
+ !deserializeBinary);
+
+ if (res != null) {
+ v = res.get1();
+ ver = res.get2();
+
+ ctx.addResult(
+ vals,
+ cacheKey,
+ v,
+ skipVals,
+ false,
+ deserializeBinary,
+ true,
+ ver);
+ }else
+ success = false;
+ }
+ else {
+ v = entry.innerGet(null,
+ /*swap*/swapOrOffheap,
+ /*read-through*/false,
+ /*fail-fast*/false,
+ /*unmarshal*/true,
+ /**update-metrics*/true,
+ /**event*/!skipVals,
+ /**temporary*/false,
+ subjId,
+ null,
+ taskName,
+ expiry,
+ !deserializeBinary);
+
+ if (v != null) {
+ ctx.addResult(vals,
+ cacheKey,
+ v,
+ skipVals,
+ false,
+ deserializeBinary,
+ true);
+ }
+ else
+ success = false;
+ }
}
else {
if (!storeEnabled && configuration().isStatisticsEnabled() && !skipVals)
@@ -638,7 +684,8 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
/*force primary*/false,
expiry,
skipVals,
- /*can remap*/true).get();
+ /*can remap*/true,
+ needVer).get();
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
index 926eaf2..c034d01 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -1342,7 +1342,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
/**
* Checks if there is a cached or swapped value for
- * {@link #getAllAsync(GridCacheContext, Collection, boolean, boolean, boolean, boolean)} method.
+ * {@link #getAllAsync(GridCacheContext, Collection, boolean, boolean, boolean, boolean, boolean)} method.
*
* @param cacheCtx Cache context.
* @param keys Key to enlist.
@@ -1368,7 +1368,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
boolean deserializeBinary,
boolean skipVals,
boolean keepCacheObjects,
- boolean skipStore
+ boolean skipStore,
+ final boolean needVer
) throws IgniteCheckedException {
assert !F.isEmpty(keys);
assert keysCnt == keys.size();
@@ -1381,7 +1382,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
AffinityTopologyVersion topVer = topologyVersion();
- boolean needReadVer = serializable() && optimistic();
+ boolean needReadVer = (serializable() && optimistic()) || needVer;
// In this loop we cover only read-committed or optimistic transactions.
// Transactions that are pessimistic and not read-committed are covered
@@ -1403,31 +1404,80 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
if (!F.isEmpty(txEntry.entryProcessors()))
val = txEntry.applyEntryProcessors(val);
- if (val != null)
- cacheCtx.addResult(map, key, val, skipVals, keepCacheObjects, deserializeBinary, false);
+ if (val != null) {
+ GridCacheVersion ver = null;
+
+ if (needVer) {
+ if (txEntry.op() != READ)
+ ver = xidVersion();
+ else {
+ if (serializable()) {
+ ver = txEntry.serializableReadVersion();
+
+ assert ver != null; // TODO: value should not be null;
+ }
+ else if (optimistic()) {
+ assert isolation() == TransactionIsolation.REPEATABLE_READ;
+
+ throw new IgniteCheckedException(
+ "Impossible to getEntry() or getEntries() after get() at " +
+ "OPTIMISTIC REPEATABLE_READ case. " +
+ "Use only getEntry() or getEntries() to solve the problem. ");
+ }
+ else {
+ assert ver != null; // TODO: value should not be null;
+ }
+ }
+ }
+
+ cacheCtx.addResult(map, key, val, skipVals, keepCacheObjects, deserializeBinary, false, ver);
+ }
}
else {
assert txEntry.op() == TRANSFORM;
while (true) {
try {
+ GridCacheVersion readVer = null;
+
Object transformClo =
- (txEntry.op() == TRANSFORM && cctx.gridEvents().isRecordable(EVT_CACHE_OBJECT_READ)) ?
+ (txEntry.op() == TRANSFORM &&
+ cctx.gridEvents().isRecordable(EVT_CACHE_OBJECT_READ)) ?
F.first(txEntry.entryProcessors()) : null;
- val = txEntry.cached().innerGet(this,
- /*swap*/true,
- /*read-through*/false,
- /*fail fast*/true,
- /*unmarshal*/true,
- /*metrics*/true,
- /*event*/!skipVals,
- /*temporary*/false,
- CU.subjectId(this, cctx),
- transformClo,
- resolveTaskName(),
- null,
- txEntry.keepBinary());
+ if (needVer) {
+ T2<CacheObject, GridCacheVersion> res = txEntry.cached().innerGetVersioned(
+ this,
+ /*swap*/true,
+ /*unmarshal*/true,
+ /*update-metrics*/true,
+ /*event*/!skipVals,
+ CU.subjectId(this, cctx),
+ transformClo,
+ resolveTaskName(),
+ null,
+ txEntry.keepBinary());
+
+ if (res != null) {
+ val = res.get1();
+ readVer = res.get2();
+ }
+ }
+ else {
+ val = txEntry.cached().innerGet(this,
+ /*swap*/true,
+ /*read-through*/false,
+ /*fail fast*/true,
+ /*unmarshal*/true,
+ /*metrics*/true,
+ /*event*/!skipVals,
+ /*temporary*/false,
+ CU.subjectId(this, cctx),
+ transformClo,
+ resolveTaskName(),
+ null,
+ txEntry.keepBinary());
+ }
if (val != null) {
if (!readCommitted() && !skipVals)
@@ -1442,7 +1492,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
skipVals,
keepCacheObjects,
deserializeBinary,
- false);
+ false,
+ readVer);
}
else
missed.put(key, txEntry.cached().version());
@@ -1517,7 +1568,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
skipVals,
keepCacheObjects,
deserializeBinary,
- false);
+ false,
+ needVer ? readVer : null);
}
else
missed.put(key, ver);
@@ -1600,7 +1652,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
/**
* Loads all missed keys for
- * {@link #getAllAsync(GridCacheContext, Collection, boolean, boolean, boolean, boolean)} method.
+ * {@link #getAllAsync(GridCacheContext, Collection, boolean, boolean, boolean, boolean, boolean)} method.
*
* @param cacheCtx Cache context.
* @param map Return map.
@@ -1618,12 +1670,14 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
final boolean deserializeBinary,
final boolean skipVals,
final boolean keepCacheObjects,
- final boolean skipStore
+ final boolean skipStore,
+ final boolean needVer
+
) {
if (log.isDebugEnabled())
log.debug("Loading missed values for missed map: " + missedMap);
- final boolean needReadVer = serializable() && optimistic();
+ final boolean needReadVer = (serializable() && optimistic()) || needVer;
return new GridEmbeddedFuture<>(
new C2<Void, Exception, Map<K, V>>() {
@@ -1685,7 +1739,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
skipVals,
keepCacheObjects,
deserializeBinary,
- false);
+ false,
+ needVer ? loadVer : null);
}
}
else {
@@ -1706,7 +1761,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
skipVals,
keepCacheObjects,
deserializeBinary,
- false);
+ false,
+ needVer ? loadVer : null);
}
}
}
@@ -1721,7 +1777,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
final boolean deserializeBinary,
final boolean skipVals,
final boolean keepCacheObjects,
- final boolean skipStore) {
+ final boolean skipStore,
+ final boolean needVer) {
if (F.isEmpty(keys))
return new GridFinishedFuture<>(Collections.<K, V>emptyMap());
@@ -1751,7 +1808,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
deserializeBinary,
skipVals,
keepCacheObjects,
- skipStore);
+ skipStore,
+ needVer);
if (single && missed.isEmpty())
return new GridFinishedFuture<>(retMap);
@@ -1797,25 +1855,48 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
while (true) {
GridCacheEntryEx cached = txEntry.cached();
+ CacheObject val = null;
+ GridCacheVersion readVer = null;
+
try {
Object transformClo =
(!F.isEmpty(txEntry.entryProcessors()) &&
cctx.gridEvents().isRecordable(EVT_CACHE_OBJECT_READ)) ?
F.first(txEntry.entryProcessors()) : null;
- CacheObject val = cached.innerGet(IgniteTxLocalAdapter.this,
- cacheCtx.isSwapOrOffheapEnabled(),
- /*read-through*/false,
- /*fail-fast*/true,
- /*unmarshal*/true,
- /*metrics*/true,
- /*events*/!skipVals,
- /*temporary*/true,
- CU.subjectId(IgniteTxLocalAdapter.this, cctx),
- transformClo,
- resolveTaskName(),
- null,
- txEntry.keepBinary());
+ if (needVer) {
+ T2<CacheObject, GridCacheVersion> res = cached.innerGetVersioned(
+ IgniteTxLocalAdapter.this,
+ /*swap*/cacheCtx.isSwapOrOffheapEnabled(),
+ /*unmarshal*/true,
+ /**update-metrics*/true,
+ /*event*/!skipVals,
+ CU.subjectId(IgniteTxLocalAdapter.this, cctx),
+ transformClo,
+ resolveTaskName(),
+ null,
+ txEntry.keepBinary());
+
+ if (res != null) {
+ val = res.get1();
+ readVer = res.get2();
+ }
+ }
+ else{
+ val = cached.innerGet(IgniteTxLocalAdapter.this,
+ cacheCtx.isSwapOrOffheapEnabled(),
+ /*read-through*/false,
+ /*fail-fast*/true,
+ /*unmarshal*/true,
+ /*metrics*/true,
+ /*events*/!skipVals,
+ /*temporary*/true,
+ CU.subjectId(IgniteTxLocalAdapter.this, cctx),
+ transformClo,
+ resolveTaskName(),
+ null,
+ txEntry.keepBinary());
+ }
// If value is in cache and passed the filter.
if (val != null) {
@@ -1832,7 +1913,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
skipVals,
keepCacheObjects,
deserializeBinary,
- false);
+ false,
+ readVer);
}
// Even though we bring the value back from lock acquisition,
@@ -1858,7 +1940,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
deserializeBinary,
skipVals,
keepCacheObjects,
- skipStore);
+ skipStore,
+ needVer);
}
return new GridFinishedFuture<>(Collections.<K, V>emptyMap());
@@ -1924,7 +2007,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
deserializeBinary,
skipVals,
keepCacheObjects,
- skipStore);
+ skipStore,
+ needVer);
}
return new GridFinishedFuture<>(retMap);
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
index a5d3373..78f517c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
@@ -76,7 +76,8 @@ public interface IgniteTxLocalEx extends IgniteInternalTx {
boolean deserializeBinary,
boolean skipVals,
boolean keepCacheObjects,
- boolean skipStore);
+ boolean skipStore,
+ boolean needVer);
/**
* @param cacheCtx Cache context.
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryAbstractSeltTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryAbstractSeltTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryAbstractSeltTest.java
new file mode 100644
index 0000000..2e94cbc
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryAbstractSeltTest.java
@@ -0,0 +1,525 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.cache.CacheEntry;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter;
+import org.apache.ignite.internal.processors.cache.transactions.TransactionProxyImpl;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessor;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+
+/**
+ * Test getEntry and getEntries methods.
+ */
+public abstract class CacheGetEntryAbstractSeltTest extends GridCacheAbstractSelfTest {
+
+ @Override protected int gridCount() {
+ return 3;
+ }
+
+ abstract protected TransactionConcurrency concurrency();
+
+ abstract protected TransactionIsolation isolation();
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ cfg.setMarshaller(null);
+
+ return cfg;
+ }
+
+ /** */
+ public void testNear() {
+ CacheConfiguration cfg = new CacheConfiguration();
+
+ cfg.setCacheMode(CacheMode.PARTITIONED);
+ cfg.setName("near");
+ cfg.setNearConfiguration(new NearCacheConfiguration());
+
+ test(cfg);
+ }
+
+ /** */
+ public void testNearTransactional() {
+ CacheConfiguration cfg = new CacheConfiguration();
+
+ cfg.setCacheMode(CacheMode.PARTITIONED);
+ cfg.setAtomicityMode(TRANSACTIONAL);
+ cfg.setName("nearT");
+ cfg.setNearConfiguration(new NearCacheConfiguration());
+
+ test(cfg);
+ }
+
+ /** */
+ public void testPartitioned() {
+ CacheConfiguration cfg = new CacheConfiguration();
+ cfg.setCacheMode(CacheMode.PARTITIONED);
+ cfg.setName("partitioned");
+
+ test(cfg);
+ }
+
+ /** */
+ public void testPartitionedTransactional() {
+ CacheConfiguration cfg = new CacheConfiguration();
+
+ cfg.setCacheMode(CacheMode.PARTITIONED);
+ cfg.setAtomicityMode(TRANSACTIONAL);
+ cfg.setName("partitionedT");
+
+ test(cfg);
+ }
+
+ /** */
+ public void testLocal() {
+ CacheConfiguration cfg = new CacheConfiguration();
+
+ cfg.setCacheMode(CacheMode.LOCAL);
+ cfg.setName("local");
+
+ test(cfg);
+ }
+
+ /** */
+ public void testLocalTransactional() {
+ CacheConfiguration cfg = new CacheConfiguration();
+
+ cfg.setCacheMode(CacheMode.LOCAL);
+ cfg.setAtomicityMode(TRANSACTIONAL);
+ cfg.setName("localT");
+
+ test(cfg);
+ }
+
+ /** */
+ public void testReplicated() {
+ CacheConfiguration cfg = new CacheConfiguration();
+
+ cfg.setCacheMode(CacheMode.REPLICATED);
+ cfg.setName("replicated");
+
+ test(cfg);
+ }
+
+ /** */
+ public void testReplicatedTransactional() {
+ CacheConfiguration cfg = new CacheConfiguration();
+
+ cfg.setCacheMode(CacheMode.REPLICATED);
+ cfg.setAtomicityMode(TRANSACTIONAL);
+ cfg.setName("replicatedT");
+
+ test(cfg);
+ }
+
+ /** */
+ private void test(CacheConfiguration cfg) {
+ test(cfg, true);
+ test(cfg, false);
+ }
+
+ /** */
+ private void test(CacheConfiguration cfg, boolean oneEntry) {
+ IgniteCache<Integer, TestValue> cache = grid(0).createCache(cfg);
+ try {
+ init(cache);
+
+ test(cache, null, null, null, oneEntry);
+
+ if (cfg.getAtomicityMode() == TRANSACTIONAL) {
+ TransactionConcurrency txConcurrency = concurrency();
+ TransactionIsolation txIsolation = isolation();
+ try (Transaction tx = grid(0).transactions().txStart(txConcurrency, txIsolation, 100000, 1000)) {
+ initTx(cache);
+
+ test(cache, txConcurrency, txIsolation, tx, oneEntry);
+
+ tx.commit();
+ }
+ }
+ }
+ finally {
+ cache.destroy();
+ }
+ }
+
+ private Set<Integer> getKeys(int base) {
+ int start = 0;
+ int finish = 100;
+
+ Set<Integer> keys = new HashSet<>(finish - start);
+
+ for (int i = base + start; i < base + finish; ++i)
+ keys.add(i);
+
+ return keys;
+ }
+
+ private Set<Integer> createdBeforeTxKeys() {
+ return getKeys(0);
+ }
+
+ private Set<Integer> createdBeforeTxWithBinaryKeys() {
+ return getKeys(1_000);
+ }
+
+ private Set<Integer> createdBeforeTxKeys2() {
+ return getKeys(2_000);
+ }
+
+ private Set<Integer> createdBeforeTxWithBinaryKeys2() {
+ return getKeys(3_000);
+ }
+
+ private Set<Integer> createdBeforeTxKeys3() {
+ return getKeys(4_000);
+ }
+
+ private Set<Integer> createdBeforeTxWithBinaryKeys3() {
+ return getKeys(5_000);
+ }
+
+ private Set<Integer> removedBeforeTxKeys() {
+ return getKeys(6_000);
+ }
+
+ private Set<Integer> removedBeforeTxWithBinaryKeys() {
+ return getKeys(7_000);
+ }
+
+ private Set<Integer> createdAtTxKeys() {
+ return getKeys(8_000);
+ }
+
+ private Set<Integer> createdAtTxWithBinaryKeys() {
+ return getKeys(9_000);
+ }
+
+ private Set<Integer> removedAtTxKeys() {
+ return getKeys(10_000);
+ }
+
+ private Set<Integer> removedAtTxWithBinaryKeys() {
+ return getKeys(11_000);
+ }
+
+ /** */
+ private void init(IgniteCache<Integer, TestValue> cache) {
+ Set<Integer> keys = new HashSet<>();
+
+ keys.addAll(createdBeforeTxKeys());
+ keys.addAll(createdBeforeTxWithBinaryKeys());
+ keys.addAll(createdBeforeTxKeys2());
+ keys.addAll(createdBeforeTxWithBinaryKeys2());
+ keys.addAll(createdBeforeTxKeys3());
+ keys.addAll(createdBeforeTxWithBinaryKeys3());
+ keys.addAll(removedBeforeTxKeys());
+ keys.addAll(removedBeforeTxWithBinaryKeys());
+ keys.addAll(removedAtTxKeys());
+ keys.addAll(removedAtTxWithBinaryKeys());
+
+ for (int i : keys)
+ cache.put(i, new TestValue(i));
+
+ for (int i : removedBeforeTxKeys())
+ cache.remove(i);
+
+ for (int i : removedBeforeTxWithBinaryKeys())
+ cache.remove(i);
+ }
+
+ /** */
+ private void initTx(IgniteCache<Integer, TestValue> cache) {
+ for (int i : createdAtTxKeys())
+ cache.put(i, new TestValue(i));
+
+ for (int i : createdAtTxWithBinaryKeys())
+ cache.put(i, new TestValue(i));
+
+ for (int i : removedAtTxKeys())
+ cache.remove(i);
+
+ for (int i : removedAtTxWithBinaryKeys())
+ cache.remove(i);
+ }
+
+ /** */
+ private void compareVersionWithPrimaryNode(CacheEntry<Integer, ?> e, IgniteCache<Integer, TestValue> cache) {
+ CacheConfiguration cfg = cache.getConfiguration(CacheConfiguration.class);
+
+ if (cfg.getCacheMode() != CacheMode.LOCAL) {
+ Ignite prim = primaryNode(e.getKey(), cache.getName());
+
+ GridCacheAdapter<Object, Object> cacheAdapter = ((IgniteKernal)prim).internalCache(cache.getName());
+
+ if (cfg.getNearConfiguration() != null)
+ cacheAdapter = ((GridNearCacheAdapter)cacheAdapter).dht();
+
+ IgniteCacheObjectProcessor cacheObjects = cacheAdapter.context().cacheObjects();
+
+ CacheObjectContext cacheObjCtx = cacheAdapter.context().cacheObjectContext();
+
+ GridCacheMapEntry me = cacheAdapter.map().getEntry(cacheObjects.toCacheKeyObject(
+ cacheObjCtx, e.getKey(), true));
+
+ try {
+ assertEquals(me.version(), e.version());
+ }
+ catch (GridCacheEntryRemovedException ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+ }
+
+ private void checkData(IgniteCache<Integer, TestValue> cache, int i, boolean oneEntry, GridCacheVersion txVer) {
+ if (oneEntry) {
+ CacheEntry<Integer, TestValue> e = cache.getEntry(i);
+
+ if (txVer != null)
+ assertEquals(txVer, e.version());
+ else
+ compareVersionWithPrimaryNode(e, cache);
+
+ assertEquals(e.getValue().val, i);
+ }
+ else {
+ Set<Integer> set = new HashSet<>();
+
+ for (int j = 0; j < 10; j++)
+ set.add(i + j);
+
+ Collection<CacheEntry<Integer, TestValue>> es = cache.getEntries(set);
+
+ for (CacheEntry<Integer, TestValue> e : es) {
+ if (txVer != null)
+ assertEquals(txVer, e.version());
+ else
+ compareVersionWithPrimaryNode(e, cache);
+
+ assertEquals((Integer)e.getValue().val, e.getKey());
+
+ assertTrue(set.contains(e.getValue().val));
+ }
+ }
+ }
+
+ private void checkBinaryData(IgniteCache<Integer, TestValue> cache, int i, boolean oneEntry,
+ GridCacheVersion txVer) {
+ IgniteCache<Integer, BinaryObject> cacheB = cache.withKeepBinary();
+
+ if (oneEntry) {
+ CacheEntry<Integer, BinaryObject> e = cacheB.getEntry(i);
+
+ if (txVer != null)
+ assertEquals(txVer, e.version());
+ else
+ compareVersionWithPrimaryNode(e, cache);
+
+ assertEquals(((TestValue)e.getValue().deserialize()).val, i);
+ }
+ else {
+ Set<Integer> set = new HashSet<>();
+
+ for (int j = 0; j < 10; j++)
+ set.add(i + j);
+
+ Collection<CacheEntry<Integer, BinaryObject>> es = cacheB.getEntries(set);
+
+ for (CacheEntry<Integer, BinaryObject> e : es) {
+ if (txVer != null)
+ assertEquals(txVer, e.version());
+ else
+ compareVersionWithPrimaryNode(e, cache);
+
+ TestValue tv = e.getValue().deserialize();
+
+ assertEquals((Integer)tv.val, e.getKey());
+
+ assertTrue(set.contains((tv).val));
+ }
+ }
+ }
+
+ private void checkRemoved(IgniteCache<Integer, TestValue> cache, int i, boolean oneEntry) {
+ if (oneEntry) {
+ CacheEntry<Integer, TestValue> e = cache.getEntry(i);
+
+ assertNull(e);
+ }
+ else {
+ Set<Integer> set = new HashSet<>();
+
+ for (int j = 0; j < 10; j++)
+ set.add(i + j);
+
+ Collection<CacheEntry<Integer, TestValue>> es = cache.getEntries(set);
+
+ assertTrue(es.isEmpty());
+ }
+ }
+
+ private void checkBinaryRemoved(IgniteCache<Integer, TestValue> cache, int i, boolean oneEntry) {
+ IgniteCache<Integer, BinaryObject> cacheB = cache.withKeepBinary();
+
+ if (oneEntry) {
+ CacheEntry<Integer, BinaryObject> e = cacheB.getEntry(i);
+
+ assertNull(e);
+ }
+ else {
+ Set<Integer> set = new HashSet<>();
+
+ for (int j = 0; j < 10; j++)
+ set.add(i + j);
+
+ Collection<CacheEntry<Integer, BinaryObject>> es = cacheB.getEntries(set);
+
+ assertTrue(es.isEmpty());
+ }
+ }
+
+ /** */
+ private void test(IgniteCache<Integer, TestValue> cache,
+ TransactionConcurrency txConcurrency,
+ TransactionIsolation txIsolation,
+ Transaction tx,
+ boolean oneEntry) {
+ if (tx == null) {
+ for (int i : createdBeforeTxKeys()) {
+ checkData(cache, i, oneEntry, null);
+ }
+
+ for (int i : createdBeforeTxWithBinaryKeys()) {
+ checkBinaryData(cache, i, oneEntry, null);
+ }
+
+ for (int i : removedBeforeTxKeys()) {
+ checkRemoved(cache, i, oneEntry);
+ }
+
+ for (int i : removedBeforeTxWithBinaryKeys()) {
+ checkBinaryRemoved(cache, i, oneEntry);
+ }
+ }
+ else {
+ GridCacheVersion txVer = ((TransactionProxyImpl)tx).tx().xidVersion();
+
+ for (int i : createdBeforeTxKeys2()) {
+ checkData(cache, i, oneEntry, null);
+ checkData(cache, i, oneEntry, null);
+ }
+
+ for (int i : createdBeforeTxWithBinaryKeys2()) {
+ checkBinaryData(cache, i, oneEntry, null);
+ checkBinaryData(cache, i, oneEntry, null);
+ }
+
+ try {
+ for (int i : createdBeforeTxKeys3()) {
+ cache.get(i);
+
+ checkData(cache, i, oneEntry, null);
+ }
+
+ for (int i : createdBeforeTxWithBinaryKeys3()) {
+ cache.get(i);
+
+ checkBinaryData(cache, i, oneEntry, null);
+ }
+
+ assertFalse(txIsolation == TransactionIsolation.REPEATABLE_READ &&
+ txConcurrency == TransactionConcurrency.OPTIMISTIC);
+ }
+ catch (Exception e) {
+ assertTrue(txIsolation == TransactionIsolation.REPEATABLE_READ &&
+ txConcurrency == TransactionConcurrency.OPTIMISTIC);
+ }
+
+ for (int i : createdAtTxKeys()) {
+ checkData(cache, i, oneEntry, txVer);
+ }
+
+ for (int i : createdAtTxWithBinaryKeys()) {
+ checkBinaryData(cache, i, oneEntry, txVer);
+ }
+
+ for (int i : removedBeforeTxKeys()) {
+ checkRemoved(cache, i, oneEntry);
+ }
+
+ for (int i : removedBeforeTxWithBinaryKeys()) {
+ checkBinaryRemoved(cache, i, oneEntry);
+ }
+ for (int i : removedAtTxKeys()) {
+ checkRemoved(cache, i, oneEntry);
+ }
+
+ for (int i : removedAtTxWithBinaryKeys()) {
+ checkBinaryRemoved(cache, i, oneEntry);
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ private static class TestValue implements Serializable {
+ /** */
+ private int val;
+
+ /**
+ * @param val Value.
+ */
+ public TestValue(int val) {
+ this.val = val;
+ }
+
+ /**
+ * @return Value.
+ */
+ public int value() {
+ return val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(TestValue.class, this);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryOptimisticReadCommittedSeltTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryOptimisticReadCommittedSeltTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryOptimisticReadCommittedSeltTest.java
new file mode 100644
index 0000000..b7cd28b
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryOptimisticReadCommittedSeltTest.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+
+/**
+ * Test getEntry and getEntries methods.
+ */
+public class CacheGetEntryOptimisticReadCommittedSeltTest extends CacheGetEntryAbstractSeltTest {
+
+ @Override protected TransactionConcurrency concurrency() {
+ return TransactionConcurrency.OPTIMISTIC;
+ }
+
+ @Override protected TransactionIsolation isolation() {
+ return TransactionIsolation.READ_COMMITTED;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryOptimisticRepeatableReadSeltTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryOptimisticRepeatableReadSeltTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryOptimisticRepeatableReadSeltTest.java
new file mode 100644
index 0000000..d8453ba
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryOptimisticRepeatableReadSeltTest.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+
+/**
+ * Test getEntry and getEntries methods.
+ */
+public class CacheGetEntryOptimisticRepeatableReadSeltTest extends CacheGetEntryAbstractSeltTest {
+
+ @Override protected TransactionConcurrency concurrency() {
+ return TransactionConcurrency.OPTIMISTIC;
+ }
+
+ @Override protected TransactionIsolation isolation() {
+ return TransactionIsolation.REPEATABLE_READ;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryOptimisticSerializableSeltTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryOptimisticSerializableSeltTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryOptimisticSerializableSeltTest.java
new file mode 100644
index 0000000..4fbe66e
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryOptimisticSerializableSeltTest.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+
+/**
+ * Test getEntry and getEntries methods.
+ */
+public class CacheGetEntryOptimisticSerializableSeltTest extends CacheGetEntryAbstractSeltTest {
+
+ @Override protected TransactionConcurrency concurrency() {
+ return TransactionConcurrency.OPTIMISTIC;
+ }
+
+ @Override protected TransactionIsolation isolation() {
+ return TransactionIsolation.SERIALIZABLE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryPessimisticReadCommittedSeltTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryPessimisticReadCommittedSeltTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryPessimisticReadCommittedSeltTest.java
new file mode 100644
index 0000000..86d6e4e
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryPessimisticReadCommittedSeltTest.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+
+/**
+ * Test getEntry and getEntries methods.
+ */
+public class CacheGetEntryPessimisticReadCommittedSeltTest extends CacheGetEntryAbstractSeltTest {
+
+ @Override protected TransactionConcurrency concurrency() {
+ return TransactionConcurrency.PESSIMISTIC;
+ }
+
+ @Override protected TransactionIsolation isolation() {
+ return TransactionIsolation.READ_COMMITTED;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryPessimisticRepeatableReadSeltTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryPessimisticRepeatableReadSeltTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryPessimisticRepeatableReadSeltTest.java
new file mode 100644
index 0000000..7fc7aa3
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryPessimisticRepeatableReadSeltTest.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+
+/**
+ * Test getEntry and getEntries methods.
+ */
+public class CacheGetEntryPessimisticRepeatableReadSeltTest extends CacheGetEntryAbstractSeltTest {
+
+ @Override protected TransactionConcurrency concurrency() {
+ return TransactionConcurrency.PESSIMISTIC;
+ }
+
+ @Override protected TransactionIsolation isolation() {
+ return TransactionIsolation.REPEATABLE_READ;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryPessimisticSerializableSeltTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryPessimisticSerializableSeltTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryPessimisticSerializableSeltTest.java
new file mode 100644
index 0000000..da28eca
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntryPessimisticSerializableSeltTest.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+
+/**
+ * Test getEntry and getEntries methods.
+ */
+public class CacheGetEntryPessimisticSerializableSeltTest extends CacheGetEntryAbstractSeltTest {
+
+ @Override protected TransactionConcurrency concurrency() {
+ return TransactionConcurrency.PESSIMISTIC;
+ }
+
+ @Override protected TransactionIsolation isolation() {
+ return TransactionIsolation.SERIALIZABLE;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheReadThroughRestartSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheReadThroughRestartSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheReadThroughRestartSelfTest.java
index c606a2a..4a4f2ed 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheReadThroughRestartSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheReadThroughRestartSelfTest.java
@@ -85,6 +85,20 @@ public class CacheReadThroughRestartSelfTest extends GridCacheAbstractSelfTest {
* @throws Exception If failed.
*/
public void testReadThroughInTx() throws Exception {
+ testReadThroughInTx(false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReadEntryThroughInTx() throws Exception {
+ testReadThroughInTx(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReadThroughInTx(boolean needVer) throws Exception {
IgniteCache<String, Integer> cache = grid(1).cache(null);
for (int k = 0; k < 1000; k++)
@@ -104,7 +118,14 @@ public class CacheReadThroughRestartSelfTest extends GridCacheAbstractSelfTest {
for (int k = 0; k < 1000; k++) {
String key = "key" + k;
- assertNotNull("Null value for key: " + key, cache.get(key));
+ if (needVer) {
+ assertNotNull("Null value for key: " + key, cache.getEntry(key));
+ assertNotNull("Null value for key: " + key, cache.getEntry(key));
+ }
+ else {
+ assertNotNull("Null value for key: " + key, cache.get(key));
+ assertNotNull("Null value for key: " + key, cache.get(key));
+ }
}
tx.commit();
@@ -117,6 +138,20 @@ public class CacheReadThroughRestartSelfTest extends GridCacheAbstractSelfTest {
* @throws Exception If failed.
*/
public void testReadThrough() throws Exception {
+ testReadThrough(false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReadEntryThrough() throws Exception {
+ testReadThrough(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReadThrough(boolean needVer) throws Exception {
IgniteCache<String, Integer> cache = grid(1).cache(null);
for (int k = 0; k < 1000; k++)
@@ -132,8 +167,10 @@ public class CacheReadThroughRestartSelfTest extends GridCacheAbstractSelfTest {
for (int k = 0; k < 1000; k++) {
String key = "key" + k;
-
- assertNotNull("Null value for key: " + key, cache.get(key));
+ if (needVer)
+ assertNotNull("Null value for key: " + key, cache.getEntry(key));
+ else
+ assertNotNull("Null value for key: " + key, cache.get(key));
}
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheSerializableTransactionsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheSerializableTransactionsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheSerializableTransactionsTest.java
index 9906ad3..4baef66 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheSerializableTransactionsTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheSerializableTransactionsTest.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.cache;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -46,6 +47,7 @@ import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteTransactions;
+import org.apache.ignite.cache.CacheEntry;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.cache.store.CacheStore;
@@ -575,6 +577,20 @@ public class CacheSerializableTransactionsTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testTxCommitReadOnlyGetAll() throws Exception {
+ testTxCommitReadOnlyGetAll(false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxCommitReadOnlyGetEntries() throws Exception {
+ testTxCommitReadOnlyGetAll(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxCommitReadOnlyGetAll(boolean needVer) throws Exception {
Ignite ignite0 = ignite(0);
final IgniteTransactions txs = ignite0.transactions();
@@ -591,9 +607,16 @@ public class CacheSerializableTransactionsTest extends GridCommonAbstractTest {
keys.add(i);
try (Transaction tx = txs.txStart(OPTIMISTIC, SERIALIZABLE)) {
- Map<Integer, Integer> map = cache.getAll(keys);
+ if (needVer) {
+ Collection<CacheEntry<Integer, Integer>> c = cache.getEntries(keys);
- assertTrue(map.isEmpty());
+ assertTrue(c.isEmpty());
+ }
+ else {
+ Map<Integer, Integer> map = cache.getAll(keys);
+
+ assertTrue(map.isEmpty());
+ }
tx.commit();
}
@@ -602,9 +625,16 @@ public class CacheSerializableTransactionsTest extends GridCommonAbstractTest {
checkValue(key, null, cache.getName());
try (Transaction tx = txs.txStart(OPTIMISTIC, SERIALIZABLE)) {
- Map<Integer, Integer> map = cache.getAll(keys);
+ if (needVer) {
+ Collection<CacheEntry<Integer, Integer>> c = cache.getEntries(keys);
- assertTrue(map.isEmpty());
+ assertTrue(c.isEmpty());
+ }
+ else {
+ Map<Integer, Integer> map = cache.getAll(keys);
+
+ assertTrue(map.isEmpty());
+ }
tx.rollback();
}
@@ -653,21 +683,35 @@ public class CacheSerializableTransactionsTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testTxConflictRead1() throws Exception {
- txConflictRead(true);
+ txConflictRead(true, false);
}
/**
* @throws Exception If failed.
*/
public void testTxConflictRead2() throws Exception {
- txConflictRead(false);
+ txConflictRead(false, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxConflictReadEntry1() throws Exception {
+ txConflictRead(true, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxConflictReadEntry2() throws Exception {
+ txConflictRead(false, true);
}
/**
* @param noVal If {@code true} there is no cache value when read in tx.
* @throws Exception If failed.
*/
- private void txConflictRead(boolean noVal) throws Exception {
+ private void txConflictRead(boolean noVal, boolean needVer) throws Exception {
Ignite ignite0 = ignite(0);
final IgniteTransactions txs = ignite0.transactions();
@@ -693,9 +737,16 @@ public class CacheSerializableTransactionsTest extends GridCommonAbstractTest {
try {
try (Transaction tx = txs.txStart(OPTIMISTIC, SERIALIZABLE)) {
- Integer val = cache.get(key);
+ if (needVer) {
+ CacheEntry<Integer, Integer> val = cache.getEntry(key);
- assertEquals(expVal, val);
+ assertEquals(expVal, val == null ? null : val.getValue());
+ }
+ else {
+ Integer val = cache.get(key);
+
+ assertEquals(expVal, val);
+ }
updateKey(cache, key, 1);
@@ -711,9 +762,16 @@ public class CacheSerializableTransactionsTest extends GridCommonAbstractTest {
checkValue(key, 1, cache.getName());
try (Transaction tx = txs.txStart(OPTIMISTIC, SERIALIZABLE)) {
- Object val = cache.get(key);
+ if (needVer) {
+ CacheEntry<Integer, Integer> val = cache.getEntry(key);
- assertEquals(1, val);
+ assertEquals((Integer)1, val.getValue());
+ }
+ else {
+ Object val = cache.get(key);
+
+ assertEquals(1, val);
+ }
tx.commit();
}
@@ -731,28 +789,56 @@ public class CacheSerializableTransactionsTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testTxConflictReadWrite1() throws Exception {
- txConflictReadWrite(true, false);
+ txConflictReadWrite(true, false, false);
}
/**
* @throws Exception If failed.
*/
public void testTxConflictReadWrite2() throws Exception {
- txConflictReadWrite(false, false);
+ txConflictReadWrite(false, false, false);
}
/**
* @throws Exception If failed.
*/
public void testTxConflictReadRemove1() throws Exception {
- txConflictReadWrite(true, true);
+ txConflictReadWrite(true, true, false);
}
/**
* @throws Exception If failed.
*/
public void testTxConflictReadRemove2() throws Exception {
- txConflictReadWrite(false, true);
+ txConflictReadWrite(false, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxConflictReadEntryWrite1() throws Exception {
+ txConflictReadWrite(true, false, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxConflictReadEntryWrite2() throws Exception {
+ txConflictReadWrite(false, false, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxConflictReadEntryRemove1() throws Exception {
+ txConflictReadWrite(true, true, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxConflictReadEntryRemove2() throws Exception {
+ txConflictReadWrite(false, true, true);
}
/**
@@ -760,7 +846,7 @@ public class CacheSerializableTransactionsTest extends GridCommonAbstractTest {
* @param rmv If {@code true} tests remove, otherwise put.
* @throws Exception If failed.
*/
- private void txConflictReadWrite(boolean noVal, boolean rmv) throws Exception {
+ private void txConflictReadWrite(boolean noVal, boolean rmv, boolean needVer) throws Exception {
Ignite ignite0 = ignite(0);
final IgniteTransactions txs = ignite0.transactions();
@@ -786,9 +872,16 @@ public class CacheSerializableTransactionsTest extends GridCommonAbstractTest {
try {
try (Transaction tx = txs.txStart(OPTIMISTIC, SERIALIZABLE)) {
- Integer val = cache.get(key);
+ if (needVer) {
+ CacheEntry<Integer, Integer> val = cache.getEntry(key);
- assertEquals(expVal, val);
+ assertEquals(expVal, val == null ? null : val.getValue());
+ }
+ else {
+ Integer val = cache.get(key);
+
+ assertEquals(expVal, val);
+ }
updateKey(cache, key, 1);
@@ -809,9 +902,16 @@ public class CacheSerializableTransactionsTest extends GridCommonAbstractTest {
checkValue(key, 1, cache.getName());
try (Transaction tx = txs.txStart(OPTIMISTIC, SERIALIZABLE)) {
- Integer val = cache.get(key);
+ if (needVer) {
+ CacheEntry<Integer, Integer> val = cache.getEntry(key);
+
+ assertEquals(1, (Object)val.getValue());
+ }
+ else {
+ Integer val = cache.get(key);
- assertEquals(1, (Object) val);
+ assertEquals(1, (Object)val);
+ }
if (rmv)
cache.remove(key);
@@ -4239,7 +4339,7 @@ public class CacheSerializableTransactionsTest extends GridCommonAbstractTest {
ignite.destroyCache(cacheName);
}
catch (IgniteException ignore) {
- // No-op.
+ // No-op.
}
GridTestSwapSpaceSpi spi = (GridTestSwapSpaceSpi)ignite.configuration().getSwapSpaceSpi();
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
index 93ff515..3bd87cf 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
@@ -53,6 +53,7 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteTransactions;
+import org.apache.ignite.cache.CacheEntry;
import org.apache.ignite.cache.CacheEntryProcessor;
import org.apache.ignite.cache.CacheMemoryMode;
import org.apache.ignite.cache.CachePeekMode;
@@ -556,6 +557,30 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
/**
* @throws Exception In case of error.
*/
+ public void testGetEntry() throws Exception {
+ IgniteCache<String, Integer> cache = jcache();
+
+ cache.put("key1", 1);
+ cache.put("key2", 2);
+
+ CacheEntry<String, Integer> key1e = cache.getEntry("key1");
+ CacheEntry<String, Integer> key2e = cache.getEntry("key2");
+ CacheEntry<String, Integer> wrongKeye = cache.getEntry("wrongKey");
+
+ assert key1e.getValue() == 1;
+ assert key1e.getKey().equals("key1");
+ assert key1e.version() != null;
+
+ assert key2e.getValue() == 2;
+ assert key2e.getKey().equals("key2");
+ assert key2e.version() != null;
+
+ assert wrongKeye == null;
+ }
+
+ /**
+ * @throws Exception In case of error.
+ */
public void testGetAsync() throws Exception {
IgniteCache<String, Integer> cache = jcache();
@@ -664,6 +689,122 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
/**
* @throws Exception In case of error.
*/
+ public void testGetEntries() throws Exception {
+ Transaction tx = txShouldBeUsed() ? transactions().txStart() : null;
+
+ final IgniteCache<String, Integer> cache = jcache();
+
+ try {
+ cache.put("key1", 1);
+ cache.put("key2", 2);
+
+ if (tx != null)
+ tx.commit();
+ }
+ finally {
+ if (tx != null)
+ tx.close();
+ }
+
+ GridTestUtils.assertThrows(log, new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ cache.getEntries(null).isEmpty();
+
+ return null;
+ }
+ }, NullPointerException.class, null);
+
+ assert cache.getEntries(Collections.<String>emptySet()).isEmpty();
+
+ Collection<CacheEntry<String, Integer>> c1 = cache.getEntries(ImmutableSet.of("key1", "key2", "key9999"));
+
+ info("Retrieved c1: " + c1);
+
+ assert 2 == c1.size() : "Invalid collection: " + c1;
+
+ boolean b1 = false;
+ boolean b2 = false;
+
+ for (CacheEntry<String, Integer> e: c1){
+ if (e.getKey().equals("key1") && e.getValue().equals(1))
+ b1 = true;
+
+ if (e.getKey().equals("key2") && e.getValue().equals(2))
+ b2 = true;
+ }
+
+ assertTrue(b1 && b2);
+
+ Collection<CacheEntry<String, Integer>> c2 = cache.getEntries(ImmutableSet.of("key1", "key2", "key9999"));
+
+ info("Retrieved c2: " + c2);
+
+ assert 2 == c2.size() : "Invalid collection: " + c2;
+
+ b1 = false;
+ b2 = false;
+
+ for (CacheEntry<String, Integer> e: c2){
+ if (e.getKey().equals("key1") && e.getValue().equals(1))
+ b1 = true;
+
+ if (e.getKey().equals("key2") && e.getValue().equals(2))
+ b2 = true;
+ }
+
+ assertTrue(b1 && b2);
+
+ // Now do the same checks but within transaction.
+ if (txShouldBeUsed()) {
+ try (Transaction tx0 = transactions().txStart()) {
+ assert cache.getEntries(Collections.<String>emptySet()).isEmpty();
+
+ c1 = cache.getEntries(ImmutableSet.of("key1", "key2", "key9999"));
+
+ info("Retrieved c1: " + c1);
+
+ assert 2 == c1.size() : "Invalid collection: " + c1;
+
+ b1 = false;
+ b2 = false;
+
+ for (CacheEntry<String, Integer> e : c1) {
+ if (e.getKey().equals("key1") && e.getValue().equals(1))
+ b1 = true;
+
+ if (e.getKey().equals("key2") && e.getValue().equals(2))
+ b2 = true;
+ }
+
+ assertTrue(b1 && b2);
+
+ c2 = cache.getEntries(ImmutableSet.of("key1", "key2", "key9999"));
+
+ info("Retrieved c2: " + c2);
+
+ assert 2 == c2.size() : "Invalid collection: " + c2;
+
+ b1 = false;
+ b2 = false;
+
+ for (CacheEntry<String, Integer> e : c2) {
+ if (e.getKey().equals("key1") && e.getValue().equals(1))
+ b1 = true;
+
+ if (e.getKey().equals("key2") && e.getValue().equals(2))
+ b2 = true;
+ }
+
+ assertTrue(b1 && b2);
+
+ tx0.commit();
+ }
+ }
+ }
+
+ /**
+ * @throws Exception In case of error.
+ */
public void testGetAllWithNulls() throws Exception {
final IgniteCache<String, Integer> cache = jcache();
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheInterceptorAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheInterceptorAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheInterceptorAbstractSelfTest.java
index c57869f..19c8639 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheInterceptorAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheInterceptorAbstractSelfTest.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.cache;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
@@ -29,6 +30,7 @@ import javax.cache.processor.EntryProcessor;
import javax.cache.processor.MutableEntry;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheAtomicWriteOrderMode;
+import org.apache.ignite.cache.CacheEntry;
import org.apache.ignite.cache.CacheInterceptor;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.affinity.Affinity;
@@ -152,19 +154,32 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
* @throws Exception If failed.
*/
public void testGet() throws Exception {
- testGet(primaryKey(0));
+ testGet(primaryKey(0), false);
afterTest();
if (cacheMode() != LOCAL)
- testGet(backupKey(0));
+ testGet(backupKey(0), false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetEntry() throws Exception {
+ testGet(primaryKey(0), true);
+
+ afterTest();
+
+ if (cacheMode() != LOCAL)
+ testGet(backupKey(0), true);
}
/**
* @param key Key.
+ * @param needVer Need version.
* @throws Exception If failed.
*/
- private void testGet(String key) throws Exception {
+ private void testGet(String key, boolean needVer) throws Exception {
// Try when value is not in cache.
interceptor.retInterceptor = new NullGetInterceptor();
@@ -173,7 +188,7 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
IgniteCache<String, Integer> cache = jcache(0);
- assertEquals(null, cache.get(key));
+ assertEquals(null, needVer ? cache.getEntry(key).getValue() : cache.get(key));
assertEquals(1, interceptor.invokeCnt.get());
@@ -185,7 +200,7 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
log.info("Get 2.");
- assertEquals((Integer)1, cache.get(key));
+ assertEquals((Integer)1, needVer ? cache.getEntry(key).getValue() : cache.get(key));
assertEquals(1, interceptor.invokeCnt.get());
@@ -207,7 +222,7 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
log.info("Get 3.");
- assertEquals(null, cache.get(key));
+ assertEquals(null, needVer ? cache.getEntry(key).getValue() : cache.get(key));
assertEquals(1, interceptor.invokeCnt.get());
@@ -223,7 +238,7 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
log.info("Get 4.");
- assertEquals((Integer)101, cache.get(key));
+ assertEquals((Integer)101, needVer ? cache.getEntry(key).getValue() : cache.get(key));
assertEquals(1, interceptor.invokeCnt.get());
@@ -241,9 +256,16 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
IgniteCache<String, Integer> cacheAsync = cache.withAsync();
- cacheAsync.get(key);
+ if (needVer) {
+ cacheAsync.getEntry(key);
- assertEquals((Integer)101, cacheAsync.<Integer>future().get());
+ assertEquals((Integer)101, cacheAsync.<CacheEntry<String, Integer>>future().get().getValue());
+ }
+ else {
+ cacheAsync.get(key);
+
+ assertEquals((Integer)101, cacheAsync.<Integer>future().get());
+ }
assertEquals(1, interceptor.invokeCnt.get());
@@ -258,6 +280,20 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
* @throws Exception If failed.
*/
public void testGetAll() throws Exception {
+ testGetAll(false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetEntries() throws Exception {
+ testGetAll(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetAll(boolean needVer) throws Exception {
Set<String> keys = new LinkedHashSet<>();
for (int i = 0; i < 1000; i++)
@@ -269,10 +305,19 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
IgniteCache<String, Integer> cacheAsync = cache.withAsync();
- Map<String, Integer> map = cache.getAll(keys);
+ Collection<CacheEntry<String, Integer>> c;
+ Map<String, Integer> map;
+
+ if (needVer){
+ c = cache.getEntries(keys);
- for (String key : keys)
- assertEquals(null, map.get(key));
+ assertTrue(c.isEmpty());
+ }else {
+ map = cache.getAll(keys);
+
+ for (String key : keys)
+ assertEquals(null, map.get(key));
+ }
assertEquals(1000, interceptor.invokeCnt.get());
@@ -280,15 +325,28 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
interceptor.retInterceptor = new GetAllInterceptor1();
- map = cache.getAll(keys);
+ if (needVer) {
+ c = cache.getEntries(keys);
- for (String key : keys) {
- int k = Integer.valueOf(key);
+ assertEquals(500, c.size());
- if (k % 2 == 0)
- assertEquals(null, map.get(key));
- else
- assertEquals((Integer)(k * 2), map.get(key));
+ for (CacheEntry<String, Integer> e : c) {
+ int k = Integer.valueOf(e.getKey());
+
+ assertEquals((Integer)(k * 2), e.getValue());
+ }
+ }
+ else {
+ map = cache.getAll(keys);
+
+ for (String key : keys) {
+ int k = Integer.valueOf(key);
+
+ if (k % 2 == 0)
+ assertEquals(null, map.get(key));
+ else
+ assertEquals((Integer)(k * 2), map.get(key));
+ }
}
assertEquals(1000, interceptor.invokeCnt.get());
@@ -307,40 +365,72 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
interceptor.retInterceptor = new GetAllInterceptor2();
- if (j == 0)
- map = cache.getAll(keys);
- else {
- cacheAsync.getAll(keys);
+ if (needVer) {
+ if (j == 0)
+ c = cache.getEntries(keys);
+ else {
+ cacheAsync.getEntries(keys);
+
+ c = cacheAsync.<Collection<CacheEntry<String, Integer>>>future().get();
+ }
+
+ for (CacheEntry<String, Integer> e : c) {
+ int k = Integer.valueOf(e.getKey());
+
+ switch (k % 3) {
+ case 1:
+ Integer exp = k < 500 ? k : null;
- map = cacheAsync.<Map<String, Integer>>future().get();
+ assertEquals(exp, e.getValue());
+
+ break;
+
+ case 2:
+ assertEquals((Integer)(k * 3), e.getValue());
+
+ break;
+
+ default:
+ fail();
+ }
+ }
}
+ else {
+ if (j == 0)
+ map = cache.getAll(keys);
+ else {
+ cacheAsync.getAll(keys);
- int i = 0;
+ map = cacheAsync.<Map<String, Integer>>future().get();
+ }
- for (String key : keys) {
- switch (i % 3) {
- case 0:
- assertEquals(null, map.get(key));
+ int i = 0;
- break;
+ for (String key : keys) {
+ switch (i % 3) {
+ case 0:
+ assertEquals(null, map.get(key));
- case 1:
- Integer exp = i < 500 ? i : null;
+ break;
- assertEquals(exp, map.get(key));
+ case 1:
+ Integer exp = i < 500 ? i : null;
- break;
+ assertEquals(exp, map.get(key));
- case 2:
- assertEquals((Integer)(i * 3), map.get(key));
+ break;
- break;
+ case 2:
+ assertEquals((Integer)(i * 3), map.get(key));
- default:
- fail();
- }
+ break;
- i++;
+ default:
+ fail();
+ }
+
+ i++;
+ }
}
assertEquals(1000, interceptor.invokeCnt.get());
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtEvictionNearReadersSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtEvictionNearReadersSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtEvictionNearReadersSelfTest.java
index 6a7416e..293ba1e 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtEvictionNearReadersSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtEvictionNearReadersSelfTest.java
@@ -263,7 +263,7 @@ public class GridCacheDhtEvictionNearReadersSelfTest extends GridCommonAbstractT
waitForLocalEvent(grid(primary).events(), nodeEvent(primary.id()), EVT_CACHE_ENTRY_EVICTED);
// Get value on other node, it should be loaded to near cache.
- assertEquals(val, nearOther.get(key, true));
+ assertEquals(val, nearOther.get(key, true, false));
entryPrimary = (GridDhtCacheEntry)dhtPrimary.peekEx(key);
entryBackup = (GridDhtCacheEntry)dhtBackup.peekEx(key);
[3/3] ignite git commit: 2224
Posted by av...@apache.org.
2224
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/b28f63e4
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/b28f63e4
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/b28f63e4
Branch: refs/heads/ignite-2224
Commit: b28f63e467a3734ba208d978ee16b062c6475fef
Parents: d2a107b
Author: Anton Vinogradov <av...@apache.org>
Authored: Fri Jan 29 15:46:15 2016 +0300
Committer: Anton Vinogradov <av...@apache.org>
Committed: Fri Jan 29 15:46:15 2016 +0300
----------------------------------------------------------------------
.../java/org/apache/ignite/IgniteCache.java | 45 +-
.../processors/cache/GridCacheAdapter.java | 297 +++++++++--
.../processors/cache/GridCacheContext.java | 33 +-
.../processors/cache/GridCacheMapEntry.java | 2 +-
.../processors/cache/GridCacheProxyImpl.java | 51 ++
.../processors/cache/IgniteCacheProxy.java | 51 ++
.../processors/cache/IgniteInternalCache.java | 85 +++
.../dht/CacheDistributedGetFutureAdapter.java | 15 -
.../distributed/dht/GridDhtCacheAdapter.java | 7 +-
.../cache/distributed/dht/GridDhtGetFuture.java | 6 +-
.../dht/GridPartitionedGetFuture.java | 38 +-
.../dht/GridPartitionedSingleGetFuture.java | 17 +-
.../dht/atomic/GridDhtAtomicCache.java | 83 ++-
.../dht/colocated/GridDhtColocatedCache.java | 42 +-
.../distributed/near/GridNearAtomicCache.java | 6 +-
.../distributed/near/GridNearCacheAdapter.java | 6 +-
.../distributed/near/GridNearCacheEntry.java | 3 +-
.../distributed/near/GridNearGetFuture.java | 45 +-
.../near/GridNearTransactionalCache.java | 9 +-
.../local/atomic/GridLocalAtomicCache.java | 97 +++-
.../transactions/IgniteTxLocalAdapter.java | 172 ++++--
.../cache/transactions/IgniteTxLocalEx.java | 3 +-
.../cache/CacheGetEntryAbstractSeltTest.java | 525 +++++++++++++++++++
...GetEntryOptimisticReadCommittedSeltTest.java | 35 ++
...etEntryOptimisticRepeatableReadSeltTest.java | 35 ++
...eGetEntryOptimisticSerializableSeltTest.java | 35 ++
...etEntryPessimisticReadCommittedSeltTest.java | 35 ++
...tEntryPessimisticRepeatableReadSeltTest.java | 35 ++
...GetEntryPessimisticSerializableSeltTest.java | 35 ++
.../cache/CacheReadThroughRestartSelfTest.java | 43 +-
.../CacheSerializableTransactionsTest.java | 142 ++++-
.../cache/GridCacheAbstractFullApiSelfTest.java | 141 +++++
.../GridCacheInterceptorAbstractSelfTest.java | 172 ++++--
...GridCacheDhtEvictionNearReadersSelfTest.java | 2 +-
.../multijvm/IgniteCacheProcessProxy.java | 59 ++-
.../testsuites/IgniteCacheTestSuite4.java | 12 +
36 files changed, 2090 insertions(+), 329 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/IgniteCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteCache.java b/modules/core/src/main/java/org/apache/ignite/IgniteCache.java
index 886dca6..a791e38 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteCache.java
@@ -31,10 +31,12 @@ import javax.cache.CacheException;
import javax.cache.configuration.Configuration;
import javax.cache.event.CacheEntryRemovedListener;
import javax.cache.expiry.ExpiryPolicy;
+import javax.cache.integration.CacheLoader;
import javax.cache.integration.CacheWriter;
import javax.cache.processor.EntryProcessor;
import javax.cache.processor.EntryProcessorException;
import javax.cache.processor.EntryProcessorResult;
+import org.apache.ignite.cache.CacheEntry;
import org.apache.ignite.cache.CacheEntryProcessor;
import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CacheMode;
@@ -390,18 +392,59 @@ public interface IgniteCache<K, V> extends javax.cache.Cache<K, V>, IgniteAsyncS
* <code>null</code> value for a key.
*/
@IgniteAsyncSupported
- <T> Map<K, EntryProcessorResult<T>> invokeAll(Map<? extends K, ? extends EntryProcessor<K, V, T>> map,
+ public <T> Map<K, EntryProcessorResult<T>> invokeAll(Map<? extends K, ? extends EntryProcessor<K, V, T>> map,
Object... args);
/** {@inheritDoc} */
@IgniteAsyncSupported
@Override public V get(K key);
+ /**
+ * Gets an entry from the cache.
+ * <p>
+ * If the cache is configured to use read-through, and get would return null
+ * because the entry is missing from the cache, the Cache's {@link CacheLoader}
+ * is called in an attempt to load the entry.
+ *
+ * @param key the key whose associated value is to be returned
+ * @return the element, or null, if it does not exist.
+ * @throws IllegalStateException if the cache is {@link #isClosed()}
+ * @throws NullPointerException if the key is null
+ * @throws CacheException if there is a problem fetching the value
+ * @throws ClassCastException if the implementation is configured to perform
+ * runtime-type-checking, and the key or value types are incompatible with those that have been
+ * configured for the {@link Cache}
+ */
+ @IgniteAsyncSupported
+ public CacheEntry<K, V> getEntry(K key);
+
/** {@inheritDoc} */
@IgniteAsyncSupported
@Override public Map<K, V> getAll(Set<? extends K> keys);
/**
+ * Gets a collection of entries from the {@link Cache}.
+ * <p>
+ * If the cache is configured read-through, and a get for a key would
+ * return null because an entry is missing from the cache, the Cache's
+ * {@link CacheLoader} is called in an attempt to load the entry. If an
+ * entry cannot be loaded for a given key, the key will not be present in
+ * the returned Collection.
+ *
+ * @param keys The keys whose associated values are to be returned.
+ * @return A collection of entries that were found for the given keys. Entries not found
+ * in the cache are not in the returned collection.
+ * @throws NullPointerException if keys is null or if keys contains a null
+ * @throws IllegalStateException if the cache is {@link #isClosed()}
+ * @throws CacheException if there is a problem fetching the values
+ * @throws ClassCastException if the implementation is configured to perform
+ * runtime-type-checking, and the key or value types are incompatible with those that have been
+ * configured for the {@link Cache}
+ */
+ @IgniteAsyncSupported
+ public Collection<CacheEntry<K, V>> getEntries(Set<? extends K> keys);
+
+ /**
* Gets values from cache. Will bypass started transaction, if any, i.e. will not enlist entries
* and will not lock any keys if pessimistic transaction is started by thread.
*
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index 3081cfb..1334554 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -52,6 +52,7 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.cache.CacheEntry;
import org.apache.ignite.cache.CacheInterceptor;
import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CachePeekMode;
@@ -607,7 +608,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/*task name*/null,
/*deserialize binary*/false,
/*skip values*/true,
- /*can remap*/true
+ /*can remap*/true,
+ false
);
}
@@ -633,7 +635,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/*task name*/null,
/*deserialize binary*/false,
/*skip values*/true,
- /*can remap*/true
+ /*can remap*/true,
+ false
).chain(new CX1<IgniteInternalFuture<Map<K, V>>, Boolean>() {
@Override public Boolean applyx(IgniteInternalFuture<Map<K, V>> fut) throws IgniteCheckedException {
Map<K, V> kvMap = fut.get();
@@ -1296,7 +1299,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
/*deserialize cache objects*/true,
/*skip values*/false,
- /*can remap*/true
+ /*can remap*/true,
+ false
).get().get(key);
}
@@ -1312,7 +1316,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
true,
false,
- /*can remap*/true
+ /*can remap*/true,
+ false
).chain(new CX1<IgniteInternalFuture<Map<K, V>>, V>() {
@Override public V applyx(IgniteInternalFuture<Map<K, V>> e) throws IgniteCheckedException {
return e.get().get(key);
@@ -1332,7 +1337,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
/*deserialize cache objects*/true,
/*skip values*/false,
- /*can remap*/false
+ /*can remap*/false,
+ false
).get().get(key);
}
@@ -1352,7 +1358,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
!ctx.keepBinary(),
/*skip values*/false,
- /*can remap*/true);
+ /*can remap*/true,
+ false);
}
/**
@@ -1372,7 +1379,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
long start = statsEnabled ? System.nanoTime() : 0L;
- V val = get(key, !ctx.keepBinary());
+ V val = get(key, !ctx.keepBinary(), false);
if (ctx.config().getInterceptor() != null)
val = (V)ctx.config().getInterceptor().onGet(key, val);
@@ -1384,6 +1391,30 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
/** {@inheritDoc} */
+ @Nullable @Override public CacheEntry<K, V> getEntry(K key) throws IgniteCheckedException {
+ A.notNull(key, "key");
+
+ boolean statsEnabled = ctx.config().isStatisticsEnabled();
+
+ long start = statsEnabled ? System.nanoTime() : 0L;
+
+ T2<V, GridCacheVersion> t = (T2<V, GridCacheVersion>)get(key, !ctx.keepBinary(), true);
+
+ CacheEntry<K, V> val = t != null ? new CacheEntryImplEx<>(key, t.get1(), t.get2()): null;
+
+ if (ctx.config().getInterceptor() != null) {
+ V val0 = (V)ctx.config().getInterceptor().onGet(key, t != null ? val.getValue() : null);
+
+ val = new CacheEntryImplEx<>(key, val0, t != null ? t.get2() : null);
+ }
+
+ if (statsEnabled)
+ metrics0().addGetTimeNanos(System.nanoTime() - start);
+
+ return val;
+ }
+
+ /** {@inheritDoc} */
@Override public IgniteInternalFuture<V> getAsync(final K key) {
A.notNull(key, "key");
@@ -1391,7 +1422,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final long start = statsEnabled ? System.nanoTime() : 0L;
- IgniteInternalFuture<V> fut = getAsync(key, !ctx.keepBinary());
+ IgniteInternalFuture<V> fut = getAsync(key, !ctx.keepBinary(), false);
if (ctx.config().getInterceptor() != null)
fut = fut.chain(new CX1<IgniteInternalFuture<V>, V>() {
@@ -1407,6 +1438,42 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
/** {@inheritDoc} */
+ @Override public IgniteInternalFuture<CacheEntry<K, V>> getEntryAsync(final K key) {
+ A.notNull(key, "key");
+
+ final boolean statsEnabled = ctx.config().isStatisticsEnabled();
+
+ final long start = statsEnabled ? System.nanoTime() : 0L;
+
+ IgniteInternalFuture<T2<V, GridCacheVersion>> fut =
+ (IgniteInternalFuture<T2<V, GridCacheVersion>>)getAsync(key, !ctx.keepBinary(), true);
+
+ final boolean intercept = ctx.config().getInterceptor() != null;
+
+ IgniteInternalFuture<CacheEntry<K, V>> fr = fut.chain(
+ new CX1<IgniteInternalFuture<T2<V, GridCacheVersion>>, CacheEntry<K, V>>() {
+ @Override public CacheEntry<K, V> applyx(IgniteInternalFuture<T2<V, GridCacheVersion>> f)
+ throws IgniteCheckedException {
+ T2<V, GridCacheVersion> t = f.get();
+
+ CacheEntry<K, V> val = t != null ? new CacheEntryImplEx<>(key, t.get1(), t.get2()) : null;
+ if (intercept) {
+ V val0 = (V)ctx.config().getInterceptor().onGet(key, t != null ? val.getValue() : null);
+
+ return new CacheEntryImplEx<>(key, val0, t != null ? t.get2() : null);
+ }
+ else
+ return val;
+ }
+ });
+
+ if (statsEnabled)
+ fut.listen(new UpdateGetTimeStatClosure<T2<V, GridCacheVersion>>(metrics0(), start));
+
+ return fr;
+ }
+
+ /** {@inheritDoc} */
@Override public Map<K, V> getAll(@Nullable Collection<? extends K> keys) throws IgniteCheckedException {
A.notNull(keys, "keys");
@@ -1414,7 +1481,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
long start = statsEnabled ? System.nanoTime() : 0L;
- Map<K, V> map = getAll(keys, !ctx.keepBinary());
+ Map<K, V> map = getAll(keys, !ctx.keepBinary(), false);
if (ctx.config().getInterceptor() != null)
map = interceptGet(keys, map);
@@ -1426,6 +1493,32 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
/** {@inheritDoc} */
+ @Override public Collection<CacheEntry<K, V>> getEntries(@Nullable Collection<? extends K> keys)
+ throws IgniteCheckedException {
+ A.notNull(keys, "keys");
+
+ boolean statsEnabled = ctx.config().isStatisticsEnabled();
+
+ long start = statsEnabled ? System.nanoTime() : 0L;
+
+ Map<K, T2<V, GridCacheVersion>> map = (Map<K, T2<V, GridCacheVersion>>)getAll(keys, !ctx.keepBinary(), true);
+
+ Collection<CacheEntry<K, V>> res = new HashSet<>();
+
+ if (ctx.config().getInterceptor() != null)
+ res = interceptGetEntries(keys, map);
+ else
+ for (Map.Entry<K, T2<V, GridCacheVersion>> e : map.entrySet())
+ res.add(new CacheEntryImplEx<>(e.getKey(), e.getValue().get1(), e.getValue().get2()));
+
+ if (statsEnabled)
+ metrics0().addGetTimeNanos(System.nanoTime() - start);
+
+ return res;
+ }
+
+
+ /** {@inheritDoc} */
@Override public IgniteInternalFuture<Map<K, V>> getAllAsync(@Nullable final Collection<? extends K> keys) {
A.notNull(keys, "keys");
@@ -1433,7 +1526,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final long start = statsEnabled ? System.nanoTime() : 0L;
- IgniteInternalFuture<Map<K, V>> fut = getAllAsync(keys, !ctx.keepBinary());
+ IgniteInternalFuture<Map<K, V>> fut = getAllAsync(keys, !ctx.keepBinary(), false);
if (ctx.config().getInterceptor() != null)
return fut.chain(new CX1<IgniteInternalFuture<Map<K, V>>, Map<K, V>>() {
@@ -1448,6 +1541,45 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
return fut;
}
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<Collection<CacheEntry<K, V>>> getEntriesAsync(
+ @Nullable final Collection<? extends K> keys) {
+ A.notNull(keys, "keys");
+
+ final boolean statsEnabled = ctx.config().isStatisticsEnabled();
+
+ final long start = statsEnabled ? System.nanoTime() : 0L;
+
+ IgniteInternalFuture<Map<K, T2<V, GridCacheVersion>>> fut =
+ (IgniteInternalFuture<Map<K, T2<V, GridCacheVersion>>>)
+ ((IgniteInternalFuture)getAllAsync(keys, !ctx.keepBinary(), true));
+
+ final boolean intercept = ctx.config().getInterceptor() != null;
+
+ IgniteInternalFuture<Collection<CacheEntry<K, V>>> rf =
+ fut.chain(new CX1<IgniteInternalFuture<Map<K, T2<V, GridCacheVersion>>>, Collection<CacheEntry<K, V>>>() {
+ @Override public Collection<CacheEntry<K, V>> applyx(
+ IgniteInternalFuture<Map<K, T2<V, GridCacheVersion>>> f) throws IgniteCheckedException {
+ if (intercept)
+ return interceptGetEntries(keys, f.get());
+ else {
+ Map<K, CacheEntry<K, V>> res = U.newHashMap(f.get().size());
+
+ for (Map.Entry<K, T2<V, GridCacheVersion>> e : f.get().entrySet())
+ res.put(e.getKey(),
+ new CacheEntryImplEx<>(e.getKey(), e.getValue().get1(), e.getValue().get2()));
+
+ return res.values();
+ }
+ }
+ });
+
+ if (statsEnabled)
+ fut.listen(new UpdateGetTimeStatClosure<Map<K, T2<V, GridCacheVersion>>>(metrics0(), start));
+
+ return rf;
+ }
+
/**
* Applies cache interceptor on result of 'get' operation.
*
@@ -1490,6 +1622,53 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
/**
+ * Applies cache interceptor on result of 'getEntries' operation.
+ *
+ * @param keys All requested keys.
+ * @param map Result map.
+ * @return Map with values returned by cache interceptor..
+ */
+ @SuppressWarnings("IfMayBeConditional")
+ private Collection<CacheEntry<K, V>> interceptGetEntries(
+ @Nullable Collection<? extends K> keys, Map<K, T2<V, GridCacheVersion>> map) {
+ Map<K, CacheEntry<K, V>> res;
+
+ if (F.isEmpty(keys)) {
+ assert map.isEmpty();
+
+ return Collections.emptySet();
+ }
+
+ res = U.newHashMap(keys.size());
+
+ CacheInterceptor<K, V> interceptor = cacheCfg.getInterceptor();
+
+ assert interceptor != null;
+
+ for (Map.Entry<K, T2<V, GridCacheVersion>> e : map.entrySet()) {
+ V val = interceptor.onGet(e.getKey(), e.getValue().get1());
+
+ if (val != null)
+ res.put(e.getKey(), new CacheEntryImplEx<>(e.getKey(), val, e.getValue().get2()));
+ }
+
+ if (map.size() != keys.size()) { // Not all requested keys were in cache.
+ for (K key : keys) {
+ if (key != null) {
+ if (!map.containsKey(key)) {
+ V val = interceptor.onGet(key, null);
+
+ if (val != null)
+ res.put(key, new CacheEntryImplEx<>(key, val, null));
+ }
+ }
+ }
+ }
+
+ return res.values();
+ }
+
+ /**
* @param key Key.
* @param forcePrimary Force primary.
* @param skipTx Skip tx.
@@ -1498,6 +1677,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param deserializeBinary Deserialize binary.
* @param skipVals Skip values.
* @param canRemap Can remap flag.
+ * @param needVer Need version.
* @return Future for the get operation.
*/
protected IgniteInternalFuture<V> getAsync(
@@ -1508,7 +1688,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
String taskName,
boolean deserializeBinary,
final boolean skipVals,
- boolean canRemap
+ boolean canRemap,
+ final boolean needVer
) {
return getAllAsync(Collections.singletonList(key),
forcePrimary,
@@ -1517,7 +1698,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
deserializeBinary,
skipVals,
- canRemap).chain(
+ canRemap,
+ needVer).chain(
new CX1<IgniteInternalFuture<Map<K, V>>, V>() {
@Override public V applyx(IgniteInternalFuture<Map<K, V>> e) throws IgniteCheckedException {
Map<K, V> map = e.get();
@@ -1544,6 +1726,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param deserializeBinary Deserialize binary.
* @param skipVals Skip values.
* @param canRemap Can remap flag.
+ * @param needVer Need version.
* @return Future for the get operation.
* @see GridCacheAdapter#getAllAsync(Collection)
*/
@@ -1555,7 +1738,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
String taskName,
boolean deserializeBinary,
boolean skipVals,
- boolean canRemap
+ boolean canRemap,
+ final boolean needVer
) {
CacheOperationContext opCtx = ctx.operationContextPerCall();
@@ -1570,7 +1754,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
forcePrimary,
skipVals ? null : expiryPolicy(opCtx != null ? opCtx.expiry() : null),
skipVals,
- canRemap);
+ canRemap,
+ needVer);
}
/**
@@ -1584,6 +1769,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param expiry Expiry policy.
* @param skipVals Skip values.
* @param canRemap Can remap flag.
+ * @param needVer Need version.
* @return Future for the get operation.
* @see GridCacheAdapter#getAllAsync(Collection)
*/
@@ -1596,7 +1782,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final boolean forcePrimary,
@Nullable IgniteCacheExpiryPolicy expiry,
final boolean skipVals,
- boolean canRemap
+ boolean canRemap,
+ final boolean needVer
) {
ctx.checkSecurity(SecurityPermission.CACHE_READ);
@@ -1613,7 +1800,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
skipVals,
false,
canRemap,
- false);
+ needVer);
}
/**
@@ -1708,20 +1895,14 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
ctx.evicts().touch(entry, topVer);
}
else {
- if (needVer) {
- assert keepCacheObjects;
-
- map.put((K1)key, (V1)new T2<>(res.get1(), res.get2()));
- }
- else {
- ctx.addResult(map,
- key,
- res.get1(),
- skipVals,
- keepCacheObjects,
- deserializeBinary,
- true);
- }
+ ctx.addResult(map,
+ key,
+ res.get1(),
+ skipVals,
+ keepCacheObjects,
+ deserializeBinary,
+ true,
+ needVer ? res.get2() : null);
if (tx == null || (!tx.implicit() && tx.isolation() == READ_COMMITTED))
ctx.evicts().touch(entry, topVer);
@@ -1783,20 +1964,14 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
// Don't put key-value pair into result map if value is null.
if (val != null) {
- if (needVer) {
- assert keepCacheObjects;
-
- map.put((K1)key, (V1)new T2<>(cacheVal, set ? verSet : ver));
- }
- else {
- ctx.addResult(map,
- key,
- cacheVal,
- skipVals,
- keepCacheObjects,
- deserializeBinary,
- false);
- }
+ ctx.addResult(map,
+ key,
+ cacheVal,
+ skipVals,
+ keepCacheObjects,
+ deserializeBinary,
+ false,
+ needVer ? set ? verSet : ver : null);
}
if (tx0 == null || (!tx0.implicit() &&
@@ -1889,11 +2064,9 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
}
else {
- assert !needVer;
-
return asyncOp(tx, new AsyncOp<Map<K1, V1>>(keys) {
@Override public IgniteInternalFuture<Map<K1, V1>> op(IgniteTxLocalAdapter tx) {
- return tx.getAllAsync(ctx, keys, deserializeBinary, skipVals, false, !readThrough);
+ return tx.getAllAsync(ctx, keys, deserializeBinary, skipVals, false, !readThrough, needVer);
}
}, ctx.operationContextPerCall());
}
@@ -4494,28 +4667,31 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/**
* @param key Key.
* @param deserializeBinary Deserialize binary flag.
+ * @param needVer Need version.
* @return Cached value.
* @throws IgniteCheckedException If failed.
*/
- @Nullable public V get(K key, boolean deserializeBinary) throws IgniteCheckedException {
+ @Nullable public V get(K key, boolean deserializeBinary, final boolean needVer) throws IgniteCheckedException {
checkJta();
String taskName = ctx.kernalContext().job().currentTaskName();
- return get(key, taskName, deserializeBinary);
+ return get(key, taskName, deserializeBinary, needVer);
}
/**
* @param key Key.
* @param taskName Task name.
* @param deserializeBinary Deserialize binary flag.
+ * @param needVer Need version.
* @return Cached value.
* @throws IgniteCheckedException If failed.
*/
protected V get(
final K key,
String taskName,
- boolean deserializeBinary) throws IgniteCheckedException {
+ boolean deserializeBinary,
+ boolean needVer) throws IgniteCheckedException {
return getAsync(key,
!ctx.config().isReadFromBackup(),
/*skip tx*/false,
@@ -4523,15 +4699,17 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
deserializeBinary,
false,
- /*can remap*/true).get();
+ /*can remap*/true,
+ needVer).get();
}
/**
* @param key Key.
* @param deserializeBinary Deserialize binary flag.
+ * @param needVer Need version.
* @return Read operation future.
*/
- public final IgniteInternalFuture<V> getAsync(final K key, boolean deserializeBinary) {
+ public final IgniteInternalFuture<V> getAsync(final K key, boolean deserializeBinary, final boolean needVer) {
try {
checkJta();
}
@@ -4548,28 +4726,32 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
deserializeBinary,
false,
- /*can remap*/true);
+ /*can remap*/true,
+ needVer);
}
/**
* @param keys Keys.
* @param deserializeBinary Deserialize binary flag.
+ * @param needVer Need version.
* @return Map of cached values.
* @throws IgniteCheckedException If read failed.
*/
- public Map<K, V> getAll(Collection<? extends K> keys, boolean deserializeBinary) throws IgniteCheckedException {
+ public Map<K, V> getAll(Collection<? extends K> keys, boolean deserializeBinary,
+ boolean needVer) throws IgniteCheckedException {
checkJta();
- return getAllAsync(keys, deserializeBinary).get();
+ return getAllAsync(keys, deserializeBinary, needVer).get();
}
/**
* @param keys Keys.
* @param deserializeBinary Deserialize binary flag.
+ * @param needVer Need version.
* @return Read future.
*/
public IgniteInternalFuture<Map<K, V>> getAllAsync(@Nullable Collection<? extends K> keys,
- boolean deserializeBinary) {
+ boolean deserializeBinary, boolean needVer) {
String taskName = ctx.kernalContext().job().currentTaskName();
return getAllAsync(keys,
@@ -4579,7 +4761,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
taskName,
deserializeBinary,
/*skip vals*/false,
- /*can remap*/true);
+ /*can remap*/true,
+ needVer);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
index fc48b9d..e875df0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
@@ -49,12 +49,12 @@ import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.IgnitionEx;
+import org.apache.ignite.internal.binary.BinaryMarshaller;
import org.apache.ignite.internal.managers.communication.GridIoManager;
import org.apache.ignite.internal.managers.deployment.GridDeploymentManager;
import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager;
import org.apache.ignite.internal.managers.swapspace.GridSwapSpaceManager;
-import org.apache.ignite.internal.binary.BinaryMarshaller;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.datastructures.CacheDataStructuresManager;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
@@ -91,6 +91,7 @@ import org.apache.ignite.internal.util.offheap.unsafe.GridUnsafeMemory;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.GPC;
@@ -1882,6 +1883,7 @@ public class GridCacheContext<K, V> implements Externalizable {
* @param keepCacheObjects Keep cache objects flag.
* @param deserializeBinary Deserialize binary flag.
* @param cpy Copy flag.
+ * @param ver GridCacheVersion.
*/
@SuppressWarnings("unchecked")
public <K1, V1> void addResult(Map<K1, V1> map,
@@ -1890,7 +1892,8 @@ public class GridCacheContext<K, V> implements Externalizable {
boolean skipVals,
boolean keepCacheObjects,
boolean deserializeBinary,
- boolean cpy) {
+ boolean cpy,
+ final GridCacheVersion ver) {
assert key != null;
assert val != null || skipVals;
@@ -1902,10 +1905,32 @@ public class GridCacheContext<K, V> implements Externalizable {
assert key0 != null : key;
assert val0 != null : val;
- map.put((K1)key0, (V1)val0);
+ map.put((K1)key0, ver != null ? (V1)new T2<>(val0, ver) : (V1)val0);
}
else
- map.put((K1)key, (V1)(skipVals ? true : val));
+ map.put((K1)key,
+ (V1)(ver != null ?
+ (V1)new T2<>(skipVals ? true : val, ver) :
+ skipVals ? true : val));
+ }
+
+ /**
+ * @param map Map.
+ * @param key Key.
+ * @param val Value.
+ * @param skipVals Skip values flag.
+ * @param keepCacheObjects Keep cache objects flag.
+ * @param deserializeBinary Deserialize binary flag.
+ * @param cpy Copy flag.
+ */
+ public <K1, V1> void addResult(Map<K1, V1> map,
+ KeyCacheObject key,
+ CacheObject val,
+ boolean skipVals,
+ boolean keepCacheObjects,
+ boolean deserializeBinary,
+ boolean cpy) {
+ addResult(map, key, val, skipVals, keepCacheObjects, deserializeBinary, cpy, null);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index c896a82..ee2c2a1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -882,7 +882,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
updateTtl(expiryPlc);
if (retVer) {
- resVer = isNear() ? ((GridNearCacheEntry)this).dhtVersion() : this.ver;
+ resVer = (isNear() && cctx.transactional()) ? ((GridNearCacheEntry)this).dhtVersion() : this.ver;
if (resVer == null)
ret = null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
index 8ffd273..cf132c5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
@@ -31,6 +31,7 @@ import javax.cache.expiry.ExpiryPolicy;
import javax.cache.processor.EntryProcessor;
import javax.cache.processor.EntryProcessorResult;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.CacheEntry;
import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CachePeekMode;
import org.apache.ignite.cache.affinity.Affinity;
@@ -305,6 +306,18 @@ public class GridCacheProxyImpl<K, V> implements IgniteInternalCache<K, V>, Exte
}
/** {@inheritDoc} */
+ @Nullable @Override public CacheEntry<K, V> getEntry(K key) throws IgniteCheckedException {
+ CacheOperationContext prev = gate.enter(opCtx);
+
+ try {
+ return delegate.getEntry(key);
+ }
+ finally {
+ gate.leave(prev);
+ }
+ }
+
+ /** {@inheritDoc} */
@Override public V getTopologySafe(K key) throws IgniteCheckedException {
CacheOperationContext prev = gate.enter(opCtx);
@@ -329,6 +342,18 @@ public class GridCacheProxyImpl<K, V> implements IgniteInternalCache<K, V>, Exte
}
/** {@inheritDoc} */
+ @Override public IgniteInternalFuture<CacheEntry<K, V>> getEntryAsync(K key) {
+ CacheOperationContext prev = gate.enter(opCtx);
+
+ try {
+ return delegate.getEntryAsync(key);
+ }
+ finally {
+ gate.leave(prev);
+ }
+ }
+
+ /** {@inheritDoc} */
@Override public V getForcePrimary(K key) throws IgniteCheckedException {
CacheOperationContext prev = gate.enter(opCtx);
@@ -449,6 +474,19 @@ public class GridCacheProxyImpl<K, V> implements IgniteInternalCache<K, V>, Exte
}
/** {@inheritDoc} */
+ @Override public Collection<CacheEntry<K, V>> getEntries(
+ @Nullable Collection<? extends K> keys) throws IgniteCheckedException {
+ CacheOperationContext prev = gate.enter(opCtx);
+
+ try {
+ return delegate.getEntries(keys);
+ }
+ finally {
+ gate.leave(prev);
+ }
+ }
+
+ /** {@inheritDoc} */
@Override public IgniteInternalFuture<Map<K, V>> getAllAsync(@Nullable Collection<? extends K> keys) {
CacheOperationContext prev = gate.enter(opCtx);
@@ -461,6 +499,19 @@ public class GridCacheProxyImpl<K, V> implements IgniteInternalCache<K, V>, Exte
}
/** {@inheritDoc} */
+ @Override public IgniteInternalFuture<Collection<CacheEntry<K, V>>> getEntriesAsync(
+ @Nullable Collection<? extends K> keys) {
+ CacheOperationContext prev = gate.enter(opCtx);
+
+ try {
+ return delegate.getEntriesAsync(keys);
+ }
+ finally {
+ gate.leave(prev);
+ }
+ }
+
+ /** {@inheritDoc} */
@Nullable @Override public V getAndPut(K key, V val)
throws IgniteCheckedException {
CacheOperationContext prev = gate.enter(opCtx);
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/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 b64c69c..d8338bd 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
@@ -44,6 +44,7 @@ import javax.cache.processor.EntryProcessorResult;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CacheEntry;
import org.apache.ignite.cache.CacheEntryProcessor;
import org.apache.ignite.cache.CacheManager;
import org.apache.ignite.cache.CacheMetrics;
@@ -873,6 +874,31 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
}
/** {@inheritDoc} */
+ @Override public CacheEntry<K, V> getEntry(K key) {
+ try {
+ GridCacheGateway<K, V> gate = this.gate;
+
+ CacheOperationContext prev = onEnter(gate, opCtx);
+
+ try {
+ if (isAsync()) {
+ setFuture(delegate.getEntryAsync(key));
+
+ return null;
+ }
+ else
+ return delegate.getEntry(key);
+ }
+ finally {
+ onLeave(gate, prev);
+ }
+ }
+ catch (IgniteCheckedException e) {
+ throw cacheException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
@Override public Map<K, V> getAll(Set<? extends K> keys) {
try {
GridCacheGateway<K, V> gate = this.gate;
@@ -898,6 +924,31 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
}
/** {@inheritDoc} */
+ @Override public Collection<CacheEntry<K, V>> getEntries(Set<? extends K> keys) {
+ try {
+ GridCacheGateway<K, V> gate = this.gate;
+
+ CacheOperationContext prev = onEnter(gate, opCtx);
+
+ try {
+ if (isAsync()) {
+ setFuture(delegate.getEntriesAsync(keys));
+
+ return null;
+ }
+ else
+ return delegate.getEntries(keys);
+ }
+ finally {
+ onLeave(gate, prev);
+ }
+ }
+ catch (IgniteCheckedException e) {
+ throw cacheException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
@Override public Map<K, V> getAllOutTx(Set<? extends K> keys) {
try {
GridCacheGateway<K, V> gate = this.gate;
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
index 433290c..68d0f06 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
@@ -31,6 +31,7 @@ import javax.cache.processor.EntryProcessorResult;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheEntry;
import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.CachePeekMode;
@@ -335,6 +336,28 @@ public interface IgniteInternalCache<K, V> extends Iterable<Cache.Entry<K, V>> {
@Nullable public V get(K key) throws IgniteCheckedException;
/**
+ * Retrieves value mapped to the specified key from cache. Value will only be returned if
+ * its entry passed the optional filter provided. Filter check is atomic, and therefore the
+ * returned value is guaranteed to be consistent with the filter. The return value of {@code null}
+ * means entry did not pass the provided filter or cache has no mapping for the
+ * key.
+ * <p>
+ * If the value is not present in cache, then it will be looked up from swap storage. If
+ * it's not present in swap, or if swap is disable, and if read-through is allowed, value
+ * will be loaded from {@link CacheStore} persistent storage via
+ * <code>CacheStore#load(Transaction, Object)</code> method.
+ * <h2 class="header">Transactions</h2>
+ * This method is transactional and will enlist the entry into ongoing transaction
+ * if there is one.
+ *
+ * @param key Key to retrieve the value for.
+ * @return Value for the given key.
+ * @throws IgniteCheckedException If get operation failed.
+ * @throws NullPointerException if the key is {@code null}.
+ */
+ @Nullable public CacheEntry<K, V> getEntry(K key) throws IgniteCheckedException;
+
+ /**
* Asynchronously retrieves value mapped to the specified key from cache. Value will only be returned if
* its entry passed the optional filter provided. Filter check is atomic, and therefore the
* returned value is guaranteed to be consistent with the filter. The return value of {@code null}
@@ -356,6 +379,27 @@ public interface IgniteInternalCache<K, V> extends Iterable<Cache.Entry<K, V>> {
public IgniteInternalFuture<V> getAsync(K key);
/**
+ * Asynchronously retrieves value mapped to the specified key from cache. Value will only be returned if
+ * its entry passed the optional filter provided. Filter check is atomic, and therefore the
+ * returned value is guaranteed to be consistent with the filter. The return value of {@code null}
+ * means entry did not pass the provided filter or cache has no mapping for the
+ * key.
+ * <p>
+ * If the value is not present in cache, then it will be looked up from swap storage. If
+ * it's not present in swap, or if swap is disabled, and if read-through is allowed, value
+ * will be loaded from {@link CacheStore} persistent storage via
+ * <code>CacheStore#load(Transaction, Object)</code> method.
+ * <h2 class="header">Transactions</h2>
+ * This method is transactional and will enlist the entry into ongoing transaction
+ * if there is one.
+ *
+ * @param key Key for the value to get.
+ * @return Future for the get operation.
+ * @throws NullPointerException if the key is {@code null}.
+ */
+ public IgniteInternalFuture<CacheEntry<K, V>> getEntryAsync(K key);
+
+ /**
* Retrieves values mapped to the specified keys from cache. Value will only be returned if
* its entry passed the optional filter provided. Filter check is atomic, and therefore the
* returned value is guaranteed to be consistent with the filter. If requested key-value pair
@@ -377,6 +421,27 @@ public interface IgniteInternalCache<K, V> extends Iterable<Cache.Entry<K, V>> {
public Map<K, V> getAll(@Nullable Collection<? extends K> keys) throws IgniteCheckedException;
/**
+ * Retrieves values mapped to the specified keys from cache. Value will only be returned if
+ * its entry passed the optional filter provided. Filter check is atomic, and therefore the
+ * returned value is guaranteed to be consistent with the filter. If requested key-value pair
+ * is not present in the returned map, then it means that its entry did not pass the provided
+ * filter or cache has no mapping for the key.
+ * <p>
+ * If some value is not present in cache, then it will be looked up from swap storage. If
+ * it's not present in swap, or if swap is disabled, and if read-through is allowed, value
+ * will be loaded from {@link CacheStore} persistent storage via
+ * <code>CacheStore#loadAll(Transaction, Collection, org.apache.ignite.lang.IgniteBiInClosure)</code> method.
+ * <h2 class="header">Transactions</h2>
+ * This method is transactional and will enlist the entry into ongoing transaction
+ * if there is one.
+ *
+ * @param keys Keys to get.
+ * @return Map of key-value pairs.
+ * @throws IgniteCheckedException If get operation failed.
+ */
+ public Collection<CacheEntry<K, V>> getEntries(@Nullable Collection<? extends K> keys) throws IgniteCheckedException;
+
+ /**
* Asynchronously retrieves values mapped to the specified keys from cache. Value will only be returned if
* its entry passed the optional filter provided. Filter check is atomic, and therefore the
* returned value is guaranteed to be consistent with the filter. If requested key-value pair
@@ -397,6 +462,26 @@ public interface IgniteInternalCache<K, V> extends Iterable<Cache.Entry<K, V>> {
public IgniteInternalFuture<Map<K, V>> getAllAsync(@Nullable Collection<? extends K> keys);
/**
+ * Asynchronously retrieves values mapped to the specified keys from cache. Value will only be returned if
+ * its entry passed the optional filter provided. Filter check is atomic, and therefore the
+ * returned value is guaranteed to be consistent with the filter. If requested key-value pair
+ * is not present in the returned map, then it means that its entry did not pass the provided
+ * filter or cache has no mapping for the key.
+ * <p>
+ * If some value is not present in cache, then it will be looked up from swap storage. If
+ * it's not present in swap, or if swap is disabled, and if read-through is allowed, value
+ * will be loaded from {@link CacheStore} persistent storage via
+ * <code>CacheStore#loadAll(Transaction, Collection, org.apache.ignite.lang.IgniteBiInClosure)</code> method.
+ * <h2 class="header">Transactions</h2>
+ * This method is transactional and will enlist the entry into ongoing transaction
+ * if there is one.
+ *
+ * @param keys Key for the value to get.
+ * @return Future for the get operation.
+ */
+ public IgniteInternalFuture<Collection<CacheEntry<K, V>>> getEntriesAsync(@Nullable Collection<? extends K> keys);
+
+ /**
* Stores given key-value pair in cache. If filters are provided, then entries will
* be stored in cache only if they pass the filter. Note that filter check is atomic,
* so value stored in cache is guaranteed to be consistent with the filters. If cache
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java
index 7efaf49..28c94dd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java
@@ -153,21 +153,6 @@ public abstract class CacheDistributedGetFutureAdapter<K, V> extends GridCompoun
}
/**
- * @param map Result map.
- * @param key Key.
- * @param val Value.
- * @param ver Version.
- */
- @SuppressWarnings("unchecked")
- protected final void versionedResult(Map map, KeyCacheObject key, Object val, GridCacheVersion ver) {
- assert needVer;
- assert skipVals || val != null;
- assert ver != null;
-
- map.put(key, new T2<>(skipVals ? true : val, ver));
- }
-
- /**
* Affinity node to send get request to.
*
* @param affNodes All affinity nodes.
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
index 9cf8084..5be4e72 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
@@ -617,6 +617,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
* @param keys {@inheritDoc}
* @param forcePrimary {@inheritDoc}
* @param skipTx {@inheritDoc}
+ * @param needVer Need version.
* @return {@inheritDoc}
*/
@Override public IgniteInternalFuture<Map<K, V>> getAllAsync(
@@ -627,7 +628,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
String taskName,
boolean deserializeBinary,
boolean skipVals,
- boolean canRemap
+ boolean canRemap,
+ boolean needVer
) {
CacheOperationContext opCtx = ctx.operationContextPerCall();
@@ -640,7 +642,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
forcePrimary,
null,
skipVals,
- canRemap);
+ canRemap,
+ needVer);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
index cb8c842..c926c13 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
@@ -383,7 +383,8 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
/*deserialize binary*/false,
skipVals,
/*keep cache objects*/true,
- /*skip store*/!readThrough);
+ /*skip store*/!readThrough,
+ false);
}
}
else {
@@ -413,7 +414,8 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
/*deserialize binary*/false,
skipVals,
/*keep cache objects*/true,
- /*skip store*/!readThrough);
+ /*skip store*/!readThrough,
+ false);
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
index 1f2d7c5..2c9a760 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
@@ -490,17 +490,14 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
cache.removeIfObsolete(key);
}
else {
- if (needVer)
- versionedResult(locVals, key, v, ver);
- else {
- cctx.addResult(locVals,
- key,
- v,
- skipVals,
- keepCacheObjects,
- deserializeBinary,
- true);
- }
+ cctx.addResult(locVals,
+ key,
+ v,
+ skipVals,
+ keepCacheObjects,
+ deserializeBinary,
+ true,
+ ver);
return true;
}
@@ -552,17 +549,14 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
for (GridCacheEntryInfo info : infos) {
assert skipVals == (info.value() == null);
- if (needVer)
- versionedResult(map, info.key(), info.value(), info.version());
- else {
- cctx.addResult(map,
- info.key(),
- info.value(),
- skipVals,
- keepCacheObjects,
- deserializeBinary,
- false);
- }
+ cctx.addResult(map,
+ info.key(),
+ info.value(),
+ skipVals,
+ keepCacheObjects,
+ deserializeBinary,
+ false,
+ needVer ? info.version() : null);
}
return map;
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
index 0c811ae..01e61bf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
@@ -625,20 +625,13 @@ public class GridPartitionedSingleGetFuture extends GridFutureAdapter<Object> im
assert !skipVals;
if (val != null) {
- if (needVer) {
- assert ver != null;
+ if (!keepCacheObjects) {
+ Object res = cctx.unwrapBinaryIfNeeded(val, !deserializeBinary);
- onDone(new T2<>(val, ver));
- }
- else {
- if (!keepCacheObjects) {
- Object res = cctx.unwrapBinaryIfNeeded(val, !deserializeBinary);
-
- onDone(res);
- }
- else
- onDone(val);
+ onDone(needVer ? new T2<>(res, ver) : res);
}
+ else
+ onDone(needVer ? new T2<>(val, ver) : val);
}
else
onDone(null);
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index aa79cfa..96ba2da 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -77,7 +77,6 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSing
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrExpirationInfo;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
-import org.apache.ignite.internal.processors.cache.transactions.IgniteTxManager;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionConflictContext;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionEx;
@@ -94,6 +93,7 @@ import org.apache.ignite.internal.util.typedef.CO;
import org.apache.ignite.internal.util.typedef.CX1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.P1;
+import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -317,7 +317,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
/** {@inheritDoc} */
- @Override protected V get(K key, String taskName, boolean deserializeBinary) throws IgniteCheckedException {
+ @Override protected V get(K key, String taskName, boolean deserializeBinary, boolean needVer)
+ throws IgniteCheckedException {
ctx.checkSecurity(SecurityPermission.CACHE_READ);
if (keyCheck)
@@ -339,7 +340,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
expiryPlc,
false,
skipStore,
- true).get();
+ true,
+ needVer).get();
}
/** {@inheritDoc} */
@@ -350,7 +352,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
final String taskName,
final boolean deserializeBinary,
final boolean skipVals,
- final boolean canRemap) {
+ final boolean canRemap,
+ final boolean needVer) {
ctx.checkSecurity(SecurityPermission.CACHE_READ);
if (keyCheck)
@@ -376,7 +379,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
expiryPlc,
skipVals,
skipStore,
- canRemap);
+ canRemap,
+ needVer);
}
});
}
@@ -390,7 +394,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
final String taskName,
final boolean deserializeBinary,
final boolean skipVals,
- final boolean canRemap
+ final boolean canRemap,
+ final boolean needVer
) {
ctx.checkSecurity(SecurityPermission.CACHE_READ);
@@ -420,7 +425,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
expiryPlc,
skipVals,
skipStore,
- canRemap);
+ canRemap,
+ needVer);
}
});
}
@@ -1036,6 +1042,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
* @param skipVals Skip values flag.
* @param skipStore Skip store flag.
* @param canRemap Can remap flag.
+ * @param needVer Need version.
* @return Get future.
*/
private IgniteInternalFuture<V> getAsync0(KeyCacheObject key,
@@ -1046,7 +1053,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
@Nullable ExpiryPolicy expiryPlc,
boolean skipVals,
boolean skipStore,
- boolean canRemap
+ boolean canRemap,
+ boolean needVer
) {
AffinityTopologyVersion topVer = canRemap ? ctx.affinity().affinityTopologyVersion() :
ctx.shared().exchange().readyAffinityVersion();
@@ -1064,7 +1072,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
expiry,
skipVals,
canRemap,
- false,
+ needVer,
false);
fut.init();
@@ -1083,6 +1091,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
* @param expiryPlc Expiry policy.
* @param skipVals Skip values flag.
* @param skipStore Skip store flag.
+ * @param needVer Need version.
* @return Get future.
*/
private IgniteInternalFuture<Map<K, V>> getAllAsync0(@Nullable Collection<KeyCacheObject> keys,
@@ -1093,7 +1102,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
@Nullable ExpiryPolicy expiryPlc,
boolean skipVals,
boolean skipStore,
- boolean canRemap
+ boolean canRemap,
+ boolean needVer
) {
AffinityTopologyVersion topVer = canRemap ? ctx.affinity().affinityTopologyVersion() :
ctx.shared().exchange().readyAffinityVersion();
@@ -1118,19 +1128,42 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
if (entry != null) {
boolean isNew = entry.isNewLocked();
- CacheObject v = entry.innerGet(null,
- /*swap*/true,
- /*read-through*/false,
- /*fail-fast*/true,
- /*unmarshal*/true,
- /**update-metrics*/false,
- /*event*/!skipVals,
- /*temporary*/false,
- subjId,
- null,
- taskName,
- expiry,
- !deserializeBinary);
+ CacheObject v = null;
+ GridCacheVersion ver = null;
+
+ if (needVer) {
+ T2<CacheObject, GridCacheVersion> res = entry.innerGetVersioned(
+ null,
+ /*swap*/true,
+ /*unmarshal*/true,
+ /**update-metrics*/false,
+ /*event*/!skipVals,
+ subjId,
+ null,
+ taskName,
+ expiry,
+ true);
+
+ if (res != null) {
+ v = res.get1();
+ ver = res.get2();
+ }
+ }
+ else {
+ v = entry.innerGet(null,
+ /*swap*/true,
+ /*read-through*/false,
+ /*fail-fast*/true,
+ /*unmarshal*/true,
+ /**update-metrics*/false,
+ /*event*/!skipVals,
+ /*temporary*/false,
+ subjId,
+ null,
+ taskName,
+ expiry,
+ !deserializeBinary);
+ }
// Entry was not in memory or in swap, so we remove it from cache.
if (v == null) {
@@ -1142,7 +1175,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
success = false;
}
else
- ctx.addResult(locVals, key, v, skipVals, false, deserializeBinary, true);
+ ctx.addResult(locVals, key, v, skipVals, false, deserializeBinary, true, ver);
}
else
success = false;
@@ -1194,7 +1227,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
expiry,
skipVals,
canRemap,
- false,
+ needVer,
false);
fut.init();
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
index 073043d..dc4b6bd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
@@ -200,7 +200,8 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
String taskName,
final boolean deserializeBinary,
final boolean skipVals,
- boolean canRemap) {
+ boolean canRemap,
+ final boolean needVer) {
ctx.checkSecurity(SecurityPermission.CACHE_READ);
if (keyCheck)
@@ -218,7 +219,8 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
deserializeBinary,
skipVals,
false,
- opCtx != null && opCtx.skipStore());
+ opCtx != null && opCtx.skipStore(),
+ needVer);
return fut.chain(new CX1<IgniteInternalFuture<Map<Object, Object>>, V>() {
@SuppressWarnings("unchecked")
@@ -258,7 +260,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
skipVals ? null : expiryPolicy(opCtx != null ? opCtx.expiry() : null),
skipVals,
canRemap,
- /*needVer*/false,
+ needVer,
/*keepCacheObjects*/false);
fut.init();
@@ -275,7 +277,8 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
String taskName,
final boolean deserializeBinary,
final boolean skipVals,
- boolean canRemap
+ boolean canRemap,
+ final boolean needVer
) {
ctx.checkSecurity(SecurityPermission.CACHE_READ);
@@ -297,7 +300,8 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
deserializeBinary,
skipVals,
false,
- opCtx != null && opCtx.skipStore());
+ opCtx != null && opCtx.skipStore(),
+ needVer);
}
}, opCtx);
}
@@ -318,7 +322,8 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
deserializeBinary,
skipVals ? null : expiryPolicy(opCtx != null ? opCtx.expiry() : null),
skipVals,
- canRemap);
+ canRemap,
+ needVer);
}
/** {@inheritDoc} */
@@ -345,6 +350,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
* @param expiryPlc Expiry policy.
* @param skipVals Skip values flag.
* @param canRemap Can remap flag.
+ * @param needVer Need version.
* @return Loaded values.
*/
public IgniteInternalFuture<Map<K, V>> loadAsync(
@@ -357,7 +363,8 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
boolean deserializeBinary,
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean skipVals,
- boolean canRemap) {
+ boolean canRemap,
+ boolean needVer) {
return loadAsync(keys,
readThrough,
forcePrimary,
@@ -367,7 +374,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
expiryPlc,
skipVals,
canRemap,
- false,
+ needVer,
false);
}
@@ -522,17 +529,14 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
if (locVals == null)
locVals = U.newHashMap(keys.size());
- if (needVer)
- locVals.put((K)key, (V)new T2<>((Object)(skipVals ? true : v), ver));
- else {
- ctx.addResult(locVals,
- key,
- v,
- skipVals,
- keepCacheObj,
- deserializeBinary,
- true);
- }
+ ctx.addResult(locVals,
+ key,
+ v,
+ skipVals,
+ keepCacheObj,
+ deserializeBinary,
+ true,
+ ver);
}
}
else
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
index a2d5adb..63c073d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
@@ -400,7 +400,8 @@ public class GridNearAtomicCache<K, V> extends GridNearCacheAdapter<K, V> {
String taskName,
boolean deserializeBinary,
boolean skipVals,
- boolean canRemap
+ boolean canRemap,
+ boolean needVer
) {
ctx.checkSecurity(SecurityPermission.CACHE_READ);
@@ -423,7 +424,8 @@ public class GridNearAtomicCache<K, V> extends GridNearCacheAdapter<K, V> {
skipVals ? null : opCtx != null ? opCtx.expiry() : null,
skipVals,
opCtx != null && opCtx.skipStore(),
- canRemap);
+ canRemap,
+ needVer);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java
index 5bf18d9..c750be6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java
@@ -230,6 +230,7 @@ public abstract class GridNearCacheAdapter<K, V> extends GridDistributedCacheAda
* @param skipVal Skip value flag.
* @param skipStore Skip store flag.
* @param canRemap Can remap flag.
+ * @param needVer Need version.
* @return Loaded values.
*/
public IgniteInternalFuture<Map<K, V>> loadAsync(@Nullable IgniteInternalTx tx,
@@ -241,7 +242,8 @@ public abstract class GridNearCacheAdapter<K, V> extends GridDistributedCacheAda
@Nullable ExpiryPolicy expiryPlc,
boolean skipVal,
boolean skipStore,
- boolean canRemap
+ boolean canRemap,
+ boolean needVer
) {
if (F.isEmpty(keys))
return new GridFinishedFuture<>(Collections.<K, V>emptyMap());
@@ -261,7 +263,7 @@ public abstract class GridNearCacheAdapter<K, V> extends GridDistributedCacheAda
expiry,
skipVal,
canRemap,
- false,
+ needVer,
false);
// init() will register future for responses if future has remote mappings.
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheEntry.java
index c0a1617..026fb4d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheEntry.java
@@ -350,7 +350,8 @@ public class GridNearCacheEntry extends GridDistributedCacheEntry {
null,
false,
/*skip store*/false,
- /*can remap*/true
+ /*can remap*/true,
+ false
).get().get(keyValue(false));
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
index 9291001..06fc0a5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
@@ -650,26 +650,25 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
*/
@SuppressWarnings("unchecked")
private void addResult(KeyCacheObject key, CacheObject v, GridCacheVersion ver) {
- if (needVer) {
- V val0 = (V)new T2<>(skipVals ? true : v, ver);
+ if (keepCacheObjects) {
+ K key0 = (K)key;
+ V val0 = needVer ?
+ (V)new T2<>(skipVals ? true : v, ver) :
+ (V)(skipVals ? true : v);
- add(new GridFinishedFuture<>(Collections.singletonMap((K)key, val0)));
+ add(new GridFinishedFuture<>(Collections.singletonMap(key0, val0)));
}
else {
- if (keepCacheObjects) {
- K key0 = (K)key;
- V val0 = (V)(skipVals ? true : v);
-
- add(new GridFinishedFuture<>(Collections.singletonMap(key0, val0)));
- }
- else {
- K key0 = (K)cctx.unwrapBinaryIfNeeded(key, !deserializeBinary, false);
- V val0 = !skipVals ?
+ K key0 = (K)cctx.unwrapBinaryIfNeeded(key, !deserializeBinary, false);
+ V val0 = needVer ?
+ (V)new T2<>(!skipVals ?
+ (V)cctx.unwrapBinaryIfNeeded(v, !deserializeBinary, false) :
+ (V)Boolean.TRUE, ver) :
+ !skipVals ?
(V)cctx.unwrapBinaryIfNeeded(v, !deserializeBinary, false) :
(V)Boolean.TRUE;
- add(new GridFinishedFuture<>(Collections.singletonMap(key0, val0)));
- }
+ add(new GridFinishedFuture<>(Collections.singletonMap(key0, val0)));
}
}
@@ -741,16 +740,14 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
assert skipVals == (info.value() == null);
- if (needVer)
- versionedResult(map, key, val, info.version());
- else
- cctx.addResult(map,
- key,
- val,
- skipVals,
- keepCacheObjects,
- deserializeBinary,
- false);
+ cctx.addResult(map,
+ key,
+ val,
+ skipVals,
+ keepCacheObjects,
+ deserializeBinary,
+ false,
+ needVer ? info.version() : null);
}
catch (GridCacheEntryRemovedException ignore) {
if (log.isDebugEnabled())
http://git-wip-us.apache.org/repos/asf/ignite/blob/b28f63e4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
index a09dec0..a3130cd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
@@ -122,7 +122,8 @@ public class GridNearTransactionalCache<K, V> extends GridNearCacheAdapter<K, V>
String taskName,
final boolean deserializeBinary,
final boolean skipVals,
- boolean canRemap
+ boolean canRemap,
+ final boolean needVer
) {
ctx.checkSecurity(SecurityPermission.CACHE_READ);
@@ -146,7 +147,8 @@ public class GridNearTransactionalCache<K, V> extends GridNearCacheAdapter<K, V>
deserializeBinary,
skipVals,
false,
- skipStore);
+ skipStore,
+ needVer);
}
}, opCtx);
}
@@ -162,7 +164,8 @@ public class GridNearTransactionalCache<K, V> extends GridNearCacheAdapter<K, V>
skipVals ? null : opCtx != null ? opCtx.expiry() : null,
skipVals,
skipStore,
- canRemap);
+ canRemap,
+ needVer);
}
/**