ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [44/50] [abbrv] incubator-ignite git commit: # ignite-41
Date Mon, 22 Dec 2014 10:42:46 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f537940c/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java
index 0000000,4fc7140..a126460
mode 000000,100644..100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@@ -1,0 -1,3179 +1,3285 @@@
+ /* @java.file.header */
+ 
+ /*  _________        _____ __________________        _____
+  *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
+  *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
+  *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
+  *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
+  */
+ 
+ package org.gridgain.grid.kernal.processors.cache.transactions;
+ 
+ import org.apache.ignite.*;
+ import org.apache.ignite.lang.*;
+ import org.apache.ignite.plugin.security.*;
+ import org.apache.ignite.portables.*;
+ import org.apache.ignite.transactions.*;
+ import org.gridgain.grid.cache.*;
+ import org.gridgain.grid.kernal.processors.cache.*;
+ import org.gridgain.grid.kernal.processors.cache.distributed.near.*;
+ import org.gridgain.grid.kernal.processors.cache.dr.*;
+ import org.gridgain.grid.kernal.processors.dr.*;
+ import org.gridgain.grid.util.*;
+ import org.gridgain.grid.util.future.*;
+ 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.*;
+ 
+ import static org.apache.ignite.events.IgniteEventType.*;
+ import static org.apache.ignite.transactions.IgniteTxState.*;
+ import static org.gridgain.grid.kernal.processors.cache.GridCacheOperation.*;
+ import static org.gridgain.grid.kernal.processors.dr.GridDrType.*;
+ 
+ /**
+  * Transaction adapter for cache transactions.
+  */
+ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V>
+     implements IgniteTxLocalEx<K, V> {
+     /** */
+     private static final long serialVersionUID = 0L;
+ 
+     /** Per-transaction read map. */
+     @GridToStringExclude
+     protected Map<IgniteTxKey<K>, IgniteTxEntry<K, V>> txMap;
+ 
+     /** Read view on transaction map. */
+     @GridToStringExclude
+     protected IgniteTxMap<K, V> readView;
+ 
+     /** Write view on transaction map. */
+     @GridToStringExclude
+     protected IgniteTxMap<K, V> writeView;
+ 
+     /** Minimal version encountered (either explicit lock or XID of this transaction). */
+     protected GridCacheVersion minVer;
+ 
+     /** Flag indicating with TM commit happened. */
+     protected AtomicBoolean doneFlag = new AtomicBoolean(false);
+ 
+     /** Committed versions, relative to base. */
+     private Collection<GridCacheVersion> committedVers = Collections.emptyList();
+ 
+     /** Rolled back versions, relative to base. */
+     private Collection<GridCacheVersion> rolledbackVers = Collections.emptyList();
+ 
+     /** Base for completed versions. */
+     private GridCacheVersion completedBase;
+ 
+     /** Flag indicating partition lock in group lock transaction. */
+     private boolean partLock;
+ 
+     /** Flag indicating that transformed values should be sent to remote nodes. */
+     private boolean sndTransformedVals;
+ 
+     /** Commit error. */
+     protected AtomicReference<Throwable> commitErr = new AtomicReference<>();
+ 
+     /** Active cache IDs. */
+     protected Set<Integer> activeCacheIds = new HashSet<>();
+ 
+     /**
+      * Empty constructor required for {@link Externalizable}.
+      */
+     protected IgniteTxLocalAdapter() {
+         // No-op.
+     }
+ 
+     /**
+      * @param cctx Cache registry.
+      * @param xidVer Transaction ID.
+      * @param implicit {@code True} if transaction was implicitly started by the system,
+      *      {@code false} if it was started explicitly by user.
+      * @param implicitSingle {@code True} if transaction is implicit with only one key.
+      * @param sys System flag.
+      * @param concurrency Concurrency.
+      * @param isolation Isolation.
+      * @param timeout Timeout.
+      * @param txSize Expected transaction size.
+      * @param grpLockKey Group lock key if this is a group-lock transaction.
+      * @param partLock {@code True} if this is a group-lock transaction and lock is acquired for whole partition.
+      */
+     protected IgniteTxLocalAdapter(
+         GridCacheSharedContext<K, V> cctx,
+         GridCacheVersion xidVer,
+         boolean implicit,
+         boolean implicitSingle,
+         boolean sys,
+         IgniteTxConcurrency concurrency,
+         IgniteTxIsolation isolation,
+         long timeout,
+         boolean invalidate,
+         boolean storeEnabled,
+         int txSize,
+         @Nullable IgniteTxKey grpLockKey,
+         boolean partLock,
+         @Nullable UUID subjId,
+         int taskNameHash
+     ) {
+         super(cctx, xidVer, implicit, implicitSingle, /*local*/true, sys, concurrency, isolation, timeout, invalidate,
+             storeEnabled, txSize, grpLockKey, subjId, taskNameHash);
+ 
+         assert !partLock || grpLockKey != null;
+ 
+         this.partLock = partLock;
+ 
+         minVer = xidVer;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public UUID eventNodeId() {
+         return cctx.localNodeId();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public UUID originatingNodeId() {
+         return cctx.localNodeId();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean empty() {
+         return txMap.isEmpty();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public Collection<UUID> masterNodeIds() {
+         return Collections.singleton(nodeId);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean partitionLock() {
+         return partLock;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public Throwable commitError() {
+         return commitErr.get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void commitError(Throwable e) {
+         commitErr.compareAndSet(null, e);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean onOwnerChanged(GridCacheEntryEx<K, V> entry, GridCacheMvccCandidate<K> owner) {
+         assert false;
+         return false;
+     }
+ 
+     /**
+      * Gets collection of active cache IDs for this transaction.
+      *
+      * @return Collection of active cache IDs.
+      */
+     @Override public Collection<Integer> activeCacheIds() {
+         return activeCacheIds;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean isStarted() {
+         return txMap != null;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean hasWriteKey(IgniteTxKey<K> key) {
+         return writeView.containsKey(key);
+     }
+ 
+     /**
+      * @return Transaction read set.
+      */
+     @Override public Set<IgniteTxKey<K>> readSet() {
+         return txMap == null ? Collections.<IgniteTxKey<K>>emptySet() : readView.keySet();
+     }
+ 
+     /**
+      * @return Transaction write set.
+      */
+     @Override public Set<IgniteTxKey<K>> writeSet() {
+         return txMap == null ? Collections.<IgniteTxKey<K>>emptySet() : writeView.keySet();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean removed(IgniteTxKey<K> key) {
+         if (txMap == null)
+             return false;
+ 
+         IgniteTxEntry<K, V> e = txMap.get(key);
+ 
+         return e != null && e.op() == DELETE;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public Map<IgniteTxKey<K>, IgniteTxEntry<K, V>> readMap() {
+         return readView == null ? Collections.<IgniteTxKey<K>, IgniteTxEntry<K, V>>emptyMap() : readView;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public Map<IgniteTxKey<K>, IgniteTxEntry<K, V>> writeMap() {
+         return writeView == null ? Collections.<IgniteTxKey<K>, IgniteTxEntry<K, V>>emptyMap() : writeView;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public Collection<IgniteTxEntry<K, V>> allEntries() {
+         return txMap == null ? Collections.<IgniteTxEntry<K, V>>emptySet() : txMap.values();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public Collection<IgniteTxEntry<K, V>> readEntries() {
+         return readView == null ? Collections.<IgniteTxEntry<K, V>>emptyList() : readView.values();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public Collection<IgniteTxEntry<K, V>> writeEntries() {
+         return writeView == null ? Collections.<IgniteTxEntry<K, V>>emptyList() : writeView.values();
+     }
+ 
+     /** {@inheritDoc} */
+     @Nullable @Override public IgniteTxEntry<K, V> entry(IgniteTxKey<K> key) {
+         return txMap == null ? null : txMap.get(key);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void seal() {
+         if (readView != null)
+             readView.seal();
+ 
+         if (writeView != null)
+             writeView.seal();
+     }
+ 
+     /**
+      * @param snd {@code True} if values in tx entries should be replaced with transformed values and sent
+      * to remote nodes.
+      */
+     public void sendTransformedValues(boolean snd) {
+         sndTransformedVals = snd;
+     }
+ 
+     /**
+      * @return {@code True} if should be committed after lock is acquired.
+      */
+     protected boolean commitAfterLock() {
+         return implicit() && (!dht() || colocated());
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings({"RedundantTypeArguments"})
+     @Nullable @Override public GridTuple<V> peek(
+         GridCacheContext<K, V> cacheCtx,
+         boolean failFast,
+         K key,
+         IgnitePredicate<GridCacheEntry<K, V>>[] filter
+     ) throws GridCacheFilterFailedException {
+         IgniteTxEntry<K, V> e = txMap == null ? null : txMap.get(cacheCtx.txKey(key));
+ 
+         if (e != null) {
+             // We should look at tx entry previous value. If this is a user peek then previous
+             // value is the same as value. If this is a filter evaluation peek then previous value holds
+             // value visible to filter while value contains value enlisted for write.
+             if (!F.isAll(e.cached().wrap(false), filter))
+                 return e.hasPreviousValue() ? F.t(CU.<V>failed(failFast, e.previousValue())) : null;
+ 
+             return e.hasPreviousValue() ? F.t(e.previousValue()) : null;
+         }
+ 
+         return null;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<Boolean> loadMissing(
+         final GridCacheContext<K, V> cacheCtx,
+         boolean async,
+         final Collection<? extends K> keys,
+         boolean deserializePortable,
+         final IgniteBiInClosure<K, V> c
+     ) {
+         if (!async) {
+             try {
+                 return new GridFinishedFuture<>(cctx.kernalContext(),
+                     cacheCtx.store().loadAllFromStore(this, keys, c));
+             }
+             catch (IgniteCheckedException e) {
+                 return new GridFinishedFuture<>(cctx.kernalContext(), e);
+             }
+         }
+         else
+             return cctx.kernalContext().closure().callLocalSafe(
+                 new GPC<Boolean>() {
+                     @Override public Boolean call() throws Exception {
+                         return cacheCtx.store().loadAllFromStore(IgniteTxLocalAdapter.this, keys, c);
+                     }
+                 },
+                 true);
+     }
+ 
+     /**
+      * Gets minimum version present in transaction.
+      *
+      * @return Minimum versions.
+      */
+     @Override public GridCacheVersion minVersion() {
+         return minVer;
+     }
+ 
+     /**
+      * @throws IgniteCheckedException If prepare step failed.
+      */
+     @SuppressWarnings({"CatchGenericClass"})
+     public void userPrepare() throws IgniteCheckedException {
+         if (state() != PREPARING) {
+             if (timedOut())
+                 throw new IgniteTxTimeoutException("Transaction timed out: " + this);
+ 
+             IgniteTxState state = state();
+ 
+             setRollbackOnly();
+ 
+             throw new IgniteCheckedException("Invalid transaction state for prepare [state=" + state + ", tx=" + this + ']');
+         }
+ 
+         checkValid();
+ 
+         try {
+             cctx.tm().prepareTx(this);
+         }
+         catch (IgniteCheckedException e) {
+             throw e;
+         }
+         catch (Throwable e) {
+             setRollbackOnly();
+ 
+             throw new IgniteCheckedException("Transaction validation produced a runtime exception: " + this, e);
+         }
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void commit() throws IgniteCheckedException {
+         try {
+             commitAsync().get();
+         }
+         finally {
+             cctx.tm().txContextReset();
+         }
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void prepare() throws IgniteCheckedException {
+         prepareAsync().get();
+     }
+ 
+     /**
+      * Checks that locks are in proper state for commit.
+      *
+      * @param entry Cache entry to check.
+      */
+     private void checkCommitLocks(GridCacheEntryEx<K, V> entry) {
+         assert ownsLockUnsafe(entry) : "Lock is not owned for commit in PESSIMISTIC mode [entry=" + entry +
+             ", tx=" + this + ']';
+     }
+ 
+     /**
+      * Uncommits transaction by invalidating all of its entries.
+      */
+     @SuppressWarnings({"CatchGenericClass"})
+     private void uncommit() {
+         for (IgniteTxEntry<K, V> e : writeMap().values()) {
+             try {
+                 GridCacheEntryEx<K, V> cacheEntry = e.cached();
+ 
+                 if (e.op() != NOOP)
+                     cacheEntry.invalidate(null, xidVer);
+             }
+             catch (Throwable t) {
+                 U.error(log, "Failed to invalidate transaction entries while reverting a commit.", t);
+ 
+                 break;
+             }
+         }
+ 
+         cctx.tm().uncommitTx(this);
+     }
+ 
+     /**
+      * Gets cache entry for given key.
+      *
+      * @param key Key.
+      * @return Cache entry.
+      */
+     protected GridCacheEntryEx<K, V> entryEx(GridCacheContext<K, V> cacheCtx, IgniteTxKey<K> key) {
+         return cacheCtx.cache().entryEx(key.key());
+     }
+ 
+     /**
+      * Gets cache entry for given key and topology version.
+      *
+      * @param key Key.
+      * @param topVer Topology version.
+      * @return Cache entry.
+      */
+     protected GridCacheEntryEx<K, V> entryEx(GridCacheContext<K, V> cacheCtx, IgniteTxKey<K> key, long topVer) {
+         return cacheCtx.cache().entryEx(key.key(), topVer);
+     }
+ 
+     /**
+      * Performs batch database operations. This commit must be called
+      * before {@link #userCommit()}. This way if there is a DB failure,
+      * cache transaction can still be rolled back.
+      *
+      * @param writeEntries Transaction write set.
+      * @throws IgniteCheckedException If batch update failed.
+      */
+     @SuppressWarnings({"CatchGenericClass"})
+     protected void batchStoreCommit(Iterable<IgniteTxEntry<K, V>> writeEntries) throws IgniteCheckedException {
+         GridCacheStoreManager<K, V> store = store();
+ 
+         if (store != null && storeEnabled() && (!internal() || groupLock()) && (near() || store.writeToStoreFromDht())) {
+             try {
+                 if (writeEntries != null) {
+                     Map<K, IgniteBiTuple<V, GridCacheVersion>> putMap = null;
+                     List<K> rmvCol = null;
+ 
+                     boolean skipNear = near() && store.writeToStoreFromDht();
+ 
+                     for (IgniteTxEntry<K, V> e : writeEntries) {
+                         if (skipNear && e.cached().isNear())
+                             continue;
+ 
+                         boolean intercept = e.context().config().getInterceptor() != null;
+ 
+                         if (intercept || !F.isEmpty(e.transformClosures()))
+                             e.cached().unswap(true, false);
+ 
+                         GridTuple3<GridCacheOperation, V, byte[]> res = applyTransformClosures(e, false);
+ 
+                         GridCacheContext<K, V> cacheCtx = e.context();
+ 
+                         GridCacheOperation op = res.get1();
+                         K key = e.key();
+                         V val = res.get2();
+                         GridCacheVersion ver = writeVersion();
+ 
+                         if (op == CREATE || op == UPDATE) {
+                             // Batch-process all removes if needed.
+                             if (rmvCol != null && !rmvCol.isEmpty()) {
+                                 store.removeAllFromStore(this, rmvCol);
+ 
+                                 // Reset.
+                                 rmvCol.clear();
+                             }
+ 
+                             if (intercept) {
+                                 V old = e.cached().rawGetOrUnmarshal(true);
+ 
+                                 val = (V)cacheCtx.config().getInterceptor().onBeforePut(key, old, val);
+ 
+                                 if (val == null)
+                                     continue;
+ 
+                                 val = cacheCtx.unwrapTemporary(val);
+                             }
+ 
+                             if (putMap == null)
+                                 putMap = new LinkedHashMap<>(writeMap().size(), 1.0f);
+ 
+                             putMap.put(key, F.t(val, ver));
+                         }
+                         else if (op == DELETE) {
+                             // Batch-process all puts if needed.
+                             if (putMap != null && !putMap.isEmpty()) {
+                                 store.putAllToStore(this, putMap);
+ 
+                                 // Reset.
+                                 putMap.clear();
+                             }
+ 
+                             if (intercept) {
+                                 V old = e.cached().rawGetOrUnmarshal(true);
+ 
+                                 IgniteBiTuple<Boolean, V> t = cacheCtx.config().<K, V>getInterceptor()
+                                     .onBeforeRemove(key, old);
+ 
+                                 if (cacheCtx.cancelRemove(t))
+                                     continue;
+                             }
+ 
+                             if (rmvCol == null)
+                                 rmvCol = new LinkedList<>();
+ 
+                             rmvCol.add(key);
+                         }
+                         else if (log.isDebugEnabled())
+                             log.debug("Ignoring NOOP entry for batch store commit: " + e);
+                     }
+ 
+                     if (putMap != null && !putMap.isEmpty()) {
+                         assert rmvCol == null || rmvCol.isEmpty();
+ 
+                         // Batch put at the end of transaction.
+                         store.putAllToStore(this, putMap);
+                     }
+ 
+                     if (rmvCol != null && !rmvCol.isEmpty()) {
+                         assert putMap == null || putMap.isEmpty();
+ 
+                         // Batch remove at the end of transaction.
+                         store.removeAllFromStore(this, rmvCol);
+                     }
+                 }
+ 
+                 // Commit while locks are held.
+                 store.txEnd(this, true);
+             }
+             catch (IgniteCheckedException ex) {
+                 commitError(ex);
+ 
+                 setRollbackOnly();
+ 
+                 // Safe to remove transaction from committed tx list because nothing was committed yet.
+                 cctx.tm().removeCommittedTx(this);
+ 
+                 throw ex;
+             }
+             catch (Throwable ex) {
+                 commitError(ex);
+ 
+                 setRollbackOnly();
+ 
+                 // Safe to remove transaction from committed tx list because nothing was committed yet.
+                 cctx.tm().removeCommittedTx(this);
+ 
+                 throw new IgniteCheckedException("Failed to commit transaction to database: " + this, ex);
+             }
+         }
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings({"CatchGenericClass"})
+     @Override public void userCommit() throws IgniteCheckedException {
+         IgniteTxState state = state();
+ 
+         if (state != COMMITTING) {
+             if (timedOut())
+                 throw new IgniteTxTimeoutException("Transaction timed out: " + this);
+ 
+             setRollbackOnly();
+ 
+             throw new IgniteCheckedException("Invalid transaction state for commit [state=" + state + ", tx=" + this + ']');
+         }
+ 
+         checkValid();
+ 
+         boolean empty = F.isEmpty(near() ? txMap : writeMap());
+ 
+         // Register this transaction as completed prior to write-phase to
+         // ensure proper lock ordering for removed entries.
+         // We add colocated transaction to committed set even if it is empty to correctly order
+         // locks on backup nodes.
+         if (!empty || colocated())
+             cctx.tm().addCommittedTx(this);
+ 
+         if (groupLock())
+             addGroupTxMapping(writeSet());
+ 
+         if (!empty) {
+             // We are holding transaction-level locks for entries here, so we can get next write version.
+             writeVersion(cctx.versions().next(topologyVersion()));
+ 
+             batchStoreCommit(writeMap().values());
+ 
+             try {
+                 cctx.tm().txContext(this);
+ 
+                 long topVer = topologyVersion();
+ 
+                 /*
+                  * Commit to cache. Note that for 'near' transaction we loop through all the entries.
+                  */
+                 for (IgniteTxEntry<K, V> txEntry : (near() ? allEntries() : writeEntries())) {
+                     GridCacheContext<K, V> cacheCtx = txEntry.context();
+ 
+                     GridDrType drType = cacheCtx.isDrEnabled() ? DR_PRIMARY : DR_NONE;
+ 
+                     UUID nodeId = txEntry.nodeId() == null ? this.nodeId : txEntry.nodeId();
+ 
+                     try {
+                         while (true) {
+                             try {
+                                 GridCacheEntryEx<K, V> cached = txEntry.cached();
+ 
+                                 // Must try to evict near entries before committing from
+                                 // transaction manager to make sure locks are held.
+                                 if (!evictNearEntry(txEntry, false)) {
+                                     if (cacheCtx.isNear() && cacheCtx.dr().receiveEnabled()) {
+                                         cached.markObsolete(xidVer);
+ 
+                                         break;
+                                     }
+ 
+                                     if (cached.detached())
+                                         break;
+ 
+                                     GridCacheEntryEx<K, V> nearCached = null;
+ 
+                                     boolean metrics = true;
+ 
+                                     if (updateNearCache(cacheCtx, txEntry.key(), topVer))
+                                         nearCached = cacheCtx.dht().near().peekEx(txEntry.key());
+                                     else if (cacheCtx.isNear() && txEntry.locallyMapped())
+                                         metrics = false;
+ 
+                                     boolean evt = !isNearLocallyMapped(txEntry, false);
+ 
+                                     // For near local transactions we must record DHT version
+                                     // in order to keep near entries on backup nodes until
+                                     // backup remote transaction completes.
+                                     if (cacheCtx.isNear())
+                                         ((GridNearCacheEntry<K, V>)cached).recordDhtVersion(txEntry.dhtVersion());
+ 
+                                     if (!F.isEmpty(txEntry.transformClosures()) || !F.isEmpty(txEntry.filters()))
+                                         txEntry.cached().unswap(true, false);
+ 
+                                     GridTuple3<GridCacheOperation, V, byte[]> res = applyTransformClosures(txEntry,
+                                         true);
+ 
+                                     GridCacheOperation op = res.get1();
+                                     V val = res.get2();
+                                     byte[] valBytes = res.get3();
+ 
 -                                    // Preserve TTL if needed.
 -                                    if (txEntry.ttl() < 0)
 -                                        txEntry.ttl(cached.ttl());
 -
+                                     // Deal with DR conflicts.
+                                     GridCacheVersion explicitVer = txEntry.drVersion() != null ?
+                                         txEntry.drVersion() : writeVersion();
+ 
++                                    if (op == CREATE || op == UPDATE && txEntry.drExpireTime() == -1L) {
++                                        ExpiryPolicy expiry = txEntry.expiry();
++
++                                        if (expiry == null)
++                                            expiry = cacheCtx.expiry();
++
++                                        if (expiry != null) {
++                                            Duration duration = cached.hasValue() ?
++                                                expiry.getExpiryForUpdate() : expiry.getExpiryForCreation();
++
++                                            txEntry.ttl(GridCacheUtils.toTtl(duration));
++                                        }
++                                    }
++
+                                     GridDrResolveResult<V> drRes = cacheCtx.dr().resolveTx(cached,
+                                         txEntry,
+                                         explicitVer,
+                                         op,
+                                         val,
+                                         valBytes,
+                                         txEntry.ttl(),
+                                         txEntry.drExpireTime());
+ 
+                                     if (drRes != null) {
+                                         op = drRes.operation();
+                                         val = drRes.value();
+                                         valBytes = drRes.valueBytes();
+ 
+                                         if (drRes.isMerge())
+                                             explicitVer = writeVersion();
++                                        else if (op == NOOP)
++                                            txEntry.ttl(-1L);
+                                     }
+                                     else
+                                         // Nullify explicit version so that innerSet/innerRemove will work as usual.
+                                         explicitVer = null;
+ 
+                                     if (sndTransformedVals || (drRes != null)) {
+                                         assert sndTransformedVals && cacheCtx.isReplicated() || (drRes != null);
+ 
+                                         txEntry.value(val, true, false);
+                                         txEntry.valueBytes(valBytes);
+                                         txEntry.op(op);
+                                         txEntry.transformClosures(null);
+                                         txEntry.drVersion(explicitVer);
+                                     }
+ 
+                                     if (op == CREATE || op == UPDATE) {
+                                         GridCacheUpdateTxResult<V> updRes = cached.innerSet(
+                                             this,
+                                             eventNodeId(),
+                                             txEntry.nodeId(),
+                                             val,
+                                             valBytes,
+                                             false,
+                                             false,
+                                             txEntry.ttl(),
+                                             evt,
+                                             metrics,
+                                             topVer,
+                                             txEntry.filters(),
+                                             cached.detached() ? DR_NONE : drType,
+                                             txEntry.drExpireTime(),
+                                             cached.isNear() ? null : explicitVer,
+                                             CU.subjectId(this, cctx),
+                                             resolveTaskName());
+ 
+                                         if (nearCached != null && updRes.success())
+                                             nearCached.innerSet(
+                                                 null,
+                                                 eventNodeId(),
+                                                 nodeId,
+                                                 val,
+                                                 valBytes,
+                                                 false,
+                                                 false,
+                                                 txEntry.ttl(),
+                                                 false,
+                                                 metrics,
+                                                 topVer,
+                                                 CU.<K, V>empty(),
+                                                 DR_NONE,
+                                                 txEntry.drExpireTime(),
+                                                 null,
+                                                 CU.subjectId(this, cctx),
+                                                 resolveTaskName());
+                                     }
+                                     else if (op == DELETE) {
+                                         GridCacheUpdateTxResult<V> updRes = cached.innerRemove(
+                                             this,
+                                             eventNodeId(),
+                                             txEntry.nodeId(),
+                                             false,
+                                             false,
+                                             evt,
+                                             metrics,
+                                             topVer,
+                                             txEntry.filters(),
+                                             cached.detached()  ? DR_NONE : drType,
+                                             cached.isNear() ? null : explicitVer,
+                                             CU.subjectId(this, cctx),
+                                             resolveTaskName());
+ 
+                                         if (nearCached != null && updRes.success())
+                                             nearCached.innerRemove(
+                                                 null,
+                                                 eventNodeId(),
+                                                 nodeId,
+                                                 false,
+                                                 false,
+                                                 false,
+                                                 metrics,
+                                                 topVer,
+                                                 CU.<K, V>empty(),
+                                                 DR_NONE,
+                                                 null,
+                                                 CU.subjectId(this, cctx),
+                                                 resolveTaskName());
+                                     }
+                                     else if (op == RELOAD) {
+                                         cached.innerReload(CU.<K, V>empty());
+ 
+                                         if (nearCached != null)
+                                             nearCached.innerReload(CU.<K, V>empty());
+                                     }
+                                     else if (op == READ) {
++                                        Duration duration = expiryForAccess(txEntry);
++
++                                        if (duration != null)
++                                            cached.updateTtl(null, GridCacheUtils.toTtl(duration));
++
+                                         if (log.isDebugEnabled())
+                                             log.debug("Ignoring READ entry when committing: " + txEntry);
+                                     }
+                                     else {
+                                         assert !groupLock() || txEntry.groupLockEntry() || ownsLock(txEntry.cached()):
+                                             "Transaction does not own lock for group lock entry during  commit [tx=" +
+                                                 this + ", txEntry=" + txEntry + ']';
+ 
++                                        if (txEntry.ttl() != -1L)
++                                            cached.updateTtl(null, txEntry.ttl());
++
+                                         if (log.isDebugEnabled())
+                                             log.debug("Ignoring NOOP entry when committing: " + txEntry);
+                                     }
+                                 }
+ 
+                                 // Check commit locks after set, to make sure that
+                                 // we are not changing obsolete entries.
+                                 // (innerSet and innerRemove will throw an exception
+                                 // if an entry is obsolete).
+                                 if (txEntry.op() != READ && !txEntry.groupLockEntry())
+                                     checkCommitLocks(cached);
+ 
+                                 // Break out of while loop.
+                                 break;
+                             }
+                             // If entry cached within transaction got removed.
+                             catch (GridCacheEntryRemovedException ignored) {
+                                 if (log.isDebugEnabled())
+                                     log.debug("Got removed entry during transaction commit (will retry): " + txEntry);
+ 
+                                 txEntry.cached(entryEx(cacheCtx, txEntry.txKey()), txEntry.keyBytes());
+                             }
+                         }
+                     }
+                     catch (Throwable ex) {
+                         // We are about to initiate transaction rollback when tx has started to committing.
+                         // Need to remove version from committed list.
+                         cctx.tm().removeCommittedTx(this);
+ 
+                         if (X.hasCause(ex, GridCacheIndexUpdateException.class) && cacheCtx.cache().isMongoDataCache()) {
+                             if (log.isDebugEnabled())
+                                 log.debug("Failed to update mongo document index (transaction entry will " +
+                                     "be ignored): " + txEntry);
+ 
+                             // Set operation to NOOP.
+                             txEntry.op(NOOP);
+ 
+                             setRollbackOnly();
+ 
+                             throw ex;
+                         }
+                         else {
+                             IgniteCheckedException err = new IgniteTxHeuristicException("Failed to locally write to cache " +
+                                 "(all transaction entries will be invalidated, however there was a window when " +
+                                 "entries for this transaction were visible to others): " + this, ex);
+ 
+                             U.error(log, "Heuristic transaction failure.", err);
+ 
+                             commitErr.compareAndSet(null, err);
+ 
+                             state(UNKNOWN);
+ 
+                             try {
+                                 // Courtesy to minimize damage.
+                                 uncommit();
+                             }
+                             catch (Throwable ex1) {
+                                 U.error(log, "Failed to uncommit transaction: " + this, ex1);
+                             }
+ 
+                             throw err;
+                         }
+                     }
+                 }
++
++                if (!near()) {
++                    for (GridCacheTxEntry<K, V> txEntry : readEntries()) {
++                        Duration duration = expiryForAccess(txEntry);
++
++                        if (duration != null)
++                            txEntry.cached().updateTtl(null, GridCacheUtils.toTtl(duration));
++                    }
++                }
+             }
+             finally {
+                 cctx.tm().txContextReset();
+             }
+         }
+         else {
++            for (GridCacheTxEntry<K, V> txEntry : readEntries()) {
++                Duration duration = expiryForAccess(txEntry);
++
++                if (duration != null)
++                    txEntry.cached().updateTtl(null, GridCacheUtils.toTtl(duration));
++            }
++
+             GridCacheStoreManager<K, V> store = store();
+ 
+             if (store != null && (!internal() || groupLock())) {
+                 try {
+                     store.txEnd(this, true);
+                 }
+                 catch (IgniteCheckedException e) {
+                     commitError(e);
+ 
+                     setRollbackOnly();
+ 
+                     cctx.tm().removeCommittedTx(this);
+ 
+                     throw e;
+                 }
+             }
+         }
+ 
+         // Do not unlock transaction entries if one-phase commit.
+         if (!onePhaseCommit()) {
+             if (doneFlag.compareAndSet(false, true)) {
+                 // Unlock all locks.
+                 cctx.tm().commitTx(this);
+ 
+                 boolean needsCompletedVersions = needsCompletedVersions();
+ 
+                 assert !needsCompletedVersions || completedBase != null;
+                 assert !needsCompletedVersions || committedVers != null;
+                 assert !needsCompletedVersions || rolledbackVers != null;
+             }
+         }
+     }
+ 
+     /**
++     * @param txEntry Tx entry.
++     * @return New duration.
++     */
++    @Nullable private Duration expiryForAccess(GridCacheTxEntry<K, V> txEntry) {
++        ExpiryPolicy expiry = txEntry.expiry();
++
++        if (expiry == null)
++            expiry = txEntry.context().expiry();
++
++        return expiry != null ? expiry.getExpiryForAccess() : null;
++    }
++
++    /**
+      * Commits transaction to transaction manager. Used for one-phase commit transactions only.
+      */
+     public void tmCommit() {
+         assert onePhaseCommit();
+ 
+         if (doneFlag.compareAndSet(false, true)) {
+             // Unlock all locks.
+             cctx.tm().commitTx(this);
+ 
+             state(COMMITTED);
+ 
+             boolean needsCompletedVersions = needsCompletedVersions();
+ 
+             assert !needsCompletedVersions || completedBase != null;
+             assert !needsCompletedVersions || committedVers != null;
+             assert !needsCompletedVersions || rolledbackVers != null;
+         }
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void completedVersions(
+         GridCacheVersion completedBase,
+         Collection<GridCacheVersion> committedVers,
+         Collection<GridCacheVersion> rolledbackVers) {
+         this.completedBase = completedBase;
+         this.committedVers = committedVers;
+         this.rolledbackVers = rolledbackVers;
+     }
+ 
+     /**
+      * @return Completed base for ordering.
+      */
+     public GridCacheVersion completedBase() {
+         return completedBase;
+     }
+ 
+     /**
+      * @return Committed versions.
+      */
+     public Collection<GridCacheVersion> committedVersions() {
+         return committedVers;
+     }
+ 
+     /**
+      * @return Rolledback versions.
+      */
+     public Collection<GridCacheVersion> rolledbackVersions() {
+         return rolledbackVers;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void userRollback() throws IgniteCheckedException {
+         IgniteTxState state = state();
+ 
+         if (state != ROLLING_BACK && state != ROLLED_BACK) {
+             setRollbackOnly();
+ 
+             throw new IgniteCheckedException("Invalid transaction state for rollback [state=" + state + ", tx=" + this + ']',
+                 commitErr.get());
+         }
+ 
+         if (doneFlag.compareAndSet(false, true)) {
+             try {
+                 if (near())
+                     // Must evict near entries before rolling back from
+                     // transaction manager, so they will be removed from cache.
+                     for (IgniteTxEntry<K, V> e : allEntries())
+                         evictNearEntry(e, false);
+ 
+                 cctx.tm().rollbackTx(this);
+ 
+                 GridCacheStoreManager<K, V> store = store();
+ 
+                 if (store != null && (near() || store.writeToStoreFromDht())) {
+                     if (!internal() || groupLock())
+                         store.txEnd(this, false);
+                 }
+             }
+             catch (Error | IgniteCheckedException | RuntimeException e) {
+                 U.addLastCause(e, commitErr.get(), log);
+ 
+                 throw e;
+             }
+         }
+     }
+ 
+     /**
+      * Checks if there is a cached or swapped value for
+      * {@link #getAllAsync(GridCacheContext, Collection, GridCacheEntryEx, boolean, IgnitePredicate[])} method.
+      *
+      *
+      * @param keys Key to enlist.
+      * @param cached Cached entry, if called from entry wrapper.
+      * @param map Return map.
+      * @param missed Map of missed keys.
+      * @param keysCnt Keys count (to avoid call to {@code Collection.size()}).
+      * @param deserializePortable Deserialize portable flag.
+      * @param filter Filter to test.
+      * @throws IgniteCheckedException If failed.
+      * @return Enlisted keys.
+      */
+     @SuppressWarnings({"RedundantTypeArguments"})
+     private Collection<K> enlistRead(
+         final GridCacheContext<K, V> cacheCtx,
+         Collection<? extends K> keys,
+         @Nullable GridCacheEntryEx<K, V> cached,
+         Map<K, V> map,
+         Map<K, GridCacheVersion> missed,
+         int keysCnt,
+         boolean deserializePortable,
+         IgnitePredicate<GridCacheEntry<K, V>>[] filter) throws IgniteCheckedException {
+         assert !F.isEmpty(keys);
+         assert keysCnt == keys.size();
+         assert cached == null || F.first(keys).equals(cached.key());
+ 
+         cacheCtx.checkSecurity(GridSecurityPermission.CACHE_READ);
+ 
+         groupLockSanityCheck(cacheCtx, keys);
+ 
+         boolean single = keysCnt == 1;
+ 
+         Collection<K> lockKeys = null;
+ 
+         long topVer = topologyVersion();
+ 
+         // In this loop we cover only read-committed or optimistic transactions.
+         // Transactions that are pessimistic and not read-committed are covered
+         // outside of this loop.
+         for (K key : keys) {
+             if (key == null)
+                 continue;
+ 
+             if (pessimistic() && !readCommitted())
+                 addActiveCache(cacheCtx);
+ 
+             IgniteTxKey<K> txKey = cacheCtx.txKey(key);
+ 
+             // Check write map (always check writes first).
+             IgniteTxEntry<K, V> txEntry = entry(txKey);
+ 
+             // Either non-read-committed or there was a previous write.
+             if (txEntry != null) {
+                 if (cacheCtx.isAll(txEntry.cached(), filter)) {
+                     V val = txEntry.value();
+ 
+                     // Read value from locked entry in group-lock transaction as well.
+                     if (txEntry.hasValue()) {
+                         if (!F.isEmpty(txEntry.transformClosures())) {
+                             for (IgniteClosure<V, V> clos : txEntry.transformClosures())
+                                 val = clos.apply(val);
+                         }
+ 
+                         if (val != null) {
+                             V val0 = val;
+ 
+                             if (cacheCtx.portableEnabled())
+                                 val0 = (V)cacheCtx.unwrapPortableIfNeeded(val, !deserializePortable);
+ 
+                             map.put(key, val0);
+                         }
+                     }
+                     else {
+                         assert txEntry.op() == TRANSFORM || (groupLock() && !txEntry.groupLockEntry());
+ 
+                         while (true) {
+                             try {
+                                 Object transformClo =
+                                     (txEntry.op() == TRANSFORM  && cctx.gridEvents().isRecordable(EVT_CACHE_OBJECT_READ)) ?
+                                         F.first(txEntry.transformClosures()) : null;
+ 
+                                 val = txEntry.cached().innerGet(this,
+                                     /*swap*/true,
+                                     /*read-through*/false,
+                                     /*fail fast*/true,
+                                     /*unmarshal*/true,
+                                     /*metrics*/true,
+                                     /*event*/true,
+                                     /*temporary*/false,
+                                     CU.subjectId(this, cctx),
+                                     transformClo,
+                                     resolveTaskName(),
 -                                    filter);
++                                    filter,
++                                    null);
+ 
+                                 if (val != null) {
+                                     if (!readCommitted())
+                                         txEntry.readValue(val);
+ 
+                                     if (!F.isEmpty(txEntry.transformClosures())) {
+                                         for (IgniteClosure<V, V> clos : txEntry.transformClosures())
+                                             val = clos.apply(val);
+                                     }
+ 
+                                     V val0 = val;
+ 
+                                     if (cacheCtx.portableEnabled())
+                                         val0 = (V)cacheCtx.unwrapPortableIfNeeded(val, !deserializePortable);
+ 
+                                     map.put(key, val0);
+                                 }
+                                 else
+                                     missed.put(key, txEntry.cached().version());
+ 
+                                 break;
+                             }
+                             catch (GridCacheFilterFailedException e) {
+                                 if (log.isDebugEnabled())
+                                     log.debug("Filter validation failed for entry: " + txEntry);
+ 
+                                 if (!readCommitted())
+                                     txEntry.readValue(e.<V>value());
+                             }
+                             catch (GridCacheEntryRemovedException ignored) {
+                                 txEntry.cached(entryEx(cacheCtx, txEntry.txKey(), topVer), txEntry.keyBytes());
+                             }
+                         }
+                     }
+                 }
+             }
+             // First time access within transaction.
+             else {
+                 if (lockKeys == null)
+                     lockKeys = single ? (Collection<K>)keys : new ArrayList<K>(keysCnt);
+ 
+                 if (!single)
+                     lockKeys.add(key);
+ 
+                 while (true) {
+                     GridCacheEntryEx<K, V> entry;
+ 
+                     if (cached != null) {
+                         entry = cached;
+ 
+                         cached = null;
+                     }
+                     else
+                         entry = entryEx(cacheCtx, txKey, topVer);
+ 
+                     try {
+                         GridCacheVersion ver = entry.version();
+ 
+                         V val = null;
+ 
+                         if (!pessimistic() || readCommitted() || groupLock()) {
+                             // This call will check for filter.
+                             val = entry.innerGet(this,
+                                 /*swap*/true,
+                                 /*no read-through*/false,
+                                 /*fail-fast*/true,
+                                 /*unmarshal*/true,
+                                 /*metrics*/true,
+                                 /*event*/true,
+                                 /*temporary*/false,
+                                 CU.subjectId(this, cctx),
+                                 null,
+                                 resolveTaskName(),
 -                                filter);
++                                filter,
++                                null);
+ 
+                             if (val != null) {
+                                 V val0 = val;
+ 
+                                 if (cacheCtx.portableEnabled())
+                                     val0 = (V)cacheCtx.unwrapPortableIfNeeded(val, !deserializePortable);
+ 
+                                 map.put(key, val0);
+                             }
+                             else
+                                 missed.put(key, ver);
+                         }
+                         else
+                             // We must wait for the lock in pessimistic mode.
+                             missed.put(key, ver);
+ 
+                         if (!readCommitted()) {
 -                            txEntry = addEntry(READ, val, null, entry, -1, filter, true, -1L, -1L, null);
++                            txEntry = addEntry(READ, val, null, entry, null, filter, true, -1L, -1L, null);
+ 
+                             if (groupLock())
+                                 txEntry.groupLockEntry(true);
+ 
+                             // As optimization, mark as checked immediately
+                             // for non-pessimistic if value is not null.
+                             if (val != null && !pessimistic())
+                                 txEntry.markValid();
+                         }
+ 
+                         break; // While.
+                     }
+                     catch (GridCacheEntryRemovedException ignored) {
+                         if (log.isDebugEnabled())
+                             log.debug("Got removed entry in transaction getAllAsync(..) (will retry): " + key);
+                     }
+                     catch (GridCacheFilterFailedException e) {
+                         if (log.isDebugEnabled())
+                             log.debug("Filter validation failed for entry: " + entry);
+ 
+                         if (!readCommitted()) {
+                             // Value for which failure occurred.
+                             V val = e.<V>value();
+ 
 -                            txEntry = addEntry(READ, val, null, entry, -1, CU.<K, V>empty(), false, -1L, -1L, null);
++                            txEntry = addEntry(READ, val, null, entry, null, CU.<K, V>empty(), false, -1L, -1L, null);
+ 
+                             // Mark as checked immediately for non-pessimistic.
+                             if (val != null && !pessimistic())
+                                 txEntry.markValid();
+                         }
+ 
+                         break; // While loop.
+                     }
+                 }
+             }
+         }
+ 
+         return lockKeys != null ? lockKeys : Collections.<K>emptyList();
+     }
+ 
+     /**
+      * Adds skipped key.
+      *
+      * @param skipped Skipped set (possibly {@code null}).
+      * @param key Key to add.
+      * @return Skipped set.
+      */
+     private Set<K> skip(Set<K> skipped, K key) {
+         if (skipped == null)
+             skipped = new GridLeanSet<>();
+ 
+         skipped.add(key);
+ 
+         if (log.isDebugEnabled())
+             log.debug("Added key to skipped set: " + key);
+ 
+         return skipped;
+     }
+ 
+     /**
+      * Loads all missed keys for
+      * {@link #getAllAsync(GridCacheContext, Collection, GridCacheEntryEx, boolean, IgnitePredicate[])} method.
+      *
+      * @param map Return map.
+      * @param missedMap Missed keys.
+      * @param redos Keys to retry.
+      * @param deserializePortable Deserialize portable flag.
+      * @param filter Filter.
+      * @return Loaded key-value pairs.
+      */
+     private IgniteFuture<Map<K, V>> checkMissed(
+         final GridCacheContext<K, V> cacheCtx,
+         final Map<K, V> map,
+         final Map<K, GridCacheVersion> missedMap,
+         @Nullable final Collection<K> redos,
+         final boolean deserializePortable,
+         final IgnitePredicate<GridCacheEntry<K, V>>[] filter
+     ) {
+         assert redos != null || pessimistic();
+ 
+         if (log.isDebugEnabled())
+             log.debug("Loading missed values for missed map: " + missedMap);
+ 
+         final Collection<K> loaded = new HashSet<>();
+ 
+         return new GridEmbeddedFuture<>(cctx.kernalContext(),
+             loadMissing(
+                 cacheCtx,
+                 false, missedMap.keySet(), deserializePortable, new CI2<K, V>() {
+                 /** */
+                 private GridCacheVersion nextVer;
+ 
+                 @Override public void apply(K key, V val) {
+                     if (isRollbackOnly()) {
+                         if (log.isDebugEnabled())
+                             log.debug("Ignoring loaded value for read because transaction was rolled back: " +
+                                 IgniteTxLocalAdapter.this);
+ 
+                         return;
+                     }
+ 
+                     GridCacheVersion ver = missedMap.get(key);
+ 
+                     if (ver == null) {
+                         if (log.isDebugEnabled())
+                             log.debug("Value from storage was never asked for [key=" + key + ", val=" + val + ']');
+ 
+                         return;
+                     }
+ 
+                     V visibleVal = val;
+ 
+                     IgniteTxKey<K> txKey = cacheCtx.txKey(key);
+ 
+                     IgniteTxEntry<K, V> txEntry = entry(txKey);
+ 
+                     if (txEntry != null) {
+                         if (!readCommitted())
+                             txEntry.readValue(val);
+ 
+                         if (!F.isEmpty(txEntry.transformClosures())) {
+                             for (IgniteClosure<V, V> clos : txEntry.transformClosures())
+                                 visibleVal = clos.apply(visibleVal);
+                         }
+                     }
+ 
+                     // In pessimistic mode we hold the lock, so filter validation
+                     // should always be valid.
+                     if (pessimistic())
+                         ver = null;
+ 
+                     // Initialize next version.
+                     if (nextVer == null)
+                         nextVer = cctx.versions().next(topologyVersion());
+ 
+                     while (true) {
+                         assert txEntry != null || readCommitted() || groupLock();
+ 
+                         GridCacheEntryEx<K, V> e = txEntry == null ? entryEx(cacheCtx, txKey) : txEntry.cached();
+ 
+                         try {
+                             boolean pass = cacheCtx.isAll(e, filter);
+ 
+                             // Must initialize to true since even if filter didn't pass,
+                             // we still record the transaction value.
+                             boolean set = true;
+ 
+                             if (pass) {
+                                 try {
+                                     set = e.versionedValue(val, ver, nextVer);
+                                 }
+                                 catch (GridCacheEntryRemovedException ignore) {
+                                     if (log.isDebugEnabled())
+                                         log.debug("Got removed entry in transaction getAll method " +
+                                             "(will try again): " + e);
+ 
+                                     if (pessimistic() && !readCommitted() && !isRollbackOnly() &&
+                                         (!groupLock() || F.eq(e.key(), groupLockKey()))) {
+                                         U.error(log, "Inconsistent transaction state (entry got removed while " +
+                                             "holding lock) [entry=" + e + ", tx=" + IgniteTxLocalAdapter.this + "]");
+ 
+                                         setRollbackOnly();
+ 
+                                         return;
+                                     }
+ 
+                                     if (txEntry != null)
+                                         txEntry.cached(entryEx(cacheCtx, txKey), txEntry.keyBytes());
+ 
+                                     continue; // While loop.
+                                 }
+                             }
+ 
+                             // In pessimistic mode, we should always be able to set.
+                             assert set || !pessimistic();
+ 
+                             if (readCommitted() || groupLock()) {
+                                 cacheCtx.evicts().touch(e, topologyVersion());
+ 
+                                 if (pass && visibleVal != null)
+                                     map.put(key, visibleVal);
+                             }
+                             else {
+                                 assert txEntry != null;
+ 
+                                 if (set || F.isEmptyOrNulls(filter)) {
+                                     txEntry.setAndMarkValid(val);
+ 
+                                     if (pass && visibleVal != null)
+                                         map.put(key, visibleVal);
+                                 }
+                                 else {
+                                     assert !pessimistic() : "Pessimistic transaction should not have to redo gets: " +
+                                         this;
+ 
+                                     if (log.isDebugEnabled())
+                                         log.debug("Failed to set versioned value for entry (will redo): " + e);
+ 
+                                     redos.add(key);
+                                 }
+                             }
+ 
+                             loaded.add(key);
+ 
+                             if (log.isDebugEnabled())
+                                 log.debug("Set value loaded from store into entry from transaction [set=" + set +
+                                     ", matchVer=" + ver + ", newVer=" + nextVer + ", entry=" + e + ']');
+ 
+                             break; // While loop.
+                         }
+                         catch (IgniteCheckedException ex) {
+                             throw new IgniteException("Failed to put value for cache entry: " + e, ex);
+                         }
+                     }
+                 }
+             }),
+             new C2<Boolean, Exception, Map<K, V>>() {
+                 @Override public Map<K, V> apply(Boolean b, Exception e) {
+                     if (e != null) {
+                         setRollbackOnly();
+ 
+                         throw new GridClosureException(e);
+                     }
+ 
+                     if (!b && !readCommitted()) {
+                         // There is no store - we must mark the entries.
+                         for (K key : missedMap.keySet()) {
+                             IgniteTxEntry<K, V> txEntry = entry(cacheCtx.txKey(key));
+ 
+                             if (txEntry != null)
+                                 txEntry.markValid();
+                         }
+                     }
+ 
+                     if (readCommitted()) {
+                         Collection<K> notFound = new HashSet<>(missedMap.keySet());
+ 
+                         notFound.removeAll(loaded);
+ 
+                         // In read-committed mode touch entries that have just been read.
+                         for (K key : notFound) {
+                             IgniteTxEntry<K, V> txEntry = entry(cacheCtx.txKey(key));
+ 
+                             GridCacheEntryEx<K, V> entry = txEntry == null ? cacheCtx.cache().peekEx(key) :
+                                 txEntry.cached();
+ 
+                             if (entry != null)
+                                 cacheCtx.evicts().touch(entry, topologyVersion());
+                         }
+                     }
+ 
+                     return map;
+                 }
+             });
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<Map<K, V>> getAllAsync(
+         final GridCacheContext<K, V> cacheCtx,
+         Collection<? extends K> keys,
+         @Nullable GridCacheEntryEx<K, V> cached, final boolean deserializePortable,
+         final IgnitePredicate<GridCacheEntry<K, V>>[] filter) {
+         if (F.isEmpty(keys))
+             return new GridFinishedFuture<>(cctx.kernalContext(), Collections.<K, V>emptyMap());
+ 
+         init();
+ 
+         int keysCnt = keys.size();
+ 
+         boolean single = keysCnt == 1;
+ 
+         try {
+             checkValid();
+ 
+             final Map<K, V> retMap = new GridLeanMap<>(keysCnt);
+ 
+             final Map<K, GridCacheVersion> missed = new GridLeanMap<>(pessimistic() ? keysCnt : 0);
+ 
+             final Collection<K> lockKeys = enlistRead(cacheCtx, keys, cached, retMap, missed, keysCnt,
+                 deserializePortable, filter);
+ 
+             if (single && missed.isEmpty())
+                 return new GridFinishedFuture<>(cctx.kernalContext(), retMap);
+ 
+             // Handle locks.
+             if (pessimistic() && !readCommitted() && !groupLock()) {
+                 IgniteFuture<Boolean> fut = cacheCtx.cache().txLockAsync(lockKeys, lockTimeout(), this, true, true,
+                     isolation, isInvalidate(), CU.<K, V>empty());
+ 
+                 PLC2<Map<K, V>> plc2 = new PLC2<Map<K, V>>() {
+                     @Override public IgniteFuture<Map<K, V>> postLock() throws IgniteCheckedException {
+                         if (log.isDebugEnabled())
+                             log.debug("Acquired transaction lock for read on keys: " + lockKeys);
+ 
+                         // Load keys only after the locks have been acquired.
+                         for (K key : lockKeys) {
+                             if (retMap.containsKey(key))
+                                 // We already have a return value.
+                                 continue;
+ 
+                             IgniteTxKey<K> txKey = cacheCtx.txKey(key);
+ 
+                             IgniteTxEntry<K, V> txEntry = entry(txKey);
+ 
+                             assert txEntry != null;
+ 
+                             // Check if there is cached value.
+                             while (true) {
+                                 GridCacheEntryEx<K, V> cached = txEntry.cached();
+ 
+                                 try {
+                                     Object transformClo =
+                                         (!F.isEmpty(txEntry.transformClosures()) &&
+                                             cctx.gridEvents().isRecordable(EVT_CACHE_OBJECT_READ)) ?
+                                             F.first(txEntry.transformClosures()) : null;
+ 
+                                     V val = cached.innerGet(IgniteTxLocalAdapter.this,
+                                         cacheCtx.isSwapOrOffheapEnabled(),
+                                         /*read-through*/false,
+                                         /*fail-fast*/true,
+                                         /*unmarshal*/true,
+                                         /*metrics*/true,
+                                         /*events*/true,
+                                         /*temporary*/true,
+                                         CU.subjectId(IgniteTxLocalAdapter.this, cctx),
+                                         transformClo,
+                                         resolveTaskName(),
 -                                        filter);
++                                        filter,
++                                        null);
+ 
+                                     // If value is in cache and passed the filter.
+                                     if (val != null) {
+                                         missed.remove(key);
+ 
+                                         txEntry.setAndMarkValid(val);
+ 
+                                         if (!F.isEmpty(txEntry.transformClosures())) {
+                                             for (IgniteClosure<V, V> clos : txEntry.transformClosures())
+                                                 val = clos.apply(val);
+                                         }
+ 
+                                         if (cacheCtx.portableEnabled())
+                                             val = (V)cacheCtx.unwrapPortableIfNeeded(val, !deserializePortable);
+ 
+                                         retMap.put(key, val);
+                                     }
+ 
+                                     // Even though we bring the value back from lock acquisition,
+                                     // we still need to recheck primary node for consistent values
+                                     // in case of concurrent transactional locks.
+ 
+                                     break; // While.
+                                 }
+                                 catch (GridCacheEntryRemovedException ignore) {
+                                     if (log.isDebugEnabled())
+                                         log.debug("Got removed exception in get postLock (will retry): " +
+                                             cached);
+ 
+                                     txEntry.cached(entryEx(cacheCtx, txKey), txEntry.keyBytes());
+                                 }
+                                 catch (GridCacheFilterFailedException e) {
+                                     // Failed value for the filter.
+                                     V val = e.value();
+ 
+                                     if (val != null) {
+                                         // If filter fails after lock is acquired, we don't reload,
+                                         // regardless if value is null or not.
+                                         missed.remove(key);
+ 
+                                         txEntry.setAndMarkValid(val);
+                                     }
+ 
+                                     break; // While.
+                                 }
+                             }
+                         }
+ 
+                         if (!missed.isEmpty() && (cacheCtx.isReplicated() || cacheCtx.isLocal()))
+                             return checkMissed(cacheCtx, retMap, missed, null, deserializePortable, filter);
+ 
+                         return new GridFinishedFuture<>(cctx.kernalContext(), Collections.<K, V>emptyMap());
+                     }
+                 };
+ 
+                 FinishClosure<Map<K, V>> finClos = new FinishClosure<Map<K, V>>() {
+                     @Override Map<K, V> finish(Map<K, V> loaded) {
+                         retMap.putAll(loaded);
+ 
+                         return retMap;
+                     }
+                 };
+ 
+                 if (fut.isDone()) {
+                     try {
+                         IgniteFuture<Map<K, V>> fut1 = plc2.apply(fut.get(), null);
+ 
+                         return fut1.isDone() ?
+                             new GridFinishedFutureEx<>(finClos.apply(fut1.get(), null)) :
+                             new GridEmbeddedFuture<>(cctx.kernalContext(), fut1, finClos);
+                     }
+                     catch (GridClosureException e) {
+                         return new GridFinishedFuture<>(cctx.kernalContext(), e.unwrap());
+                     }
+                     catch (IgniteCheckedException e) {
+                         try {
+                             return plc2.apply(false, e);
+                         }
+                         catch (Exception e1) {
+                             return new GridFinishedFuture<>(cctx.kernalContext(), e1);
+                         }
+                     }
+                 }
+                 else {
+                     return new GridEmbeddedFuture<>(
+                         cctx.kernalContext(),
+                         fut,
+                         plc2,
+                         finClos);
+                 }
+             }
+             else {
+                 assert optimistic() || readCommitted() || groupLock();
+ 
+                 final Collection<K> redos = new LinkedList<>();
+ 
+                 if (!missed.isEmpty()) {
+                     if (!readCommitted())
+                         for (Iterator<K> it = missed.keySet().iterator(); it.hasNext(); )
+                             if (retMap.containsKey(it.next()))
+                                 it.remove();
+ 
+                     if (missed.isEmpty())
+                         return new GridFinishedFuture<>(cctx.kernalContext(), retMap);
+ 
+                     return new GridEmbeddedFuture<>(
+                         cctx.kernalContext(),
+                         // First future.
+                         checkMissed(cacheCtx, retMap, missed, redos, deserializePortable, filter),
+                         // Closure that returns another future, based on result from first.
+                         new PMC<Map<K, V>>() {
+                             @Override public IgniteFuture<Map<K, V>> postMiss(Map<K, V> map) {
+                                 if (redos.isEmpty())
+                                     return new GridFinishedFuture<>(cctx.kernalContext(),
+                                         Collections.<K, V>emptyMap());
+ 
+                                 if (log.isDebugEnabled())
+                                     log.debug("Starting to future-recursively get values for keys: " + redos);
+ 
+                                 // Future recursion.
+                                 return getAllAsync(cacheCtx, redos, null, deserializePortable, filter);
+                             }
+                         },
+                         // Finalize.
+                         new FinishClosure<Map<K, V>>() {
+                             @Override Map<K, V> finish(Map<K, V> loaded) {
+                                 for (Map.Entry<K, V> entry : loaded.entrySet()) {
+                                     IgniteTxEntry<K, V> txEntry = entry(cacheCtx.txKey(entry.getKey()));
+ 
+                                     V val = entry.getValue();
+ 
+                                     if (!readCommitted())
+                                         txEntry.readValue(val);
+ 
+                                     if (!F.isEmpty(txEntry.transformClosures())) {
+                                         for (IgniteClosure<V, V> clos : txEntry.transformClosures())
+                                             val = clos.apply(val);
+                                     }
+ 
+                                     retMap.put(entry.getKey(), val);
+                                 }
+ 
+                                 return retMap;
+                             }
+                         }
+                     );
+                 }
+ 
+                 return new GridFinishedFuture<>(cctx.kernalContext(), retMap);
+             }
+         }
+         catch (IgniteCheckedException e) {
+             setRollbackOnly();
+ 
+             return new GridFinishedFuture<>(cctx.kernalContext(), e);
+         }
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<GridCacheReturn<V>> putAllAsync(
+         GridCacheContext<K, V> cacheCtx,
+         Map<? extends K, ? extends V> map,
+         boolean retval,
+         @Nullable GridCacheEntryEx<K, V> cached,
+         long ttl,
+         IgnitePredicate<GridCacheEntry<K, V>>[] filter
+     ) {
+         return putAllAsync0(cacheCtx, map, null, null, retval, cached, ttl, filter);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<?> putAllDrAsync(
+         GridCacheContext<K, V> cacheCtx,
+         Map<? extends K, GridCacheDrInfo<V>> drMap
+     ) {
+         return putAllAsync0(cacheCtx, null, null, drMap, false, null, -1, null);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<GridCacheReturn<V>> transformAllAsync(
+         GridCacheContext<K, V> cacheCtx,
+         @Nullable Map<? extends K, ? extends IgniteClosure<V, V>> map,
+         boolean retval,
+         @Nullable GridCacheEntryEx<K, V> cached,
+         long ttl
+     ) {
+         return putAllAsync0(cacheCtx, null, map, null, retval, null, -1, null);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<?> removeAllDrAsync(
+         GridCacheContext<K, V> cacheCtx,
+         Map<? extends K, GridCacheVersion> drMap
+     ) {
+         return removeAllAsync0(cacheCtx, null, drMap, null, false, null);
+     }
+ 
+     /**
+      * Checks filter for non-pessimistic transactions.
+      *
+      * @param cached Cached entry.
+      * @param filter Filter to check.
+      * @return {@code True} if passed or pessimistic.
+      * @throws IgniteCheckedException If failed.
+      */
+     private boolean filter(GridCacheEntryEx<K, V> cached,
+         IgnitePredicate<GridCacheEntry<K, V>>[] filter) throws IgniteCheckedException {
+         return pessimistic() || cached.context().isAll(cached, filter);
+     }
+ 
+     /**
+      * Internal routine for <tt>putAll(..)</tt>
+      *
++     * @param cacheCtx Cache context.
+      * @param keys Keys to enlist.
+      * @param cached Cached entry.
 -     * @param ttl Time to live for entry. If negative, leave unchanged.
++     * @param expiryPlc Explicitly specified expiry policy for entry.
+      * @param implicit Implicit flag.
+      * @param lookup Value lookup map ({@code null} for remove).
+      * @param transformMap Map with transform closures if this is a transform operation.
+      * @param retval Flag indicating whether a value should be returned.
+      * @param lockOnly If {@code true}, then entry will be enlisted as noop.
+      * @param filter User filters.
+      * @param ret Return value.
+      * @param enlisted Collection of keys enlisted into this transaction.
+      * @param drPutMap DR put map (optional).
+      * @param drRmvMap DR remove map (optional).
+      * @return Future with skipped keys (the ones that didn't pass filter for pessimistic transactions).
+      */
+     protected IgniteFuture<Set<K>> enlistWrite(
+         GridCacheContext<K, V> cacheCtx,
+         Collection<? extends K> keys,
+         @Nullable GridCacheEntryEx<K, V> cached,
 -        long ttl,
++        @Nullable ExpiryPolicy expiryPlc,
+         boolean implicit,
+         @Nullable Map<? extends K, ? extends V> lookup,
+         @Nullable Map<? extends K, ? extends IgniteClosure<V, V>> transformMap,
+         boolean retval,
+         boolean lockOnly,
+         IgnitePredicate<GridCacheEntry<K, V>>[] filter,
+         final GridCacheReturn<V> ret,
+         Collection<K> enlisted,
+         @Nullable Map<? extends K, GridCacheDrInfo<V>> drPutMap,
+         @Nullable Map<? extends K, GridCacheVersion> drRmvMap
+     ) {
+         assert cached == null || keys.size() == 1;
+         assert cached == null || F.first(keys).equals(cached.key());
+ 
+         try {
+             addActiveCache(cacheCtx);
+         }
+         catch (IgniteCheckedException e) {
+             return new GridFinishedFuture<>(cctx.kernalContext(), e);
+         }
+ 
+         Set<K> skipped = null;
+ 
+         boolean rmv = lookup == null && transformMap == null;
+ 
+         try {
+             // Set transform flag for transaction.
+             if (transformMap != null)
+                 transform = true;
+ 
+             groupLockSanityCheck(cacheCtx, keys);
+ 
+             for (K key : keys) {
+                 V val = rmv || lookup == null ? null : lookup.get(key);
+                 IgniteClosure<V, V> transformClo = transformMap == null ? null : transformMap.get(key);
+ 
+                 GridCacheVersion drVer;
+                 long drTtl;
+                 long drExpireTime;
+ 
+                 if (drPutMap != null) {
+                     GridCacheDrInfo<V> info = drPutMap.get(key);
+ 
+                     assert info != null;
+ 
+                     drVer = info.version();
+                     drTtl = info.ttl();
+                     drExpireTime = info.expireTime();
+                 }
+                 else if (drRmvMap != null) {
+                     assert drRmvMap.get(key) != null;
+ 
+                     drVer = drRmvMap.get(key);
+                     drTtl = -1L;
+                     drExpireTime = -1L;
+                 }
+                 else {
+                     drVer = null;
+                     drTtl = -1L;
+                     drExpireTime = -1L;
+                 }
+ 
+                 if (key == null)
+                     continue;
+ 
+                 if (!rmv && val == null && transformClo == null) {
+                     skipped = skip(skipped, key);
+ 
+                     continue;
+                 }
+ 
+                 if (cacheCtx.portableEnabled())
+                     key = (K)cacheCtx.marshalToPortable(key);
+ 
+                 IgniteTxKey<K> txKey = cacheCtx.txKey(key);
+ 
+                 IgniteTxEntry<K, V> txEntry = entry(txKey);
+ 
+                 // First time access.
+                 if (txEntry == null) {
+                     while (true) {
+                         GridCacheEntryEx<K, V> entry;
+ 
+                         if (cached != null) {
+                             entry = cached;
+ 
+                             cached = null;
+                         }
+                         else {
+                             entry = entryEx(cacheCtx, txKey, topologyVersion());
+ 
+                             entry.unswap(true, false);
+                         }
+ 
+                         try {
+                             // Check if lock is being explicitly acquired by the same thread.
+                             if (!implicit && cctx.kernalContext().config().isCacheSanityCheckEnabled() &&
+                                 entry.lockedByThread(threadId, xidVer))
+                                 throw new IgniteCheckedException("Cannot access key within transaction if lock is " +
+                                     "externally held [key=" + key + ", entry=" + entry + ", xidVer=" + xidVer +
+                                     ", threadId=" + threadId +
+                                     ", locNodeId=" + cctx.localNodeId() + ']');
+ 
+                             V old = null;
+ 
+                             boolean readThrough = !F.isEmptyOrNulls(filter) && !F.isAlwaysTrue(filter);
+ 
+                             if (optimistic()) {
+                                 try {
 -                                    //Should read through if filter is specified.
++                                    // Should read through if filter is specified.
+                                     old = entry.innerGet(this,
+                                         /*swap*/false,
+                                         /*read-through*/readThrough,
+                                         /*fail-fast*/false,
+                                         /*unmarshal*/retval,
+                                         /*metrics*/retval,
+                                         /*events*/retval,
+                                         /*temporary*/false,
+                                         CU.subjectId(this, cctx),
+                                         transformClo,
+                                         resolveTaskName(),
 -                                        CU.<K, V>empty());
++                                        CU.<K, V>empty(),
++                                        null);
+                                 }
+                                 catch (GridCacheFilterFailedException e) {
+                                     e.printStackTrace();
+ 
+                                     assert false : "Empty filter failed: " + e;
+                                 }
+                             }
+                             else
+                                 old = retval ? entry.rawGetOrUnmarshal(false) : entry.rawGet();
+ 
+                             if (!filter(entry, filter)) {
+                                 skipped = skip(skipped, key);
+ 
+                                 ret.set(old, false);
+ 
+                                 if (!readCommitted() && old != null) {
+                                     // Enlist failed filters as reads for non-read-committed mode,
+                                     // so future ops will get the same values.
 -                                    txEntry = addEntry(READ, old, null, entry, -1, CU.<K, V>empty(), false, -1L, -1L,
++                                    txEntry = addEntry(READ, old, null, entry, null, CU.<K, V>empty(), false, -1L, -1L,
+                                         null);
+ 
+                                     txEntry.markValid();
+                                 }
+ 
+                                 if (readCommitted() || old == null)
+                                     cacheCtx.evicts().touch(entry, topologyVersion());
+ 
+                                 break; // While.
+                             }
+ 
+                             txEntry = addEntry(lockOnly ? NOOP : rmv ? DELETE : transformClo != null ? TRANSFORM :
 -                                old != null ? UPDATE : CREATE, val, transformClo, entry, ttl, filter, true, drTtl,
++                                old != null ? UPDATE : CREATE, val, transformClo, entry, expiryPlc, filter, true, drTtl,
+                                 drExpireTime, drVer);
+ 
+                             if (!implicit() && readCommitted())
+                                 cacheCtx.evicts().touch(entry, topologyVersion());
+ 
+                             if (groupLock() && !lockOnly)
+                                 txEntry.groupLockEntry(true);
+ 
+                             enlisted.add(key);
+ 
+                             if (!pessimistic() || (groupLock() && !lockOnly)) {
+                                 txEntry.markValid();
+ 
+                                 if (old == null) {
+                                     if (retval && !readThrough) {
+                                         // If return value is required, then we know for sure that there is only
+                                         // one key in the keys collection.
+                                         assert keys.size() == 1;
+ 
+                                         IgniteFuture<Boolean> fut = loadMissing(
+                                             cacheCtx,
+                                             true,
+                                             F.asList(key),
+                                             deserializePortables(cacheCtx),
+                                             new CI2<K, V>() {
+                                                 @Override public void apply(K k, V v) {
+                                                     if (log.isDebugEnabled())
+                                                         log.debug("Loaded value from remote node [key=" + k + ", val=" +
+                                                             v + ']');
+ 
+                                                     ret.set(v, true);
+                                                 }
+                                             });
+ 
+                                         return new GridEmbeddedFuture<>(
+                                             cctx.kernalContext(),
+                                             fut,
+                                             new C2<Boolean, Exception, Set<K>>() {
+                                                 @Override public Set<K> apply(Boolean b, Exception e) {
+                                                     if (e != null)
+                                                         throw new GridClosureException(e);
+ 
+                                                     return Collections.emptySet();
+                                                 }
+                                             }
+                                         );
+                                     }
+                                     else
+                                         ret.set(null, true);
+                                 }
+                                 else
+                                     ret.set(old, true);
+                             }
+                             // Pessimistic.
+                             else
+                                 ret.set(old, true);
+ 
+                             break; // While.
+                         }
+                         catch (GridCacheEntryRemovedException ignore) {
+                             if (log.isDebugEnabled())
+                                 log.debug("Got removed entry in transaction putAll0 method: " + entry);
+                         }
+                     }
+                 }
+                 else {
+                     if (transformClo == null && txEntry.op() == TRANSFORM)
+                         throw new IgniteCheckedException("Failed to enlist write value for key (cannot have update value in " +
+                             "transaction after transform closure is applied): " + key);
+ 
+                     GridCacheEntryEx<K, V> entry = txEntry.cached();
+ 
+                     V v = txEntry.value();
+ 
+                     boolean del = txEntry.op() == DELETE && rmv;
+ 
+                     if (!del) {
+                         if (!filter(entry, filter)) {
+                             skipped = skip(skipped, key);
+ 
+                             ret.set(v, false);
+ 
+                             continue;
+                         }
+ 
+                         txEntry = addEntry(rmv ? DELETE : transformClo != null ? TRANSFORM :
 -                            v != null ? UPDATE : CREATE, val, transformClo, entry, ttl, filter, true, drTtl,
++                            v != null ? UPDATE : CREATE, val, transformClo, entry, expiryPlc, filter, true, drTtl,
+                             drExpireTime, drVer);
+ 
+                         enlisted.add(key);
+                     }
+ 
+                     if (!pessimistic()) {
+                         txEntry.markValid();
+ 
+                         // Set tx entry and return values.
+                         ret.set(v, true);
+                     }
+                 }
+             }
+         }
+         catch (IgniteCheckedException e) {
+             return new GridFinishedFuture<>(cctx.kernalContext(), e);
+         }
+ 
+         return new GridFinishedFuture<>(cctx.kernalContext(), skipped);
+     }
+ 
+     /**
+      * Post lock processing for put or remove.
+      *
+      * @param keys Keys.
+      * @param failed Collection of potentially failed keys (need to populate in this method).
+      * @param transformed Output map where transformed values will be placed.
+      * @param transformMap Transform map.
+      * @param ret Return value.
+      * @param rmv {@code True} if remove.
+      * @param retval Flag to return value or not.
+      * @param filter Filter to check entries.
+      * @return Failed keys.
+      * @throws IgniteCheckedException If error.
+      */
+  

<TRUNCATED>
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f537940c/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/cache/GridCacheCommandHandler.java
----------------------------------------------------------------------


Mime
View raw message