/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.ignite.internal.processors.cache.transactions;

import java.io.Externalizable;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.LinkedList;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import javax.cache.expiry.ExpiryPolicy;
import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.GridDirectTransient;
import org.apache.ignite.internal.IgniteCodeGeneratingFail;
import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
import org.apache.ignite.internal.processors.cache.CacheInvalidStateException;
import org.apache.ignite.internal.processors.cache.CacheInvokeEntry;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.CacheObjectValueContext;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.IgniteExternalizableExpiryPolicy;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.lang.GridAbsClosureX;
import org.apache.ignite.internal.util.lang.GridPeerDeployAware;
import org.apache.ignite.internal.util.tostring.GridToStringBuilder;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
import org.apache.ignite.thread.IgniteThread;
import org.jetbrains.annotations.Nullable;

import static org.apache.ignite.internal.processors.cache.GridCacheOperation.READ;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.TRANSFORM;

/**
 * Transaction entry. Note that it is essential that this class does not override
 * {@link #equals(Object)} method, as transaction entries should use referential
 * equality.
 */
@IgniteCodeGeneratingFail // Field filters should not be generated by MessageCodeGenerator.
public class IgniteTxEntry implements GridPeerDeployAware, Message {
    /** */
    private static final long serialVersionUID = 0L;

    /** Dummy version for non-existing entry read in SERIALIZABLE transaction. */
    public static final GridCacheVersion SER_READ_EMPTY_ENTRY_VER = new GridCacheVersion(0, 0, 0);

    /** Dummy version for any existing entry read in SERIALIZABLE transaction. */
    public static final GridCacheVersion SER_READ_NOT_EMPTY_VER = new GridCacheVersion(0, 0, 1);

    /** */
    public static final GridCacheVersion GET_ENTRY_INVALID_VER_UPDATED = new GridCacheVersion(0, 0, 2);

    /** */
    public static final GridCacheVersion GET_ENTRY_INVALID_VER_AFTER_GET = new GridCacheVersion(0, 0, 3);

    /** Skip store flag bit mask. */
    private static final int TX_ENTRY_SKIP_STORE_FLAG_MASK = 0x01;

    /** Keep binary flag. */
    private static final int TX_ENTRY_KEEP_BINARY_FLAG_MASK = 0x02;

    /** Flag indicating that old value for 'invoke' operation was non null on primary node. */
    private static final int TX_ENTRY_OLD_VAL_ON_PRIMARY = 0x04;

    /** Flag indicating that near cache is enabled on originating node and it should be added as reader. */
    private static final int TX_ENTRY_ADD_READER_FLAG_MASK = 0x08;

    /** Prepared flag updater. */
    private static final AtomicIntegerFieldUpdater<IgniteTxEntry> PREPARED_UPD =
        AtomicIntegerFieldUpdater.newUpdater(IgniteTxEntry.class, "prepared");

    /** Owning transaction. */
    @GridToStringExclude
    @GridDirectTransient
    public IgniteInternalTx tx;

    /** Cache key. */
    @GridToStringExclude
    private KeyCacheObject key;

    /** Cache ID. */
    @GridToStringExclude
    private int cacheId;

    /** Transient tx key. */
    @GridDirectTransient
    private IgniteTxKey txKey;

    /** Cache value. */
    @GridToStringInclude
    private TxEntryValueHolder val = new TxEntryValueHolder();

    /** Visible value for peek. */
    @GridToStringInclude
    @GridDirectTransient
    private TxEntryValueHolder prevVal = new TxEntryValueHolder();

    /** Old value before update. */
    @GridToStringInclude
    private TxEntryValueHolder oldVal = new TxEntryValueHolder();

    /** Transform. */
    @GridToStringInclude
    @GridDirectTransient
    private Collection<T2<EntryProcessor<Object, Object, Object>, Object[]>> entryProcessorsCol;

    /** Transient field for calculated entry processor value. */
    @GridDirectTransient
    private T2<GridCacheOperation, CacheObject> entryProcessorCalcVal;

    /** Transform closure bytes. */
    @GridToStringExclude
    private byte[] transformClosBytes;

