You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2014/12/23 09:22:43 UTC
[07/38] incubator-ignite git commit: # ignite-41
# ignite-41
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/d84da7d0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/d84da7d0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/d84da7d0
Branch: refs/heads/ignite-1
Commit: d84da7d0846036eddf8f300c56735f7c1ab1e948
Parents: 688a2e7
Author: sboikov <sb...@gridgain.com>
Authored: Wed Dec 17 12:16:32 2014 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Wed Dec 17 13:52:23 2014 +0300
----------------------------------------------------------------------
.../processors/cache/GridCacheContext.java | 2 +-
.../processors/cache/GridCacheEntryEx.java | 4 +-
.../processors/cache/GridCacheIoManager.java | 2 +-
.../processors/cache/GridCacheMapEntry.java | 40 +-
.../processors/cache/GridCacheMessage.java | 9 +
.../processors/cache/GridCacheTxEntry.java | 56 +-
.../cache/GridCacheTxLocalAdapter.java | 47 +-
.../dht/GridDhtTransactionalCacheAdapter.java | 12 +-
.../distributed/dht/GridDhtTxLocalAdapter.java | 3 +-
.../distributed/near/GridNearLockRequest.java | 8 +-
.../local/atomic/GridLocalAtomicCache.java | 65 +-
.../cache/IgniteCacheAbstractTest.java | 152 ++++
.../processors/cache/IgniteCacheTest.java | 124 ----
.../IgniteCacheAtomicExpiryPolicyTest.java | 41 ++
.../IgniteCacheAtomicLocalExpiryPolicyTest.java | 41 ++
...teCacheAtomicReplicatedExpiryPolicyTest.java | 24 +
.../IgniteCacheExpiryPolicyAbstractTest.java | 714 +++++++++++++++++++
.../expiry/IgniteCacheExpiryPolicyTest.java | 507 -------------
.../IgniteCacheExpiryPolicyTestSuite.java | 35 +
.../expiry/IgniteCacheTxExpiryPolicyTest.java | 41 ++
.../IgniteCacheTxLocalExpiryPolicyTest.java | 41 ++
...IgniteCacheTxReplicatedExpiryPolicyTest.java | 26 +
.../processors/cache/GridCacheTestEntryEx.java | 15 +-
.../testframework/junits/GridAbstractTest.java | 8 +
.../junits/common/GridCommonAbstractTest.java | 25 +-
.../bamboo/GridDataGridTestSuite.java | 2 +-
26 files changed, 1343 insertions(+), 701 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheContext.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheContext.java
index 58fd1cd..931e243 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheContext.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheContext.java
@@ -283,7 +283,7 @@ public class GridCacheContext<K, V> implements Externalizable {
Factory<ExpiryPolicy> factory = cacheCfg.getExpiryPolicyFactory();
- expiryPlc = factory.create();
+ expiryPlc = factory != null ? factory.create() : null;
if (expiryPlc instanceof EternalExpiryPolicy)
expiryPlc = null;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheEntryEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheEntryEx.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheEntryEx.java
index fc8aaaa..76d73b3 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheEntryEx.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheEntryEx.java
@@ -447,7 +447,7 @@ public interface GridCacheEntryEx<K, V> extends GridMetadataAware {
* @param writeObj Value. Type depends on operation.
* @param writeThrough Write through flag.
* @param retval Return value flag.
- * @param ttl Time to live.
+ * @param expiryPlc Expiry policy..
* @param evt Event flag.
* @param metrics Metrics update flag.
* @param filter Optional filter to check.
@@ -464,7 +464,7 @@ public interface GridCacheEntryEx<K, V> extends GridMetadataAware {
@Nullable Object writeObj,
boolean writeThrough,
boolean retval,
- long ttl,
+ @Nullable ExpiryPolicy expiryPlc,
boolean evt,
boolean metrics,
@Nullable IgnitePredicate<GridCacheEntry<K, V>>[] filter,
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheIoManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheIoManager.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheIoManager.java
index a50e461..7320595 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheIoManager.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheIoManager.java
@@ -200,7 +200,7 @@ public class GridCacheIoManager<K, V> extends GridCacheSharedManagerAdapter<K, V
unmarshall(nodeId, cacheMsg);
- log.info("Message: " + cacheMsg);
+ //log.info("Message: " + cacheMsg);
if (cacheMsg.allowForStartup())
processMessage(nodeId, cacheMsg, c);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
index 2c377e7..879f796 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
@@ -1112,7 +1112,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
@Nullable UUID subjId,
String taskName
) throws IgniteCheckedException, GridCacheEntryRemovedException {
- log.info("Inner set " + key + " " + val + " " + ttl);
+ // log.info("Inner set " + key + " " + val + " " + ttl);
V old;
@@ -1432,7 +1432,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
@Nullable Object writeObj,
boolean writeThrough,
boolean retval,
- long ttl,
+ @Nullable ExpiryPolicy expiryPlc,
boolean evt,
boolean metrics,
@Nullable IgnitePredicate<GridCacheEntry<K, V>>[] filter,
@@ -1456,13 +1456,6 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
if (isNew())
unswap(true, retval);
- long newTtl = ttl;
-
- if (newTtl < 0)
- newTtl = ttlExtras();
-
- long newExpireTime = toExpireTime(newTtl);
-
// Possibly get old value form store.
old = needVal ? rawGetOrUnmarshalUnlocked(!retval) : val;
@@ -1541,11 +1534,36 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
// Must persist inside synchronization in non-tx mode.
cctx.store().putToStore(null, key, updated, ver);
+ long ttl;
+ long expireTime;
+
+ if (expiryPlc != null) {
+ if (!hadVal) {
+ Duration duration = expiryPlc.getExpiryForCreation();
+
+ if (duration != null && duration.isZero())
+ return new IgniteBiTuple<>(false, cctx.<V>unwrapTemporary(old));
+
+ ttl = toTtl(duration);
+ }
+ else
+ ttl = toTtl(expiryPlc.getExpiryForUpdate());
+
+ ttl = ttl < 0 ? ttlExtras() : ttl;
+
+ expireTime = toExpireTime(ttl);
+ }
+ else {
+ ttl = ttlExtras();
+
+ expireTime = toExpireTime(ttl);
+ }
+
// Update index inside synchronization since it can be updated
// in load methods without actually holding entry lock.
- updateIndex(updated, null, newExpireTime, ver, old);
+ updateIndex(updated, null, expireTime, ver, old);
- update(updated, null, newExpireTime, newTtl, ver);
+ update(updated, null, expireTime, ttl, ver);
if (evt) {
V evtOld = null;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMessage.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMessage.java
index 71eac41..882fc5a 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMessage.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMessage.java
@@ -330,7 +330,12 @@ public abstract class GridCacheMessage<K, V> extends GridTcpCommunicationMessage
assert ctx != null;
if (txEntries != null) {
+ boolean transferExpiry = transferExpiryPolicy();
+
for (GridCacheTxEntry<K, V> e : txEntries) {
+ if (transferExpiry)
+ e.transferExpiryPolicyIfNeeded();
+
e.marshal(ctx);
if (ctx.deploymentEnabled()) {
@@ -342,6 +347,10 @@ public abstract class GridCacheMessage<K, V> extends GridTcpCommunicationMessage
}
}
+ protected boolean transferExpiryPolicy() {
+ return false;
+ }
+
/**
* @param txEntries Entries to unmarshal.
* @param ctx Context.
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxEntry.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxEntry.java
index 91b9cc0..df888e7 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxEntry.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxEntry.java
@@ -13,12 +13,14 @@ import org.apache.ignite.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.marshaller.optimized.*;
import org.gridgain.grid.cache.*;
+import org.gridgain.grid.kernal.processors.cache.distributed.*;
import org.gridgain.grid.util.lang.*;
import org.gridgain.grid.util.tostring.*;
import org.gridgain.grid.util.typedef.*;
import org.gridgain.grid.util.typedef.internal.*;
import org.jetbrains.annotations.*;
+import javax.cache.expiry.*;
import java.io.*;
import java.util.*;
import java.util.concurrent.atomic.*;
@@ -128,6 +130,12 @@ public class GridCacheTxEntry<K, V> implements GridPeerDeployAware, Externalizab
/** Data center replication version. */
private GridCacheVersion drVer;
+ /** Expiry policy. */
+ private ExpiryPolicy expiryPlc;
+
+ /** */
+ private boolean transferExpiryPlc;
+
/**
* Required by {@link Externalizable}
*/
@@ -147,8 +155,14 @@ public class GridCacheTxEntry<K, V> implements GridPeerDeployAware, Externalizab
* @param entry Cache entry.
* @param drVer Data center replication version.
*/
- public GridCacheTxEntry(GridCacheContext<K, V> ctx, GridCacheTxEx<K, V> tx, GridCacheOperation op, V val,
- long ttl, long drExpireTime, GridCacheEntryEx<K, V> entry, @Nullable GridCacheVersion drVer) {
+ public GridCacheTxEntry(GridCacheContext<K, V> ctx,
+ GridCacheTxEx<K, V> tx,
+ GridCacheOperation op,
+ V val,
+ long ttl,
+ long drExpireTime,
+ GridCacheEntryEx<K, V> entry,
+ @Nullable GridCacheVersion drVer) {
assert ctx != null;
assert tx != null;
assert op != null;
@@ -183,9 +197,15 @@ public class GridCacheTxEntry<K, V> implements GridPeerDeployAware, Externalizab
* @param filters Put filters.
* @param drVer Data center replication version.
*/
- public GridCacheTxEntry(GridCacheContext<K, V> ctx, GridCacheTxEx<K, V> tx, GridCacheOperation op,
- V val, IgniteClosure<V, V> transformClos, long ttl, GridCacheEntryEx<K,V> entry,
- IgnitePredicate<GridCacheEntry<K, V>>[] filters, GridCacheVersion drVer) {
+ public GridCacheTxEntry(GridCacheContext<K, V> ctx,
+ GridCacheTxEx<K, V> tx,
+ GridCacheOperation op,
+ V val,
+ IgniteClosure<V, V> transformClos,
+ long ttl,
+ GridCacheEntryEx<K,V> entry,
+ IgnitePredicate<GridCacheEntry<K, V>>[] filters,
+ GridCacheVersion drVer) {
assert ctx != null;
assert tx != null;
assert op != null;
@@ -285,6 +305,7 @@ public class GridCacheTxEntry<K, V> implements GridPeerDeployAware, Externalizab
cp.grpLock = grpLock;
cp.depEnabled = depEnabled;
cp.drVer = drVer;
+ cp.expiryPlc = expiryPlc;
return cp;
}
@@ -708,6 +729,13 @@ public class GridCacheTxEntry<K, V> implements GridPeerDeployAware, Externalizab
}
/**
+ * Marks expiry policy for transfer if it explicitly set and differs from default one.
+ */
+ public void transferExpiryPolicyIfNeeded() {
+ transferExpiryPlc = expiryPlc != null && expiryPlc != ctx.expiry();
+ }
+
+ /**
* @param ctx Context.
* @throws IgniteCheckedException If failed.
*/
@@ -768,6 +796,20 @@ public class GridCacheTxEntry<K, V> implements GridPeerDeployAware, Externalizab
val.unmarshal(this.ctx, clsLdr, depEnabled);
}
+ /**
+ * @param expiryPlc Expiry policy.
+ */
+ public void expiry(@Nullable ExpiryPolicy expiryPlc) {
+ this.expiryPlc = expiryPlc;
+ }
+
+ /**
+ * @return Expiry policy.
+ */
+ @Nullable public ExpiryPolicy expiry() {
+ return expiryPlc;
+ }
+
/** {@inheritDoc} */
@Override public void writeExternal(ObjectOutput out) throws IOException {
out.writeBoolean(depEnabled);
@@ -793,6 +835,8 @@ public class GridCacheTxEntry<K, V> implements GridPeerDeployAware, Externalizab
CU.writeVersion(out, explicitVer);
out.writeBoolean(grpLock);
CU.writeVersion(out, drVer);
+
+ out.writeObject(transferExpiryPlc ? new GridCacheExpiryPolicy(expiryPlc) : null);
}
/** {@inheritDoc} */
@@ -821,6 +865,8 @@ public class GridCacheTxEntry<K, V> implements GridPeerDeployAware, Externalizab
explicitVer = CU.readVersion(in);
grpLock = in.readBoolean();
drVer = CU.readVersion(in);
+
+ expiryPlc = (ExpiryPolicy)in.readObject();
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxLocalAdapter.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxLocalAdapter.java
index c6888f1..7b6f266 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheTxLocalAdapter.java
@@ -653,7 +653,10 @@ public abstract class GridCacheTxLocalAdapter<K, V> extends GridCacheTxAdapter<K
byte[] valBytes = res.get3();
if (op == CREATE || op == UPDATE && txEntry.drExpireTime() == -1L) {
- ExpiryPolicy expiry = cacheCtx.expiry();
+ ExpiryPolicy expiry = txEntry.expiry();
+
+ if (expiry == null)
+ expiry = cacheCtx.expiry();
if (expiry != null) {
Duration duration = cached.hasValue() ?
@@ -661,7 +664,7 @@ public abstract class GridCacheTxLocalAdapter<K, V> extends GridCacheTxAdapter<K
txEntry.ttl(GridCacheMapEntry.toTtl(duration));
- log.info("Calculated expiry (userCommit), update=" + cached.hasValue() + ", ttl=" + txEntry.ttl() + ", detached=" + cached.detached());
+ log.info("Calculated expiry (userCommit), update=" + cached.hasValue() + ", ttl=" + txEntry.ttl() + ", plc=" + expiry);
}
}
@@ -2684,7 +2687,7 @@ public abstract class GridCacheTxLocalAdapter<K, V> extends GridCacheTxAdapter<K
@Nullable V val,
@Nullable IgniteClosure<V, V> transformClos,
GridCacheEntryEx<K, V> entry,
- @Nullable ExpiryPolicy expiryPlc, // TODO IGNITE-41
+ @Nullable ExpiryPolicy expiryPlc,
IgnitePredicate<GridCacheEntry<K, V>>[] filter,
boolean filtersSet,
long drTtl,
@@ -2739,7 +2742,7 @@ public abstract class GridCacheTxLocalAdapter<K, V> extends GridCacheTxAdapter<K
entryTtlDr(key, drTtl, drExpireTime);
}
else
- entryTtl(key, ttl);
+ entryExpiry(key, expiryPlc);
txEntry = old;
@@ -2747,15 +2750,23 @@ public abstract class GridCacheTxLocalAdapter<K, V> extends GridCacheTxAdapter<K
log.debug("Updated transaction entry: " + txEntry);
}
else {
- long ttl = -1L;
-
- if (drTtl >= 0L)
- ttl = drTtl;
-
- txEntry = new GridCacheTxEntry<>(entry.context(), this, op, val, transformClos, ttl, entry, filter, drVer);
+ boolean hasDrTtl = drTtl >= 0;
+
+ txEntry = new GridCacheTxEntry<>(entry.context(),
+ this,
+ op,
+ val,
+ transformClos,
+ hasDrTtl ? drTtl : -1L,
+ entry,
+ filter,
+ drVer);
txEntry.drExpireTime(drExpireTime);
+ if (!hasDrTtl)
+ txEntry.expiry(expiryPlc);
+
txMap.put(key, txEntry);
if (log.isDebugEnabled())
@@ -2844,6 +2855,19 @@ public abstract class GridCacheTxLocalAdapter<K, V> extends GridCacheTxAdapter<K
/**
* @param key Key.
+ * @param expiryPlc Expiry policy.
+ */
+ void entryExpiry(GridCacheTxKey<K> key, @Nullable ExpiryPolicy expiryPlc) {
+ assert key != null;
+
+ GridCacheTxEntry<K, V> e = entry(key);
+
+ if (e != null)
+ e.expiry(expiryPlc);
+ }
+
+ /**
+ * @param key Key.
* @param ttl TTL.
* @param expireTime Expire time.
* @return {@code true} if tx entry exists for this key, {@code false} otherwise.
@@ -2856,7 +2880,10 @@ public abstract class GridCacheTxLocalAdapter<K, V> extends GridCacheTxAdapter<K
if (e != null) {
e.ttl(ttl);
+
e.drExpireTime(expireTime);
+
+ e.expiry(null);
}
return e != null;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
index 3f30801..7cee7d9 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
@@ -692,8 +692,16 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
GridDhtLockFuture<K, V> fut = null;
if (!req.inTx()) {
- fut = new GridDhtLockFuture<>(ctx, nearNode.id(), req.version(),
- req.topologyVersion(), cnt, req.txRead(), req.timeout(), tx, req.threadId(), filter);
+ fut = new GridDhtLockFuture<>(ctx,
+ nearNode.id(),
+ req.version(),
+ req.topologyVersion(),
+ cnt,
+ req.txRead(),
+ req.timeout(),
+ tx,
+ req.threadId(),
+ filter);
// Add before mapping.
if (!ctx.mvcc().addFuture(fut))
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
index 461ea04..44c397a 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
@@ -512,7 +512,7 @@ public abstract class GridDhtTxLocalAdapter<K, V> extends GridCacheTxLocalAdapte
GridCacheTxEntry<K, V> w = writeEntries == null ? null : writeEntries.get(idx++);
- txEntry = addEntry(NOOP, null, null, cached, -1, CU.<K, V>empty(), false, -1L, -1L,
+ txEntry = addEntry(NOOP, null, null, cached, null, CU.<K, V>empty(), false, -1L, -1L,
drVers != null ? drVers[drVerIdx++] : null);
if (w != null) {
@@ -526,6 +526,7 @@ public abstract class GridDhtTxLocalAdapter<K, V> extends GridCacheTxLocalAdapte
txEntry.ttl(w.ttl());
txEntry.filters(w.filters());
txEntry.drExpireTime(w.drExpireTime());
+ txEntry.expiry(w.expiry());
}
txEntry.cached(cached, txEntry.keyBytes());
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/near/GridNearLockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/near/GridNearLockRequest.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/near/GridNearLockRequest.java
index abcc189..4a1d501 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/near/GridNearLockRequest.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/near/GridNearLockRequest.java
@@ -284,8 +284,12 @@ public class GridNearLockRequest<K, V> extends GridDistributedLockRequest<K, V>
return dhtVers[idx];
}
- /** {@inheritDoc}
- * @param ctx*/
+ /** {@inheritDoc} */
+ @Override protected boolean transferExpiryPolicy() {
+ return true;
+ }
+
+ /** {@inheritDoc} */
@Override public void prepareMarshal(GridCacheSharedContext<K, V> ctx) throws IgniteCheckedException {
super.prepareMarshal(ctx);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java
index e4d6b00..433d199 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java
@@ -22,6 +22,7 @@ import org.gridgain.grid.util.typedef.internal.*;
import org.jetbrains.annotations.*;
import sun.misc.*;
+import javax.cache.expiry.*;
import java.io.*;
import java.util.*;
import java.util.concurrent.*;
@@ -103,7 +104,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
return (V)updateAllInternal(UPDATE,
Collections.singleton(key),
Collections.singleton(val),
- ttl,
+ expiryPerCall(),
true,
false,
filter,
@@ -124,7 +125,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
return (Boolean)updateAllInternal(UPDATE,
Collections.singleton(key),
Collections.singleton(val),
- ttl,
+ expiryPerCall(),
false,
false,
filter,
@@ -142,7 +143,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
return (Boolean)updateAllInternal(UPDATE,
Collections.singleton(key),
Collections.singleton(val),
- -1,
+ expiryPerCall(),
false,
false,
filter,
@@ -239,7 +240,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
return (GridCacheReturn<V>)updateAllInternal(UPDATE,
Collections.singleton(key),
Collections.singleton(newVal),
- 0,
+ expiryPerCall(),
true,
true,
ctx.equalsPeekArray(oldVal),
@@ -256,7 +257,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
return (GridCacheReturn<V>)updateAllInternal(DELETE,
Collections.singleton(key),
null,
- 0,
+ null,
true,
true,
ctx.equalsPeekArray(val),
@@ -292,7 +293,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
updateAllInternal(UPDATE,
m.keySet(),
m.values(),
- 0,
+ expiryPerCall(),
false,
false,
filter,
@@ -314,7 +315,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
updateAllInternal(TRANSFORM,
Collections.singleton(key),
Collections.singleton(transformer),
- -1,
+ expiryPerCall(),
false,
false,
null,
@@ -328,7 +329,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
return (R)updateAllInternal(TRANSFORM,
Collections.singleton(key),
Collections.singleton(new GridCacheTransformComputeClosure<>(transformer)),
- -1,
+ expiryPerCall(),
true,
false,
null,
@@ -356,7 +357,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
updateAllInternal(TRANSFORM,
m.keySet(),
m.values(),
- 0,
+ expiryPerCall(),
false,
false,
null,
@@ -383,7 +384,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
return (V)updateAllInternal(DELETE,
Collections.singleton(key),
null,
- 0,
+ null,
true,
false,
filter,
@@ -409,7 +410,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
updateAllInternal(DELETE,
keys,
null,
- 0,
+ null,
false,
false,
filter,
@@ -436,7 +437,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
return (Boolean)updateAllInternal(DELETE,
Collections.singleton(key),
null,
- 0,
+ null,
false,
false,
filter,
@@ -464,7 +465,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
return (Boolean)updateAllInternal(DELETE,
Collections.singleton(key),
null,
- 0,
+ null,
false,
false,
ctx.equalsPeekArray(val),
@@ -678,13 +679,14 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
map != null ? map.keySet() : transformMap != null ? transformMap.keySet() : null;
final Collection<?> vals = map != null ? map.values() : transformMap != null ? transformMap.values() : null;
final boolean storeEnabled = ctx.isStoreEnabled();
+ final ExpiryPolicy expiry = expiryPerCall();
return asyncOp(new Callable<Object>() {
@Override public Object call() throws Exception {
return updateAllInternal(op,
keys,
vals,
- ttl,
+ expiry,
retval,
rawRetval,
filter,
@@ -715,7 +717,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
return updateAllInternal(DELETE,
keys,
null,
- 0,
+ null,
retval,
rawRetval,
filter,
@@ -730,7 +732,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
* @param op Operation.
* @param keys Keys.
* @param vals Values.
- * @param ttl Time to live.
+ * @param expiryPlc Expiry policy.
* @param retval Return value required flag.
* @param rawRetval Return {@code GridCacheReturn} instance.
* @param filter Cache entry filter.
@@ -742,7 +744,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
private Object updateAllInternal(GridCacheOperation op,
Collection<? extends K> keys,
@Nullable Iterable<?> vals,
- long ttl,
+ @Nullable ExpiryPolicy expiryPlc,
boolean retval,
boolean rawRetval,
IgnitePredicate<GridCacheEntry<K, V>>[] filter,
@@ -762,7 +764,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
UUID subjId = ctx.subjectIdPerCall(null);
if (storeEnabled && keys.size() > 1) {
- updateWithBatch(op, keys, vals, ver, filter, subjId, taskName);
+ updateWithBatch(op, keys, vals, expiryPlc, ver, filter, subjId, taskName);
return null;
}
@@ -793,7 +795,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
val,
storeEnabled,
retval,
- ttl,
+ expiryPlc,
true,
true,
filter,
@@ -860,6 +862,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
GridCacheOperation op,
Collection<? extends K> keys,
@Nullable Iterable<?> vals,
+ @Nullable ExpiryPolicy expiryPlc,
GridCacheVersion ver,
@Nullable IgnitePredicate<GridCacheEntry<K, V>>[] filter,
UUID subjId,
@@ -941,6 +944,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
ver,
putMap,
null,
+ expiryPlc,
err,
subjId,
taskName);
@@ -971,6 +975,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
ver,
null,
rmvKeys,
+ expiryPlc,
err,
subjId,
taskName);
@@ -1067,6 +1072,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
ver,
putMap,
rmvKeys,
+ expiryPlc,
err,
subjId,
taskName);
@@ -1087,16 +1093,19 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
* @param ver Cache version.
* @param putMap Values to put.
* @param rmvKeys Keys to remove.
+ * @param expiryPlc Expiry policy.
* @param err Optional partial update exception.
* @param subjId Subject ID.
* @param taskName Task name.
* @return Partial update exception.
*/
@SuppressWarnings({"unchecked", "ConstantConditions", "ForLoopReplaceableByForEach"})
- @Nullable private GridCachePartialUpdateException updatePartialBatch(List<GridCacheEntryEx<K, V>> entries,
+ @Nullable private GridCachePartialUpdateException updatePartialBatch(
+ List<GridCacheEntryEx<K, V>> entries,
final GridCacheVersion ver,
@Nullable Map<K, V> putMap,
@Nullable Collection<K> rmvKeys,
+ @Nullable ExpiryPolicy expiryPlc,
@Nullable GridCachePartialUpdateException err,
UUID subjId,
String taskName
@@ -1151,7 +1160,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
writeVal,
false,
false,
- 0,
+ expiryPlc,
true,
true,
null,
@@ -1275,6 +1284,20 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
}
/**
+ * @return Expiry policy.
+ */
+ @Nullable private ExpiryPolicy expiryPerCall() {
+ GridCacheProjectionImpl<K, V> prj = ctx.projectionPerCall();
+
+ ExpiryPolicy expiry = prj != null ? prj.expiry() : null;
+
+ if (expiry == null)
+ expiry = ctx.expiry();
+
+ return expiry;
+ }
+
+ /**
* @param op Operation closure.
* @return Future.
*/
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractTest.java
new file mode 100644
index 0000000..9293431
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractTest.java
@@ -0,0 +1,152 @@
+/* @java.file.header */
+
+/* _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.apache.ignite.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.marshaller.optimized.*;
+import org.apache.ignite.spi.discovery.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
+import org.gridgain.grid.cache.*;
+import org.gridgain.testframework.junits.common.*;
+
+import static org.gridgain.grid.cache.GridCacheMode.*;
+import static org.gridgain.grid.cache.GridCacheWriteSynchronizationMode.*;
+
+/**
+ * Abstract class for cache tests.
+ */
+public abstract class IgniteCacheAbstractTest extends GridCommonAbstractTest {
+ /** */
+ private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /**
+ * @return Grids count to start.
+ */
+ protected abstract int gridCount();
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ startGrids();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ protected void startGrids() throws Exception {
+ int cnt = gridCount();
+
+ assert cnt >= 1 : "At least one grid must be started";
+
+ startGridsMultiThreaded(cnt);
+
+ awaitPartitionMapExchange();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ TcpDiscoverySpi disco = new TcpDiscoverySpi();
+
+ disco.setMaxMissedHeartbeats(Integer.MAX_VALUE);
+
+ disco.setIpFinder(ipFinder);
+
+ if (isDebug())
+ disco.setAckTimeout(Integer.MAX_VALUE);
+
+ cfg.setDiscoverySpi(disco);
+
+ cfg.setCacheConfiguration(cacheConfiguration(gridName));
+
+ cfg.setMarshaller(new IgniteOptimizedMarshaller(false));
+
+ return cfg;
+ }
+
+ /**
+ * @param gridName Grid name.
+ * @return Cache configuration.
+ * @throws Exception In case of error.
+ */
+ protected GridCacheConfiguration cacheConfiguration(String gridName) throws Exception {
+ GridCacheConfiguration cfg = defaultCacheConfiguration();
+
+ cfg.setSwapEnabled(swapEnabled());
+ cfg.setCacheMode(cacheMode());
+ cfg.setAtomicityMode(atomicityMode());
+ cfg.setWriteSynchronizationMode(writeSynchronization());
+ cfg.setDistributionMode(distributionMode());
+ cfg.setPortableEnabled(portableEnabled());
+
+ if (cacheMode() == PARTITIONED)
+ cfg.setBackups(1);
+
+ return cfg;
+ }
+
+ /**
+ * @return Default cache mode.
+ */
+ protected abstract GridCacheMode cacheMode();
+
+ /**
+ * @return Cache atomicity mode.
+ */
+ protected abstract GridCacheAtomicityMode atomicityMode();
+
+ /**
+ * @return Partitioned mode.
+ */
+ protected abstract GridCacheDistributionMode distributionMode();
+
+ /**
+ * @return Write synchronization.
+ */
+ protected GridCacheWriteSynchronizationMode writeSynchronization() {
+ return FULL_SYNC;
+ }
+
+ /**
+ * @return Whether portable mode is enabled.
+ */
+ protected boolean portableEnabled() {
+ return false;
+ }
+
+ /**
+ * @return {@code true} if swap should be enabled.
+ */
+ protected boolean swapEnabled() {
+ return false;
+ }
+
+ /**
+ * @return Cache.
+ */
+ protected <K, V> IgniteCache<K, V> jcache() {
+ return jcache(0);
+ }
+
+ /**
+ * @param idx Grid index.
+ * @return Cache.
+ */
+ protected <K, V> IgniteCache<K, V> jcache(int idx) {
+ return grid(idx).jcache(null);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheTest.java
deleted file mode 100644
index ef0b4c3..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheTest.java
+++ /dev/null
@@ -1,124 +0,0 @@
-/* @java.file.header */
-
-/* _________ _____ __________________ _____
- * __ ____/___________(_)______ /__ ____/______ ____(_)_______
- * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
- * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
- * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
- */
-
-package org.apache.ignite.internal.processors.cache;
-
-import org.apache.ignite.*;
-import org.apache.ignite.configuration.*;
-import org.apache.ignite.marshaller.optimized.*;
-import org.apache.ignite.spi.discovery.tcp.*;
-import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
-import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
-import org.gridgain.grid.cache.*;
-import org.gridgain.testframework.junits.common.*;
-
-/**
- *
- */
-public class IgniteCacheTest extends GridCommonAbstractTest {
- /** */
- private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
-
- /**
- * @return Grids count to start.
- */
- protected int gridCount() {
- return 2;
- }
-
- /** {@inheritDoc} */
- @Override protected void beforeTestsStarted() throws Exception {
- startGrids();
- }
-
- /**
- * @throws Exception If failed.
- */
- protected void startGrids() throws Exception {
- int cnt = gridCount();
-
- assert cnt >= 1 : "At least one grid must be started";
-
- startGridsMultiThreaded(cnt);
-
- awaitPartitionMapExchange();
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTestsStopped() throws Exception {
- stopAllGrids();
- }
-
- /** {@inheritDoc} */
- @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
- IgniteConfiguration cfg = super.getConfiguration(gridName);
-
- TcpDiscoverySpi disco = new TcpDiscoverySpi();
-
- disco.setMaxMissedHeartbeats(Integer.MAX_VALUE);
-
- disco.setIpFinder(ipFinder);
-
- if (isDebug())
- disco.setAckTimeout(Integer.MAX_VALUE);
-
- cfg.setDiscoverySpi(disco);
-
- cfg.setCacheConfiguration(cacheConfiguration(gridName));
-
- cfg.setMarshaller(new IgniteOptimizedMarshaller(false));
-
- return cfg;
- }
-
- /**
- * @param gridName Grid name.
- * @return Cache configuration.
- * @throws Exception In case of error.
- */
- protected GridCacheConfiguration cacheConfiguration(String gridName) throws Exception {
- GridCacheConfiguration cfg = defaultCacheConfiguration();
-
- return cfg;
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testPutGetRemove() throws Exception {
- IgniteCache<Integer, String> cache = jcache();
-
- for (int i = 0; i < 10; i++)
- cache.put(i, String.valueOf(i));
-
- for (int i = 0; i < 10; i++)
- assertEquals(String.valueOf(i), cache.get(i));
-
- for (int i = 0; i < 10; i++)
- cache.remove(i);
-
- for (int i = 0; i < 10; i++)
- assertNull(cache.get(i));
- }
-
- /**
- * @return Cache.
- */
- protected <K, V> IgniteCache<K, V> jcache() {
- return jcache(0);
- }
-
- /**
- * @param idx Grid index.
- * @return Cache.
- */
- protected <K, V> IgniteCache<K, V> jcache(int idx) {
- return grid(idx).jcache(null);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheAtomicExpiryPolicyTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheAtomicExpiryPolicyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheAtomicExpiryPolicyTest.java
new file mode 100644
index 0000000..251bc05
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheAtomicExpiryPolicyTest.java
@@ -0,0 +1,41 @@
+/* @java.file.header */
+
+/* _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache.expiry;
+
+import org.gridgain.grid.cache.*;
+
+import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
+import static org.gridgain.grid.cache.GridCacheDistributionMode.*;
+import static org.gridgain.grid.cache.GridCacheMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheAtomicExpiryPolicyTest extends IgniteCacheExpiryPolicyAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected int gridCount() {
+ return 3;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheMode cacheMode() {
+ return PARTITIONED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheAtomicityMode atomicityMode() {
+ return ATOMIC;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheDistributionMode distributionMode() {
+ return PARTITIONED_ONLY;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheAtomicLocalExpiryPolicyTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheAtomicLocalExpiryPolicyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheAtomicLocalExpiryPolicyTest.java
new file mode 100644
index 0000000..20c9666
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheAtomicLocalExpiryPolicyTest.java
@@ -0,0 +1,41 @@
+/* @java.file.header */
+
+/* _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache.expiry;
+
+import org.gridgain.grid.cache.*;
+
+import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
+import static org.gridgain.grid.cache.GridCacheDistributionMode.*;
+import static org.gridgain.grid.cache.GridCacheMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheAtomicLocalExpiryPolicyTest extends IgniteCacheExpiryPolicyAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected int gridCount() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheMode cacheMode() {
+ return LOCAL;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheAtomicityMode atomicityMode() {
+ return ATOMIC;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheDistributionMode distributionMode() {
+ return PARTITIONED_ONLY;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheAtomicReplicatedExpiryPolicyTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheAtomicReplicatedExpiryPolicyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheAtomicReplicatedExpiryPolicyTest.java
new file mode 100644
index 0000000..bca7705
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheAtomicReplicatedExpiryPolicyTest.java
@@ -0,0 +1,24 @@
+/* @java.file.header */
+
+/* _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache.expiry;
+
+import org.gridgain.grid.cache.*;
+
+import static org.gridgain.grid.cache.GridCacheMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheAtomicReplicatedExpiryPolicyTest extends IgniteCacheAtomicExpiryPolicyTest {
+ /** {@inheritDoc} */
+ @Override protected GridCacheMode cacheMode() {
+ return REPLICATED;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyAbstractTest.java
new file mode 100644
index 0000000..c8abd0e
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyAbstractTest.java
@@ -0,0 +1,714 @@
+/* @java.file.header */
+
+/* _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache.expiry;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cluster.*;
+import org.apache.ignite.internal.processors.cache.*;
+import org.gridgain.grid.cache.*;
+import org.gridgain.grid.kernal.*;
+import org.gridgain.grid.kernal.processors.cache.*;
+import org.gridgain.grid.util.lang.*;
+import org.gridgain.grid.util.typedef.internal.*;
+import org.gridgain.testframework.*;
+import org.jetbrains.annotations.*;
+
+import javax.cache.configuration.*;
+import javax.cache.expiry.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
+import static org.gridgain.grid.cache.GridCacheDistributionMode.*;
+import static org.gridgain.grid.cache.GridCacheMode.*;
+import static org.gridgain.grid.cache.GridCacheTxConcurrency.*;
+import static org.gridgain.grid.cache.GridCacheTxIsolation.*;
+
+/**
+ *
+ */
+public abstract class IgniteCacheExpiryPolicyAbstractTest extends IgniteCacheAbstractTest {
+ /** */
+ private Factory<? extends ExpiryPolicy> factory;
+
+ /** */
+ private boolean nearCache;
+
+ /** */
+ private Integer lastKey = 0;
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testEternal() throws Exception {
+ factory = EternalExpiryPolicy.factoryOf();
+
+ startGrids();
+
+ for (final Integer key : keys()) {
+ log.info("Test eternalPolicy, key: " + key);
+
+ eternal(key);
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNullFactory() throws Exception {
+ factory = null;
+
+ startGrids();
+
+ for (final Integer key : keys()) {
+ log.info("Test eternalPolicy, key: " + key);
+
+ eternal(key);
+ }
+ }
+
+ /**
+ * @param key Key.
+ * @throws Exception If failed.
+ */
+ private void eternal(Integer key) throws Exception {
+ IgniteCache<Integer, Integer> cache = jcache();
+
+ cache.put(key, 1); // Create.
+
+ checkTtl(key, 0);
+
+ assertEquals((Integer) 1, cache.get(key)); // Get.
+
+ checkTtl(key, 0);
+
+ cache.put(key, 2); // Update.
+
+ checkTtl(key, 0);
+
+ assertTrue(cache.remove(key)); // Remove.
+
+ /*
+ cache.withExpiryPolicy(new TestPolicy(60_000L, null, null)).put(key, 1); // Create with custom.
+
+ checkTtl(key, 60_000L);
+
+ cache.put(key, 2); // Update.
+
+ checkTtl(key, 0);
+
+ cache.withExpiryPolicy(new TestPolicy(null, 1000L, null)).put(key, 1);
+
+ checkTtl(key, 1000L);
+
+ waitExpired(key);
+ */
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testCreateUpdate() throws Exception {
+ factory = new FactoryBuilder.SingletonFactory<>(new TestPolicy(60_000L, 61_000L, null));
+
+ startGrids();
+
+ for (final Integer key : keys()) {
+ log.info("Test createUpdate [key=" + key + ']');
+
+ createUpdate(key, null);
+ }
+
+ for (final Integer key : keys()) {
+ log.info("Test createUpdateCustomPolicy [key=" + key + ']');
+
+ createUpdateCustomPolicy(key, null);
+ }
+
+ createUpdatePutAll(null);
+
+ GridCacheTxConcurrency[] txModes = {PESSIMISTIC};
+
+ if (atomicityMode() == TRANSACTIONAL) {
+ for (GridCacheTxConcurrency tx : txModes) {
+ for (final Integer key : keys()) {
+ log.info("Test createUpdate [key=" + key + ", tx=" + tx + ']');
+
+ createUpdate(key, tx);
+ }
+
+ for (final Integer key : keys()) {
+ log.info("Test createUpdateCustomPolicy [key=" + key + ", tx=" + tx + ']');
+
+ createUpdateCustomPolicy(key, tx);
+ }
+
+ createUpdatePutAll(tx);
+ }
+ }
+ }
+
+ /**
+ * @param txConcurrency Not null transaction concurrency mode if explicit transaction should be started.
+ * @throws Exception If failed.
+ */
+ private void createUpdatePutAll(@Nullable GridCacheTxConcurrency txConcurrency) throws Exception {
+ Map<Integer, Integer> vals = new HashMap<>();
+
+ for (int i = 0; i < 1000; i++)
+ vals.put(i, i);
+
+ IgniteCache<Integer, Integer> cache = jcache(0);
+
+ cache.removeAll(vals.keySet());
+
+ GridCacheTx tx = startTx(txConcurrency);
+
+ // Create.
+ cache.putAll(vals);
+
+ if (tx != null)
+ tx.commit();
+
+ for (Integer key : vals.keySet())
+ checkTtl(key, 60_000);
+
+ tx = startTx(txConcurrency);
+
+ // Update.
+ cache.putAll(vals);
+
+ if (tx != null)
+ tx.commit();
+
+ for (Integer key : vals.keySet())
+ checkTtl(key, 61_000);
+
+ tx = startTx(txConcurrency);
+
+ // Update with provided TTL.
+ cache.withExpiryPolicy(new TestPolicy(null, 1000L, null)).putAll(vals);
+
+ if (tx != null)
+ tx.commit();
+
+ for (Integer key : vals.keySet())
+ checkTtl(key, 1000L);
+
+ waitExpired(vals.keySet());
+
+ tx = startTx(txConcurrency);
+
+ // Try create again.
+ cache.putAll(vals);
+
+ if (tx != null)
+ tx.commit();
+
+ for (Integer key : vals.keySet())
+ checkTtl(key, 60_000L);
+
+ Map<Integer, Integer> newVals = new HashMap<>(vals);
+
+ newVals.put(100_000, 1);
+
+ // Updates and create.
+ cache.putAll(newVals);
+
+ for (Integer key : vals.keySet())
+ checkTtl(key, 61_000L);
+
+ checkTtl(100_000, 60_000L);
+
+ cache.removeAll(newVals.keySet());
+ }
+
+ /**
+ * @param key Key.
+ * @param txConcurrency Not null transaction concurrency mode if explicit transaction should be started.
+ * @throws Exception If failed.
+ */
+ private void createUpdateCustomPolicy(Integer key, @Nullable GridCacheTxConcurrency txConcurrency)
+ throws Exception {
+ IgniteCache<Integer, Integer> cache = jcache();
+
+ assertNull(cache.get(key));
+
+ GridCacheTx tx = startTx(txConcurrency);
+
+ cache.withExpiryPolicy(new TestPolicy(10_000L, 20_000L, 30_000L)).put(key, 1);
+
+ if (tx != null)
+ tx.commit();
+
+ checkTtl(key, 10_000L);
+
+ for (int idx = 0; idx < gridCount(); idx++) {
+ assertEquals(1, cache(idx).get(key)); // Try get.
+
+ checkTtl(key, 10_000);
+ }
+
+ tx = startTx(txConcurrency);
+
+ // Update, returns null duration, should not change TTL.
+ cache.withExpiryPolicy(new TestPolicy(20_000L, null, null)).put(key, 2);
+
+ if (tx != null)
+ tx.commit();
+
+ checkTtl(key, 10_000L);
+
+ tx = startTx(txConcurrency);
+
+ // Update with provided TTL.
+ cache.withExpiryPolicy(new TestPolicy(null, 1000L, null)).put(key, 2);
+
+ if (tx != null)
+ tx.commit();
+
+ checkTtl(key, 1000L);
+
+ waitExpired(key);
+
+ tx = startTx(txConcurrency);
+
+ // Create, returns null duration, should create with 0 TTL.
+ cache.withExpiryPolicy(new TestPolicy(null, 20_000L, 30_000L)).put(key, 1);
+
+ if (tx != null)
+ tx.commit();
+
+ checkTtl(key, 0L);
+ }
+
+ public void _testPrimary() throws Exception {
+ factory = new FactoryBuilder.SingletonFactory<>(new TestPolicy(60_000L, 61_000L, null));
+
+ nearCache = true;
+
+ boolean inTx = true;
+
+ startGrids();
+
+ IgniteCache<Integer, Integer> cache = jcache(0);
+
+ GridCache<Integer, Object> cache0 = cache(0);
+
+ Integer key = primaryKey(cache0);
+
+ log.info("Create: " + key);
+
+ GridCacheTx tx = inTx ? grid(0).transactions().txStart(OPTIMISTIC, READ_COMMITTED) : null;
+
+ cache.put(key, 1);
+
+ if (tx != null)
+ tx.commit();
+
+ checkTtl(key, 60_000);
+
+ tx = inTx ? grid(0).transactions().txStart(OPTIMISTIC, READ_COMMITTED) : null;
+
+ log.info("Update: " + key);
+
+ cache.put(key, 2);
+
+ if (tx != null)
+ tx.commit();
+
+ checkTtl(key, 61_000);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void _test1() throws Exception {
+ factory = new FactoryBuilder.SingletonFactory<>(new TestPolicy(60_000L, 61_000L, null));
+
+ nearCache = false;
+
+ boolean inTx = true;
+
+ startGrids();
+
+ Collection<Integer> keys = keys();
+
+ IgniteCache<Integer, Integer> cache = jcache(0);
+
+ for (final Integer key : keys) {
+ log.info("Test key1: " + key);
+
+ GridCacheTx tx = inTx ? grid(0).transactions().txStart(OPTIMISTIC, READ_COMMITTED) : null;
+
+ cache.put(key, 1);
+
+ if (tx != null)
+ tx.commit();
+
+ log.info("Test key2: " + key);
+
+ tx = inTx ? grid(0).transactions().txStart(OPTIMISTIC, READ_COMMITTED) : null;
+
+ cache.put(key, 2);
+
+ if (tx != null)
+ tx.commit();
+
+ log.info("Done");
+ }
+ }
+
+ /**
+ * @param key Key.
+ * @param txConcurrency Not null transaction concurrency mode if explicit transaction should be started.
+ * @throws Exception If failed.
+ */
+ private void createUpdate(Integer key, @Nullable GridCacheTxConcurrency txConcurrency)
+ throws Exception {
+ IgniteCache<Integer, Integer> cache = jcache();
+
+ // Run several times to make sure create after remove works as expected.
+ for (int i = 0; i < 3; i++) {
+ log.info("Iteration: " + i);
+
+ GridCacheTx tx = startTx(txConcurrency);
+
+ cache.put(key, 1); // Create.
+
+ if (tx != null)
+ tx.commit();
+
+ checkTtl(key, 60_000);
+
+ for (int idx = 0; idx < gridCount(); idx++) {
+ assertEquals(1, cache(idx).get(key)); // Try get.
+
+ checkTtl(key, 60_000);
+ }
+
+ tx = startTx(txConcurrency);
+
+ cache.put(key, 2); // Update.
+
+ if (tx != null)
+ tx.commit();
+
+ checkTtl(key, 61_000);
+
+ for (int idx = 0; idx < gridCount(); idx++) {
+ assertEquals(2, cache(idx).get(key)); // Try get.
+
+ checkTtl(key, 61_000);
+ }
+
+ tx = startTx(txConcurrency);
+
+ assertTrue(cache.remove(key));
+
+ if (tx != null)
+ tx.commit();
+
+ for (int idx = 0; idx < gridCount(); idx++)
+ assertNull(cache(idx).get(key));
+ }
+ }
+
+ /**
+ * @param txConcurrency Transaction concurrency mode.
+ * @return Transaction.
+ */
+ @Nullable private GridCacheTx startTx(@Nullable GridCacheTxConcurrency txConcurrency) {
+ return txConcurrency == null ? null : ignite(0).transactions().txStart(txConcurrency, READ_COMMITTED);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNearCreateUpdate() throws Exception {
+ if (cacheMode() != PARTITIONED)
+ return;
+
+ factory = new FactoryBuilder.SingletonFactory<>(new TestPolicy(60_000L, 61_000L, null));
+
+ nearCache = true;
+
+ startGrids();
+
+ Integer key = nearKey(cache(0));
+
+ IgniteCache<Integer, Integer> jcache0 = jcache(0);
+
+ jcache0.put(key, 1);
+
+ checkTtl(key, 60_000);
+
+ IgniteCache<Integer, Integer> jcache1 = jcache(1);
+
+ // Update from another node.
+ jcache1.put(key, 2);
+
+ checkTtl(key, 61_000L);
+
+ // Update from another node with provided TTL.
+ jcache1.withExpiryPolicy(new TestPolicy(null, 1000L, null)).put(key, 3);
+
+ checkTtl(key, 1000);
+
+ waitExpired(key);
+
+ // Try create again.
+ jcache0.put(key, 1);
+
+ checkTtl(key, 60_000);
+
+ // Update from near node with provided TTL.
+ jcache0.withExpiryPolicy(new TestPolicy(null, 1100L, null)).put(key, 2);
+
+ checkTtl(key, 1100);
+
+ waitExpired(key);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNearPutAll() throws Exception {
+ if (cacheMode() != PARTITIONED)
+ return;
+
+ factory = new FactoryBuilder.SingletonFactory<>(new TestPolicy(60_000L, 61_000L, null));
+
+ nearCache = true;
+
+ startGrids();
+
+ Map<Integer, Integer> vals = new HashMap<>();
+
+ for (int i = 0; i < 1000; i++)
+ vals.put(i, i);
+
+ IgniteCache<Integer, Integer> jcache0 = jcache(0);
+
+ jcache0.putAll(vals);
+
+ for (Integer key : vals.keySet())
+ checkTtl(key, 60_000);
+
+ IgniteCache<Integer, Integer> jcache1 = jcache(1);
+
+ // Update from another node.
+ jcache1.putAll(vals);
+
+ for (Integer key : vals.keySet())
+ checkTtl(key, 61_000);
+
+ // Update from another node with provided TTL.
+ jcache1.withExpiryPolicy(new TestPolicy(null, 1000L, null)).putAll(vals);
+
+ for (Integer key : vals.keySet())
+ checkTtl(key, 1000);
+
+ waitExpired(vals.keySet());
+
+ // Try create again.
+ jcache0.putAll(vals);
+
+ // Update from near node with provided TTL.
+ jcache1.withExpiryPolicy(new TestPolicy(null, 1101L, null)).putAll(vals);
+
+ for (Integer key : vals.keySet())
+ checkTtl(key, 1101L);
+
+ waitExpired(vals.keySet());
+ }
+
+ /**
+ * @return Test keys.
+ * @throws Exception If failed.
+ */
+ private Collection<Integer> keys() throws Exception {
+ GridCache<Integer, Object> cache = cache(0);
+
+ ArrayList<Integer> keys = new ArrayList<>();
+
+ keys.add(primaryKeys(cache, 1, lastKey).get(0));
+
+ if (gridCount() > 1) {
+ keys.add(backupKeys(cache, 1, lastKey).get(0));
+
+ if (cache.configuration().getCacheMode() != REPLICATED)
+ keys.add(nearKeys(cache, 1, lastKey).get(0));
+ }
+
+ lastKey = Collections.max(keys) + 1;
+
+ return keys;
+ }
+
+ /**
+ * @param key Key.
+ * @throws Exception If failed.
+ */
+ private void waitExpired(Integer key) throws Exception {
+ waitExpired(Collections.singleton(key));
+ }
+
+ /**
+ * @param keys Keys.
+ * @throws Exception If failed.
+ */
+ private void waitExpired(final Collection<Integer> keys) throws Exception {
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ for (int i = 0; i < gridCount(); i++) {
+ for (Integer key : keys) {
+ Object val = jcache(i).localPeek(key);
+
+ if (val != null) {
+ // log.info("Value [grid=" + i + ", val=" + val + ']');
+
+ return false;
+ }
+ }
+ }
+
+ return false;
+ }
+ }, 3000);
+
+ GridCache<Integer, Object> cache = cache(0);
+
+ for (int i = 0; i < gridCount(); i++) {
+ ClusterNode node = grid(i).cluster().localNode();
+
+ for (Integer key : keys) {
+ Object val = jcache(i).localPeek(key);
+
+ if (val != null) {
+ log.info("Unexpected value [grid=" + i +
+ ", primary=" + cache.affinity().isPrimary(node, key) +
+ ", backup=" + cache.affinity().isBackup(node, key) + ']');
+ }
+
+ assertNull("Unexpected non-null value for grid " + i, val);
+ }
+ }
+
+ for (int i = 0; i < gridCount(); i++) {
+ for (Integer key : keys)
+ assertNull("Unexpected non-null value for grid " + i, jcache(i).get(key));
+ }
+ }
+
+ /**
+ * @param key Key.
+ * @param ttl TTL.
+ * @throws Exception If failed.
+ */
+ private void checkTtl(Object key, long ttl) throws Exception {
+ boolean found = false;
+
+ for (int i = 0; i < gridCount(); i++) {
+ GridKernal grid = (GridKernal)grid(i);
+
+ GridCacheAdapter<Object, Object> cache = grid.context().cache().internalCache();
+
+ GridCacheEntryEx<Object, Object> e = cache.peekEx(key);
+
+ if (e == null && cache.context().isNear())
+ e = cache.context().near().dht().peekEx(key);
+
+ if (e == null)
+ assertTrue(!cache.affinity().isPrimaryOrBackup(grid.localNode(), key));
+ else {
+ found = true;
+
+ assertEquals("Unexpected ttl for grid " + i, ttl, e.ttl());
+
+ if (ttl > 0)
+ assertTrue(e.expireTime() > 0);
+ else
+ assertEquals(0, e.expireTime());
+ }
+ }
+
+ assertTrue(found);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheConfiguration cacheConfiguration(String gridName) throws Exception {
+ GridCacheConfiguration cfg = super.cacheConfiguration(gridName);
+
+ if (nearCache && gridName.equals(getTestGridName(0)))
+ cfg.setDistributionMode(NEAR_PARTITIONED);
+
+ cfg.setExpiryPolicyFactory(factory);
+
+ return cfg;
+ }
+
+ /**
+ *
+ */
+ private class TestPolicy implements ExpiryPolicy {
+ /** */
+ private Long create;
+
+ /** */
+ private Long access;
+
+ /** */
+ private Long update;
+
+ /**
+ * @param create TTL for creation.
+ * @param access TTL for access.
+ * @param update TTL for update.
+ */
+ TestPolicy(@Nullable Long create,
+ @Nullable Long update,
+ @Nullable Long access) {
+ this.create = create;
+ this.update = update;
+ this.access = access;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Duration getExpiryForCreation() {
+ return create != null ? new Duration(TimeUnit.MILLISECONDS, create) : null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Duration getExpiryForAccess() {
+ return access != null ? new Duration(TimeUnit.MILLISECONDS, access) : null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Duration getExpiryForUpdate() {
+ return update != null ? new Duration(TimeUnit.MILLISECONDS, update) : null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(TestPolicy.class, this);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyTest.java
deleted file mode 100644
index f96e5a7..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyTest.java
+++ /dev/null
@@ -1,507 +0,0 @@
-/* @java.file.header */
-
-/* _________ _____ __________________ _____
- * __ ____/___________(_)______ /__ ____/______ ____(_)_______
- * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
- * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
- * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
- */
-
-package org.apache.ignite.internal.processors.cache.expiry;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cluster.*;
-import org.apache.ignite.internal.processors.cache.*;
-import org.gridgain.grid.cache.*;
-import org.gridgain.grid.kernal.*;
-import org.gridgain.grid.kernal.processors.cache.*;
-import org.gridgain.grid.util.lang.*;
-import org.gridgain.grid.util.typedef.internal.*;
-import org.gridgain.testframework.*;
-import org.jetbrains.annotations.*;
-
-import javax.cache.configuration.*;
-import javax.cache.expiry.*;
-import java.util.*;
-import java.util.concurrent.*;
-
-import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
-import static org.gridgain.grid.cache.GridCacheDistributionMode.*;
-import static org.gridgain.grid.cache.GridCacheMode.*;
-
-/**
- *
- */
-public class IgniteCacheExpiryPolicyTest extends IgniteCacheTest {
- /** */
- private Factory<? extends ExpiryPolicy> factory;
-
- /** */
- private boolean nearCache;
-
- /** {@inheritDoc} */
- @Override protected void beforeTestsStarted() throws Exception {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTest() throws Exception {
- stopAllGrids();
- }
-
- /** {@inheritDoc} */
- @Override protected int gridCount() {
- return 3;
- }
-
- public void testPrimary() throws Exception {
- factory = new FactoryBuilder.SingletonFactory<>(new TestPolicy(60_000L, 61_000L, null));
-
- nearCache = false;
-
- boolean inTx = false;
-
- startGrids();
-
- IgniteCache<Integer, Integer> cache = jcache(0);
-
- GridCache<Integer, Object> cache0 = cache(0);
-
- Integer key = primaryKey(cache0);
-
- log.info("Create: " + key);
-
- GridCacheTx tx = inTx ? grid(0).transactions().txStart() : null;
-
- cache.put(key, 1);
-
- if (tx != null)
- tx.commit();
-
- checkTtl(key, 60_000);
-
- tx = inTx ? grid(0).transactions().txStart() : null;
-
- log.info("Update: " + key);
-
- cache.put(key, 2);
-
- if (tx != null)
- tx.commit();
-
- checkTtl(key, 61_000);
- }
-
- public void testBackup() throws Exception {
- factory = new FactoryBuilder.SingletonFactory<>(new TestPolicy(60_000L, 61_000L, null));
-
- nearCache = false;
-
- boolean inTx = false;
-
- startGrids();
-
- IgniteCache<Integer, Integer> cache = jcache(0);
-
- GridCache<Integer, Object> cache0 = cache(0);
-
- Integer key = backupKey(cache0);
-
- log.info("Create: " + key);
-
- GridCacheTx tx = inTx ? grid(0).transactions().txStart() : null;
-
- cache.put(key, 1);
-
- if (tx != null)
- tx.commit();
-
- checkTtl(key, 60_000);
-
- tx = inTx ? grid(0).transactions().txStart() : null;
-
- log.info("Update: " + key);
-
- cache.put(key, 2);
-
- if (tx != null)
- tx.commit();
-
- checkTtl(key, 61_000);
- }
-
- public void testNear() throws Exception {
- factory = new FactoryBuilder.SingletonFactory<>(new TestPolicy(60_000L, 61_000L, null));
-
- nearCache = false;
-
- boolean inTx = true;
-
- startGrids();
-
- IgniteCache<Integer, Integer> cache = jcache(0);
-
- GridCache<Integer, Object> cache0 = cache(0);
-
- Integer key = nearKey(cache0);
-
- log.info("Create: " + key);
-
- GridCacheTx tx = inTx ? grid(0).transactions().txStart() : null;
-
- cache.put(key, 1);
-
- if (tx != null)
- tx.commit();
-
- checkTtl(key, 60_000);
-
- tx = inTx ? grid(0).transactions().txStart() : null;
-
- log.info("Update: " + key);
-
- cache.put(key, 2);
-
- if (tx != null)
- tx.commit();
-
- checkTtl(key, 61_000);
- }
-
- /**
- * @throws Exception If failed.
- */
- public void test1() throws Exception {
- factory = new FactoryBuilder.SingletonFactory<>(new TestPolicy(60_000L, null, null));
-
- nearCache = false;
-
- boolean inTx = true;
-
- startGrids();
-
- Collection<Integer> keys = keys();
-
- IgniteCache<Integer, Integer> cache = jcache(0);
-
- for (final Integer key : keys) {
- log.info("Test key1: " + key);
-
- GridCacheTx tx = inTx ? grid(0).transactions().txStart() : null;
-
- cache.put(key, 1);
-
- if (tx != null)
- tx.commit();
- }
-
- for (final Integer key : keys) {
- log.info("Test key2: " + key);
-
- GridCacheTx tx = inTx ? grid(0).transactions().txStart() : null;
-
- cache.put(key, 2);
-
- if (tx != null)
- tx.commit();
- }
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testCreated() throws Exception {
- factory = new FactoryBuilder.SingletonFactory<>(new TestPolicy(60_000L, null, null));
-
- startGrids();
-
- Collection<Integer> keys = keys();
-
- IgniteCache<Integer, Integer> cache = jcache(0);
-
- for (final Integer key : keys) {
- log.info("Test key: " + key);
-
- cache.put(key, 1);
-
- checkTtl(key, 60_000);
-
- for (int i = 0; i < gridCount(); i++) {
- assertEquals((Integer)1, cache.get(key));
-
- checkTtl(key, 60_000);
- }
-
- cache.withExpiryPolicy(new TestPolicy(1000L, null, null)).put(key, 2); // Update, should not change TTL.
-
- checkTtl(key, 60_000);
-
- assertEquals((Integer)2, cache.get(key));
-
- assertTrue(cache.remove(key));
-
- cache.withExpiryPolicy(new TestPolicy(1000L, null, null)).put(key, 3); // Create with provided TTL.
-
- checkTtl(key, 1000);
-
- waitExpired(key);
- }
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testNearPut() throws Exception {
- factory = new FactoryBuilder.SingletonFactory<>(new TestPolicy(60_000L, null, null));
-
- nearCache = true;
-
- startGrids();
-
- GridCache<Integer, Object> cache0 = cache(0);
-
- Integer key = nearKey(cache0);
-
- IgniteCache<Integer, Integer> jcache0 = jcache(0);
-
- jcache0.put(key, 1);
-
- checkTtl(key, 60_000);
-
- IgniteCache<Integer, Integer> jcache1 = jcache(1);
-
- // Update from another node with provided TTL.
- jcache1.withExpiryPolicy(new TestPolicy(null, 1000L, null)).put(key, 2);
-
- checkTtl(key, 1000);
-
- waitExpired(key);
-
- jcache1.remove(key);
-
- jcache0.put(key, 1);
-
- checkTtl(key, 60_000);
-
- // Update from near node with provided TTL.
- jcache0.withExpiryPolicy(new TestPolicy(null, 1100L, null)).put(key, 2);
-
- checkTtl(key, 1100);
-
- waitExpired(key);
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testNearPutAll() throws Exception {
- factory = new FactoryBuilder.SingletonFactory<>(new TestPolicy(60_000L, null, null));
-
- nearCache = true;
-
- startGrids();
-
- Map<Integer, Integer> vals = new HashMap<>();
-
- for (int i = 0; i < 1000; i++)
- vals.put(i, i);
-
- IgniteCache<Integer, Integer> jcache0 = jcache(0);
-
- jcache0.putAll(vals);
-
- for (Integer key : vals.keySet())
- checkTtl(key, 60_000);
-
- IgniteCache<Integer, Integer> jcache1 = jcache(1);
-
- // Update from another node with provided TTL.
- jcache1.withExpiryPolicy(new TestPolicy(null, 1000L, null)).putAll(vals);
-
- for (Integer key : vals.keySet())
- checkTtl(key, 1000);
-
- waitExpired(vals.keySet());
-
- jcache0.removeAll(vals.keySet());
-
- jcache0.putAll(vals);
-
- // Update from near node with provided TTL.
- jcache1.withExpiryPolicy(new TestPolicy(null, 1101L, null)).putAll(vals);
-
- for (Integer key : vals.keySet())
- checkTtl(key, 1101L);
-
- waitExpired(vals.keySet());
- }
-
- /**
- * @return Test keys.
- * @throws Exception If failed.
- */
- private Collection<Integer> keys() throws Exception {
- GridCache<Integer, Object> cache = cache(0);
-
- Collection<Integer> keys = new ArrayList<>();
-
- keys.add(primaryKey(cache));
-
- if (gridCount() > 1) {
- keys.add(backupKey(cache));
-
- if (cache.configuration().getCacheMode() != REPLICATED)
- keys.add(nearKey(cache));
- }
-
- return keys;
- }
-
- /**
- * @param key Key.
- * @throws Exception If failed.
- */
- private void waitExpired(Integer key) throws Exception {
- waitExpired(Collections.singleton(key));
- }
-
- /**
- * @param keys Keys.
- * @throws Exception If failed.
- */
- private void waitExpired(final Collection<Integer> keys) throws Exception {
- GridTestUtils.waitForCondition(new GridAbsPredicate() {
- @Override public boolean apply() {
- for (int i = 0; i < gridCount(); i++) {
- for (Integer key : keys) {
- Object val = jcache(i).localPeek(key);
-
- if (val != null) {
- // log.info("Value [grid=" + i + ", val=" + val + ']');
-
- return false;
- }
- }
- }
-
- return false;
- }
- }, 3000);
-
- GridCache<Integer, Object> cache = cache(0);
-
- for (int i = 0; i < gridCount(); i++) {
- ClusterNode node = grid(i).cluster().localNode();
-
- for (Integer key : keys) {
- Object val = jcache(i).localPeek(key);
-
- if (val != null) {
- log.info("Unexpected value [grid=" + i +
- ", primary=" + cache.affinity().isPrimary(node, key) +
- ", backup=" + cache.affinity().isBackup(node, key) + ']');
- }
-
- assertNull("Unexpected non-null value for grid " + i, val);
- }
- }
-
- for (int i = 0; i < gridCount(); i++) {
- for (Integer key : keys)
- assertNull("Unexpected non-null value for grid " + i, jcache(i).get(key));
- }
- }
-
- /**
- * @param key Key.
- * @param ttl TTL.
- * @throws Exception If failed.
- */
- private void checkTtl(Object key, long ttl) throws Exception {
- for (int i = 0; i < gridCount(); i++) {
- GridKernal grid = (GridKernal)grid(i);
-
- GridCacheAdapter<Object, Object> cache = grid.context().cache().internalCache();
-
- GridCacheEntryEx<Object, Object> e = cache.peekEx(key);
-
- if (e == null && cache.context().isNear())
- e = cache.context().near().dht().peekEx(key);
-
- if (e == null)
- assertTrue(!cache.affinity().isPrimaryOrBackup(grid.localNode(), key));
- else
- assertEquals("Unexpected ttl for grid " + i, ttl, e.ttl());
- }
- }
-
- /** {@inheritDoc} */
- @Override protected GridCacheConfiguration cacheConfiguration(String gridName) throws Exception {
- assert factory != null;
-
- GridCacheConfiguration cfg = super.cacheConfiguration(gridName);
-
- cfg.setCacheMode(PARTITIONED);
- cfg.setAtomicityMode(TRANSACTIONAL);
-
- //cfg.setAtomicityMode(ATOMIC);
-
- cfg.setBackups(1);
-
- if (nearCache && gridName.equals(getTestGridName(0)))
- cfg.setDistributionMode(NEAR_PARTITIONED);
- else
- cfg.setDistributionMode(PARTITIONED_ONLY);
-
- cfg.setExpiryPolicyFactory(factory);
-
- return cfg;
- }
-
- /**
- *
- */
- private class TestPolicy implements ExpiryPolicy {
- /** */
- private Long create;
-
- /** */
- private Long access;
-
- /** */
- private Long update;
-
- /**
- * @param create TTL for creation.
- * @param access TTL for access.
- * @param update TTL for update.
- */
- TestPolicy(@Nullable Long create,
- @Nullable Long update,
- @Nullable Long access) {
- this.create = create;
- this.update = update;
- this.access = access;
- }
-
- /** {@inheritDoc} */
- @Override public Duration getExpiryForCreation() {
- return create != null ? new Duration(TimeUnit.MILLISECONDS, create) : null;
- }
-
- /** {@inheritDoc} */
- @Override public Duration getExpiryForAccess() {
- return access != null ? new Duration(TimeUnit.MILLISECONDS, access) : null;
- }
-
- /** {@inheritDoc} */
- @Override public Duration getExpiryForUpdate() {
- return update != null ? new Duration(TimeUnit.MILLISECONDS, update) : null;
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(TestPolicy.class, this);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyTestSuite.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyTestSuite.java
new file mode 100644
index 0000000..fd2d205
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyTestSuite.java
@@ -0,0 +1,35 @@
+/* @java.file.header */
+
+/* _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache.expiry;
+
+import junit.framework.*;
+
+/**
+ *
+ */
+public class IgniteCacheExpiryPolicyTestSuite extends TestSuite {
+ /**
+ * @return Cache API test suite.
+ * @throws Exception If failed.
+ */
+ public static TestSuite suite() throws Exception {
+ TestSuite suite = new TestSuite("Cache Expiry Policy Test Suite");
+
+ suite.addTestSuite(IgniteCacheAtomicLocalExpiryPolicyTest.class);
+ suite.addTestSuite(IgniteCacheAtomicExpiryPolicyTest.class);
+ suite.addTestSuite(IgniteCacheAtomicReplicatedExpiryPolicyTest.class);
+
+ suite.addTestSuite(IgniteCacheTxLocalExpiryPolicyTest.class);
+ suite.addTestSuite(IgniteCacheTxExpiryPolicyTest.class);
+ suite.addTestSuite(IgniteCacheTxReplicatedExpiryPolicyTest.class);
+
+ return suite;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d84da7d0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheTxExpiryPolicyTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheTxExpiryPolicyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheTxExpiryPolicyTest.java
new file mode 100644
index 0000000..0abe27d
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheTxExpiryPolicyTest.java
@@ -0,0 +1,41 @@
+/* @java.file.header */
+
+/* _________ _____ __________________ _____
+ * __ ____/___________(_)______ /__ ____/______ ____(_)_______
+ * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \
+ * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / /
+ * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache.expiry;
+
+import org.gridgain.grid.cache.*;
+
+import static org.gridgain.grid.cache.GridCacheAtomicityMode.*;
+import static org.gridgain.grid.cache.GridCacheDistributionMode.*;
+import static org.gridgain.grid.cache.GridCacheMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheTxExpiryPolicyTest extends IgniteCacheExpiryPolicyAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected int gridCount() {
+ return 3;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheMode cacheMode() {
+ return PARTITIONED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheAtomicityMode atomicityMode() {
+ return TRANSACTIONAL;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected GridCacheDistributionMode distributionMode() {
+ return PARTITIONED_ONLY;
+ }
+}