    /** Time to live. */
    private long ttl;

    /** DR expire time (explicit) */
    private long conflictExpireTime = CU.EXPIRE_TIME_CALCULATE;

    /** Conflict version. */
    private GridCacheVersion conflictVer;

    /** Explicit lock version if there is one. */
    @GridToStringInclude
    private GridCacheVersion explicitVer;

    /** DHT version. */
    @GridDirectTransient
    private volatile GridCacheVersion dhtVer;

    /** Put filters. */
    @GridToStringInclude
    private CacheEntryPredicate[] filters;

    /** Flag indicating whether filters passed. Used for fast-commit transactions. */
    @GridDirectTransient
    private boolean filtersPassed;

    /** Flag indicating that filter is set and can not be replaced. */
    @GridDirectTransient
    private boolean filtersSet;

    /** Underlying cache entry. */
    @GridDirectTransient
    private volatile GridCacheEntryEx entry;

    /** Cache registry. */
    @GridDirectTransient
    private GridCacheContext<?, ?> ctx;

    /** Prepared flag to prevent multiple candidate add. */
    @GridDirectTransient
    private transient volatile int prepared;

    /** Lock flag for collocated cache. */
    @GridDirectTransient
    private transient boolean locked;

    /** Assigned node ID (required only for partitioned cache). */
    @GridDirectTransient
    private UUID nodeId;

    /** Flag if this node is a back up node. */
    @GridDirectTransient
    private boolean locMapped;

    /** Expiry policy. */
    @GridDirectTransient
    private ExpiryPolicy expiryPlc;

    /** Expiry policy transfer flag. */
    @GridDirectTransient
    private boolean transferExpiryPlc;

    /** Expiry policy bytes. */
    private byte[] expiryPlcBytes;

    /** Additional flags. */
    private byte flags;

    /** Partition update counter. */
    @GridDirectTransient
    private long partUpdateCntr;

    /** */
    private GridCacheVersion serReadVer;

    /** */
    @GridDirectTransient
    @GridToStringExclude
    private transient @Nullable GridAbsClosureX cqNotifyC;

    /**
     * Required by {@link Externalizable}
     */
    public IgniteTxEntry() {
        /* No-op. */
    }

    /**
     * This constructor is meant for remote transactions.
     *
     * @param ctx Cache registry.
     * @param tx Owning transaction.
     * @param op Operation.
     * @param val Value.
     * @param ttl Time to live.
     * @param conflictExpireTime DR expire time.
     * @param entry Cache entry.
     * @param conflictVer Data center replication version.
     * @param skipStore Skip store flag.
     */
    public IgniteTxEntry(GridCacheContext<?, ?> ctx,
        IgniteInternalTx tx,
        GridCacheOperation op,
        CacheObject val,
        long ttl,
        long conflictExpireTime,
        GridCacheEntryEx entry,
        @Nullable GridCacheVersion conflictVer,
        boolean skipStore,
        boolean keepBinary
    ) {
        assert ctx != null;
        assert tx != null;
        assert op != null;
        assert entry != null;

        this.ctx = ctx;
        this.tx = tx;
        this.val.value(op, val, false, false);
        this.entry = entry;
        this.ttl = ttl;
        this.conflictExpireTime = conflictExpireTime;
        this.conflictVer = conflictVer;

        skipStore(skipStore);
        keepBinary(keepBinary);

        key = entry.key();

        cacheId = entry.context().cacheId();
    }

    /**
     * This constructor is meant for local transactions.
     *
     * @param ctx Cache registry.
     * @param tx Owning transaction.
     * @param op Operation.
     * @param val Value.
     * @param entryProcessor Entry processor.
     * @param invokeArgs Optional arguments for EntryProcessor.
     * @param ttl Time to live.
     * @param entry Cache entry.
     * @param filters Put filters.
     * @param conflictVer Data center replication version.
     * @param skipStore Skip store flag.
     * @param addReader Add reader flag.
     */
    public IgniteTxEntry(GridCacheContext<?, ?> ctx,
        IgniteInternalTx tx,
        GridCacheOperation op,
        CacheObject val,
        EntryProcessor<Object, Object, Object> entryProcessor,
        Object[] invokeArgs,
        long ttl,
        GridCacheEntryEx entry,
        CacheEntryPredicate[] filters,
        GridCacheVersion conflictVer,
        boolean skipStore,
        boolean keepBinary,
        boolean addReader
    ) {
        assert ctx != null;
        assert tx != null;
        assert op != null;
        assert entry != null;

        this.ctx = ctx;
        this.tx = tx;
        this.val.value(op, val, false, false);
        this.entry = entry;
        this.ttl = ttl;
        this.filters = filters;
        this.conflictVer = conflictVer;

        skipStore(skipStore);
        keepBinary(keepBinary);
        addReader(addReader);

        if (entryProcessor != null)
            addEntryProcessor(entryProcessor, invokeArgs);

        key = entry.key();

        cacheId = entry.context().cacheId();
    }

    /**
     * @return Cache context for this tx entry.
     */
    public GridCacheContext<?, ?> context() {
        return ctx;
    }

    /**
     * @param ctx Cache context for this tx entry.
     */
    public void context(GridCacheContext<?, ?> ctx) {
        this.ctx = ctx;
    }

    /**
     * @return Flag indicating if this entry is affinity mapped to the same node.
     */
    public boolean locallyMapped() {
        return locMapped;
    }

    /**
     * @param locMapped Flag indicating if this entry is affinity mapped to the same node.
     */
    public void locallyMapped(boolean locMapped) {
        this.locMapped = locMapped;
    }

    /**
     * @param ctx Context.
     * @return Clean copy of this entry.
     */
    public IgniteTxEntry cleanCopy(GridCacheContext<?, ?> ctx) {
        IgniteTxEntry cp = new IgniteTxEntry();

        cp.key = key;
        cp.cacheId = cacheId;
        cp.ctx = ctx;

        cp.val = new TxEntryValueHolder();

        cp.filters = filters;
        cp.val.value(val.op(), val.value(), val.hasWriteValue(), val.hasReadValue());
        cp.entryProcessorsCol = entryProcessorsCol;
        cp.ttl = ttl;
        cp.conflictExpireTime = conflictExpireTime;
        cp.explicitVer = explicitVer;
        cp.conflictVer = conflictVer;
        cp.expiryPlc = expiryPlc;
        cp.flags = flags;
        cp.serReadVer = serReadVer;

        return cp;
    }

    /**
     * @return Node ID.
     */
    public UUID nodeId() {
        return nodeId;
    }

    /**
     * @param nodeId Node ID.
     */
    public void nodeId(UUID nodeId) {
        this.nodeId = nodeId;
    }

    /**
     * @return DHT version.
     */
    public GridCacheVersion dhtVersion() {
        return dhtVer;
    }

    /**
     * @param dhtVer DHT version.
     */
    public void dhtVersion(GridCacheVersion dhtVer) {
        this.dhtVer = dhtVer;
    }

    /**
     * @return {@code True} if tx entry was marked as locked.
     */
    public boolean locked() {
        return locked;
    }

    /**
     * Marks tx entry as locked.
     */
    public void markLocked() {
        locked = true;
    }

    /**
     * Sets partition counter.
     *
     * @param partCntr Partition counter.
     */
    public void updateCounter(long partCntr) {
        this.partUpdateCntr = partCntr;
    }

    /**
     * @return Partition index.
     */
    public long updateCounter() {
        return partUpdateCntr;
    }

    /**
     * @param val Value to set.
     */
    public void setAndMarkValid(CacheObject val) {
        setAndMarkValid(op(), val, this.val.hasWriteValue(), this.val.hasReadValue());
    }

    /**
     * @param op Operation.
     * @param val Value to set.
     */
    void setAndMarkValid(GridCacheOperation op, CacheObject val) {
        setAndMarkValid(op, val, this.val.hasWriteValue(), this.val.hasReadValue());
    }

    /**
     * @param op Operation.
     * @param val Value to set.
     * @param hasReadVal Has read value flag.
     * @param hasWriteVal Has write value flag.
     */
    void setAndMarkValid(GridCacheOperation op, CacheObject val, boolean hasWriteVal, boolean hasReadVal) {
        this.val.value(op, val, hasWriteVal, hasReadVal);

        markValid();
    }

    /**
     * Marks this entry as value-has-bean-read. Effectively, makes values enlisted to transaction visible
     * to further peek operations.
     */
    public void markValid() {
        prevVal.value(val.op(), val.value(), val.hasWriteValue(), val.hasReadValue());
    }

    /**
     * Marks entry as prepared.
     *
     * @return True if entry was marked prepared by this call.
     */
    boolean markPrepared() {
        return PREPARED_UPD.compareAndSet(this, 0, 1);
    }

    /**
     * @return Entry key.
     */
    public KeyCacheObject key() {
        return key;
    }

    /**
     * @return Cache ID.
     */
    public int cacheId() {
        return cacheId;
    }

    /**
     * Sets skip store flag value.
     *
     * @param skipStore Skip store flag.
     */
    public void skipStore(boolean skipStore) {
        setFlag(skipStore, TX_ENTRY_SKIP_STORE_FLAG_MASK);
    }

    /**
     * @return Skip store flag.
     */
    public boolean skipStore() {
        return isFlag(TX_ENTRY_SKIP_STORE_FLAG_MASK);
    }

    /**
     * @param oldValOnPrimary {@code True} If old value for was non null on primary node.
     */
    public void oldValueOnPrimary(boolean oldValOnPrimary) {
        setFlag(oldValOnPrimary, TX_ENTRY_OLD_VAL_ON_PRIMARY);
    }

    /**
     * @return {@code True} If old value for 'invoke' operation was non null on primary node.
     */
    boolean oldValueOnPrimary() {
        return isFlag(TX_ENTRY_OLD_VAL_ON_PRIMARY);
    }

    /**
     * Sets keep binary flag value.
     *
     * @param keepBinary Keep binary flag value.
     */
    public void keepBinary(boolean keepBinary) {
        setFlag(keepBinary, TX_ENTRY_KEEP_BINARY_FLAG_MASK);
    }

    /**
     * @return Keep binary flag value.
     */
    public boolean keepBinary() {
        return isFlag(TX_ENTRY_KEEP_BINARY_FLAG_MASK);
    }

    /**
     * @param addReader Add reader flag.
     */
    public void addReader(boolean addReader) {
        setFlag(addReader, TX_ENTRY_ADD_READER_FLAG_MASK);
    }

    /**
     * @return Add reader flag.
     */
    public boolean addReader() {
        return isFlag(TX_ENTRY_ADD_READER_FLAG_MASK);
    }

    /**
     * Sets flag mask.
     *
     * @param flag Set or clear.
     * @param mask Mask.
     */
    private void setFlag(boolean flag, int mask) {
        flags = flag ? (byte)(flags | mask) : (byte)(flags & ~mask);
    }

    /**
     * Reads flag mask.
     *
     * @param mask Mask to read.
     * @return Flag value.
     */
    private boolean isFlag(int mask) {
        return (flags & mask) != 0;
    }

    /**
     * @return Tx key.
     */
    public IgniteTxKey txKey() {
        if (txKey == null)
            txKey = new IgniteTxKey(key, cacheId);

        return txKey;
    }

    /**
     * @return Underlying cache entry.
     */
    public GridCacheEntryEx cached() {
        return entry;
    }

    /**
     * @param entry Cache entry.
     */
    public void cached(GridCacheEntryEx entry) {
        assert entry == null || entry.context() == ctx : "Invalid entry assigned to tx entry [txEntry=" + this +
            ", entry=" + entry +
            ", ctxNear=" + ctx.isNear() +
            ", ctxDht=" + ctx.isDht() + ']';

        this.entry = entry;
    }

    /**
     * @return Entry value.
     */
    @Nullable public CacheObject value() {
        return val.value();
    }

    /**
     * @return Old value.
     */
    @Nullable public CacheObject oldValue() {
        return oldVal != null ? oldVal.value() : null;
    }

    /**
     * @param oldVal Old value.
     */
    public void oldValue(CacheObject oldVal) {
        if (this.oldVal == null)
            this.oldVal = new TxEntryValueHolder();

        this.oldVal.value(op(), oldVal, true, true);
    }

    /**
     * @return {@code True} if old value present.
     */
    public boolean hasOldValue() {
        return oldVal != null && oldVal.hasValue();
    }

    /**
     * @return {@code True} if has value explicitly set.
     */
    public boolean hasValue() {
        return val.hasValue();
    }

    /**
     * @return {@code True} if has write value set.
     */
    public boolean hasWriteValue() {
        return val.hasWriteValue();
    }

    /**
     * @return {@code True} if has read value set.
     */
    public boolean hasReadValue() {
        return val.hasReadValue();
    }

    /**
     * @return Value visible for peek.
     */
    @Nullable public CacheObject previousValue() {
        return prevVal.value();
    }

    /**
     * @return {@code True} if has previous value explicitly set.
     */
    public boolean hasPreviousValue() {
        return prevVal.hasValue();
    }

    /**
     * @return Previous operation to revert entry in case of filter failure.
     */
    @Nullable public GridCacheOperation previousOperation() {
        return prevVal.op();
    }

    /**
     * @return Time to live.
     */
    public long ttl() {
        return ttl;
    }

    /**
     * @param ttl Time to live.
     */
    public void ttl(long ttl) {
        this.ttl = ttl;
    }

    /**
     * @return Conflict expire time.
     */
    public long conflictExpireTime() {
        return conflictExpireTime;
    }

    /**
     * @param conflictExpireTime Conflict expire time.
     */
    public void conflictExpireTime(long conflictExpireTime) {
        this.conflictExpireTime = conflictExpireTime;
    }

    /**
     * @param val Entry value.
     * @param writeVal Write value flag.
     * @param readVal Read value flag.
     */
    public void value(@Nullable CacheObject val, boolean writeVal, boolean readVal) {
        this.val.value(this.val.op(), val, writeVal, readVal);
    }

    /**
     * Sets read value if this tx entry does not have write value yet.
     *
     * @param val Read value to set.
     */
    public void readValue(@Nullable CacheObject val) {
        this.val.value(this.val.op(), val, false, true);
    }

    /**
     * @param entryProcessor Entry processor.
     * @param invokeArgs Optional arguments for EntryProcessor.
     */
    public void addEntryProcessor(EntryProcessor<Object, Object, Object> entryProcessor, Object[] invokeArgs) {
        if (entryProcessorsCol == null)
            entryProcessorsCol = new LinkedList<>();

        entryProcessorsCol.add(new T2<>(entryProcessor, invokeArgs));

        // Must clear transform closure bytes since collection has changed.
        transformClosBytes = null;

        val.op(TRANSFORM);
    }

    /**
     * @return Collection of entry processors.
     */
    public Collection<T2<EntryProcessor<Object, Object, Object>, Object[]>> entryProcessors() {
        return entryProcessorsCol;
    }

    /**
     * @param cacheVal Value.
     * @return New value.
     */
    @SuppressWarnings("unchecked")
    public CacheObject applyEntryProcessors(CacheObject cacheVal) {
        GridCacheVersion ver;

        try {
            ver = entry.version();
        }
        catch (GridCacheEntryRemovedException ignore) {
            assert tx == null || tx.optimistic() : tx;

            ver = null;
        }

        Object val = null;
        Object keyVal = null;

        for (T2<EntryProcessor<Object, Object, Object>, Object[]> t : entryProcessors()) {
            IgniteThread.onEntryProcessorEntered(true);

            try {
                CacheInvokeEntry<Object, Object> invokeEntry = new CacheInvokeEntry(key, keyVal, cacheVal, val,
                    ver, keepBinary(), cached());

                EntryProcessor processor = t.get1();

                processor.process(invokeEntry, t.get2());

                val = invokeEntry.getValue();

                keyVal = invokeEntry.key();
            }
            catch (Exception ignore) {
                // No-op.
            }
            finally {
                IgniteThread.onEntryProcessorLeft();
            }
        }

        return ctx.toCacheObject(val);
    }

    /**
     * @param entryProcessorsCol Collection of entry processors.
     */
    public void entryProcessors(
        @Nullable Collection<T2<EntryProcessor<Object, Object, Object>, Object[]>> entryProcessorsCol) {
        this.entryProcessorsCol = entryProcessorsCol;

        // Must clear transform closure bytes since collection has changed.
        transformClosBytes = null;
    }

    /**
     * @return Cache operation.
     */
    public GridCacheOperation op() {
        return val.op();
    }

    /**
     * @param op Cache operation.
     */
    public void op(GridCacheOperation op) {
        val.op(op);
    }

    /**
     * @return {@code True} if read entry.
     */
    public boolean isRead() {
        return op() == READ;
    }

    /**
     * @param explicitVer Explicit version.
     */
    public void explicitVersion(GridCacheVersion explicitVer) {
        this.explicitVer = explicitVer;
    }

    /**
     * @return Explicit version.
     */
    public GridCacheVersion explicitVersion() {
        return explicitVer;
    }

    /**
     * @return Conflict version.
     */
    @Nullable public GridCacheVersion conflictVersion() {
        return conflictVer;
    }

    /**
     * @param conflictVer Conflict version.
     */
    public void conflictVersion(@Nullable GridCacheVersion conflictVer) {
        this.conflictVer = conflictVer;
    }

    /**
     * @return Put filters.
     */
    public CacheEntryPredicate[] filters() {
        return filters;
    }

    /**
     * @param filters Put filters.
     */
    public void filters(CacheEntryPredicate[] filters) {
        this.filters = filters;
    }

    /**
     * @return {@code True} if filters passed for fast-commit transactions.
     */
    public boolean filtersPassed() {
        return filtersPassed;
    }

    /**
     * @param filtersPassed {@code True} if filters passed for fast-commit transactions.
     */
    public void filtersPassed(boolean filtersPassed) {
        this.filtersPassed = filtersPassed;
    }

    /**
     * @return {@code True} if filters are set.
     */
    public boolean filtersSet() {
        return filtersSet;
    }

    /**
     * @param filtersSet {@code True} if filters are set and should not be replaced.
     */
    public void filtersSet(boolean filtersSet) {
        this.filtersSet = filtersSet;
    }

    /**
     * @param ctx Context.
     * @param transferExpiry {@code True} if expire policy should be marshalled.
     * @throws IgniteCheckedException If failed.
     */
    public void marshal(GridCacheSharedContext<?, ?> ctx, boolean transferExpiry) throws IgniteCheckedException {
        if (filters != null) {
            for (CacheEntryPredicate p : filters) {
                if (p != null)
                    p.prepareMarshal(this.ctx);
            }
        }

        // Do not serialize filters if they are null.
        if (transformClosBytes == null && entryProcessorsCol != null)
            transformClosBytes = CU.marshal(this.ctx, entryProcessorsCol);

        if (transferExpiry)
            transferExpiryPlc = expiryPlc != null && expiryPlc != this.ctx.expiry();

        key.prepareMarshal(context().cacheObjectContext());

        val.marshal(context());

        if (transferExpiryPlc) {
            if (expiryPlcBytes == null)
                expiryPlcBytes = CU.marshal(this.ctx, new IgniteExternalizableExpiryPolicy(expiryPlc));
        }
        else
            expiryPlcBytes = null;
    }

    /**
     * Unmarshalls entry.
     *
     * @param ctx Cache context.
     * @param near Near flag.
     * @param clsLdr Class loader.
     * @throws IgniteCheckedException If un-marshalling failed.
     */
    public void unmarshal(
        GridCacheSharedContext<?, ?> ctx,
        boolean near,
        ClassLoader clsLdr
    ) throws IgniteCheckedException {

        if (this.ctx == null) {
            GridCacheContext<?, ?> cacheCtx = ctx.cacheContext(cacheId);

            if (cacheCtx == null)
                throw new CacheInvalidStateException(
                    "Failed to perform cache operation (cache is stopped), cacheId=" + cacheId);

            if (cacheCtx.isNear() && !near)
                cacheCtx = cacheCtx.near().dht().context();
            else if (!cacheCtx.isNear() && near)
                cacheCtx = cacheCtx.dht().near().context();

            this.ctx = cacheCtx;
        }

        CacheObjectValueContext coctx = this.ctx.cacheObjectContext();

        if (coctx == null)
            throw new CacheInvalidStateException(
                    "Failed to perform cache operation (cache is stopped), cacheId=" + cacheId);

        // Unmarshal transform closure anyway if it exists.
        if (transformClosBytes != null && entryProcessorsCol == null)
            entryProcessorsCol = U.unmarshal(ctx, transformClosBytes, U.resolveClassLoader(clsLdr, ctx.gridConfig()));

        if (filters == null)
            filters = CU.empty0();
        else {
            for (CacheEntryPredicate p : filters) {
                if (p != null)
                    p.finishUnmarshal(this.ctx, clsLdr);
            }
        }

        key.finishUnmarshal(coctx, clsLdr);

        val.unmarshal(coctx, clsLdr);

        if (expiryPlcBytes != null && expiryPlc == null)
            expiryPlc = U.unmarshal(ctx, expiryPlcBytes, U.resolveClassLoader(clsLdr, ctx.gridConfig()));
    }

    /**
     * @param expiryPlc Expiry policy.
     */
    public void expiry(@Nullable ExpiryPolicy expiryPlc) {
        this.expiryPlc = expiryPlc;
    }

    /**
     * @return Expiry policy.
     */
    @Nullable public ExpiryPolicy expiry() {
        return expiryPlc;
    }

    /**
     * @return Entry processor calculated value.
     */
    public T2<GridCacheOperation, CacheObject> entryProcessorCalculatedValue() {
        return entryProcessorCalcVal;
    }

    /**
     * @param entryProcessorCalcVal Entry processor calculated value.
     */
    public void entryProcessorCalculatedValue(T2<GridCacheOperation, CacheObject> entryProcessorCalcVal) {
        assert entryProcessorCalcVal != null;

        this.entryProcessorCalcVal = entryProcessorCalcVal;
    }

    /**
     * Gets stored entry version. Version is stored for all entries in serializable transaction or
     * when value is read using {@link IgniteCache#getEntry(Object)} method.
     *
     * @return Entry version.
     */
    @Nullable public GridCacheVersion entryReadVersion() {
        return serReadVer;
    }

    /**
     * @param ver Entry version.
     */
    public void entryReadVersion(GridCacheVersion ver) {
        assert this.serReadVer == null : "Wrong version [serReadVer=" + serReadVer + ", ver=" + ver + "]";
        assert ver != null;

        this.serReadVer = ver;
    }

    /**
     * Clears recorded read version, should be done before starting commit of not serializable/optimistic transaction.
     */
    public void clearEntryReadVersion() {
        serReadVer = null;
    }

    /** {@inheritDoc} */
    @Override public void onAckReceived() {
        // No-op.
    }

    /** {@inheritDoc} */
    @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
        writer.setBuffer(buf);

        if (!writer.isHeaderWritten()) {
            if (!writer.writeHeader(directType(), fieldsCount()))
                return false;

            writer.onHeaderWritten();
        }

        switch (writer.state()) {
            case 0:
                if (!writer.writeInt("cacheId", cacheId))
                    return false;

                writer.incrementState();

            case 1:
                if (!writer.writeLong("conflictExpireTime", conflictExpireTime))
                    return false;

                writer.incrementState();

            case 2:
                if (!writer.writeMessage("conflictVer", conflictVer))
                    return false;

                writer.incrementState();

            case 3:
                if (!writer.writeByteArray("expiryPlcBytes", expiryPlcBytes))
                    return false;

                writer.incrementState();

            case 4:
                if (!writer.writeMessage("explicitVer", explicitVer))
                    return false;

                writer.incrementState();

            case 5:
                if (!writer.writeObjectArray("filters",
                    !F.isEmptyOrNulls(filters) ? filters : null, MessageCollectionItemType.MSG))
                    return false;

                writer.incrementState();

            case 6:
                if (!writer.writeByte("flags", flags))
                    return false;

                writer.incrementState();

            case 7:
                if (!writer.writeMessage("key", key))
                    return false;

                writer.incrementState();

            case 8:
                if (!writer.writeMessage("oldVal", oldVal))
                    return false;

                writer.incrementState();

            case 9:
                if (!writer.writeMessage("serReadVer", serReadVer))
                    return false;

                writer.incrementState();

            case 10:
                if (!writer.writeByteArray("transformClosBytes", transformClosBytes))
                    return false;

                writer.incrementState();

            case 11:
                if (!writer.writeLong("ttl", ttl))
                    return false;

                writer.incrementState();

            case 12:
                if (!writer.writeMessage("val", val))
                    return false;

                writer.incrementState();

        }

        return true;
    }

    /** {@inheritDoc} */
    @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
        reader.setBuffer(buf);

        if (!reader.beforeMessageRead())
            return false;

        switch (reader.state()) {
            case 0:
                cacheId = reader.readInt("cacheId");

                if (!reader.isLastRead())
                    return false;

                reader.incrementState();

            case 1:
                conflictExpireTime = reader.readLong("conflictExpireTime");

                if (!reader.isLastRead())
                    return false;

                reader.incrementState();

            case 2:
                conflictVer = reader.readMessage("conflictVer");

                if (!reader.isLastRead())
                    return false;

                reader.incrementState();

            case 3:
                expiryPlcBytes = reader.readByteArray("expiryPlcBytes");

                if (!reader.isLastRead())
                    return false;

                reader.incrementState();

            case 4:
                explicitVer = reader.readMessage("explicitVer");

                if (!reader.isLastRead())
                    return false;

                reader.incrementState();

            case 5:
                filters = reader.readObjectArray("filters", MessageCollectionItemType.MSG, CacheEntryPredicate.class);

                if (!reader.isLastRead())
                    return false;

                reader.incrementState();

            case 6:
                flags = reader.readByte("flags");

                if (!reader.isLastRead())
                    return false;

                reader.incrementState();

            case 7:
                key = reader.readMessage("key");

                if (!reader.isLastRead())
                    return false;

                reader.incrementState();

            case 8:
                oldVal = reader.readMessage("oldVal");

                if (!reader.isLastRead())
                    return false;

                reader.incrementState();

            case 9:
                serReadVer = reader.readMessage("serReadVer");

                if (!reader.isLastRead())
                    return false;

                reader.incrementState();

            case 10:
                transformClosBytes = reader.readByteArray("transformClosBytes");

                if (!reader.isLastRead())
                    return false;

                reader.incrementState();

            case 11:
                ttl = reader.readLong("ttl");

                if (!reader.isLastRead())
                    return false;

                reader.incrementState();

            case 12:
                val = reader.readMessage("val");

                if (!reader.isLastRead())
                    return false;

                reader.incrementState();

        }

        return reader.afterMessageRead(IgniteTxEntry.class);
    }

    /** {@inheritDoc} */
    @Override public short directType() {
        return 100;
    }

    /** {@inheritDoc} */
    @Override public byte fieldsCount() {
        return 13;
    }

    /** {@inheritDoc} */
    @Override public Class<?> deployClass() {
        ClassLoader clsLdr = getClass().getClassLoader();

        CacheObject val = value();

        // First of all check classes that may be loaded by class loader other than application one.
        return key != null && !clsLdr.equals(key.getClass().getClassLoader()) ?
            key.getClass() : val != null ? val.getClass() : getClass();
    }

    /**
     */
    public GridAbsClosureX cqNotifyClosure() {
        return cqNotifyC;
    }

    /**
     * @param clo Clo.
     */
    public void cqNotifyClosure(GridAbsClosureX clo) {
        cqNotifyC = clo;
    }

    /** {@inheritDoc} */
    @Override public ClassLoader classLoader() {
        return deployClass().getClassLoader();
    }

    /** {@inheritDoc} */
    @Override public String toString() {
        return GridToStringBuilder.toString(IgniteTxEntry.class, this, "xidVer", tx == null ? "null" : tx.xidVersion());
    }
}
