/*
 * 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.datastreamer;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import javax.cache.CacheException;
import javax.cache.expiry.ExpiryPolicy;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.IgniteDataStreamerTimeoutException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteInterruptedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.cluster.ClusterTopologyException;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
import org.apache.ignite.internal.managers.communication.GridIoPolicy;
import org.apache.ignite.internal.managers.communication.GridMessageListener;
import org.apache.ignite.internal.managers.deployment.GridDeployment;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.affinity.GridAffinityProcessor;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.CacheObjectContext;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
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.GridCacheGateway;
import org.apache.ignite.internal.processors.cache.GridCacheUtils;
import org.apache.ignite.internal.processors.cache.IgniteCacheFutureImpl;
import org.apache.ignite.internal.processors.cache.IgniteCacheProxy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtInvalidPartitionException;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessor;
import org.apache.ignite.internal.processors.dr.GridDrType;
import org.apache.ignite.internal.util.GridConcurrentHashSet;
import org.apache.ignite.internal.util.GridSpinBusyLock;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.future.IgniteFinishedFutureImpl;
import org.apache.ignite.internal.util.lang.GridPeerDeployAware;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.GPC;
import org.apache.ignite.internal.util.typedef.internal.LT;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.security.SecurityPermission;
import org.apache.ignite.stream.StreamReceiver;
import org.jetbrains.annotations.Nullable;
import org.jsr166.ConcurrentHashMap8;
import org.jsr166.LongAdder8;

import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
import static org.apache.ignite.internal.GridTopic.TOPIC_DATASTREAM;
import static org.apache.ignite.internal.managers.communication.GridIoPolicy.PUBLIC_POOL;

/**
 * Data streamer implementation.
 */
@SuppressWarnings("unchecked")
public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed {
    /** Default policy resolver. */
    private static final DefaultIoPolicyResolver DFLT_IO_PLC_RSLVR = new DefaultIoPolicyResolver();

    /** Isolated receiver. */
    private static final StreamReceiver ISOLATED_UPDATER = new IsolatedUpdater();

    /** Amount of permissions should be available to continue new data processing. */
    private static final int REMAP_SEMAPHORE_PERMISSIONS_COUNT = Integer.MAX_VALUE;

    /** Cache receiver. */
    private StreamReceiver<K, V> rcvr = ISOLATED_UPDATER;

    /** */
    private byte[] updaterBytes;

    /** IO policy resovler for data load request. */
    private IgniteClosure<ClusterNode, Byte> ioPlcRslvr = DFLT_IO_PLC_RSLVR;

    /** Max remap count before issuing an error. */
    private static final int DFLT_MAX_REMAP_CNT = 32;

    /** Log reference. */
    private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();

    /** Logger. */
    private static IgniteLogger log;

    /** Cache name ({@code null} for default cache). */
    private final String cacheName;

    /** Per-node buffer size. */
    @SuppressWarnings("FieldAccessedSynchronizedAndUnsynchronized")
    private int bufSize = DFLT_PER_NODE_BUFFER_SIZE;

    /** */
    private int parallelOps = DFLT_MAX_PARALLEL_OPS;

    /** */
    private long timeout = DFLT_UNLIMIT_TIMEOUT;

    /** */
    private long autoFlushFreq;

    /** Mapping. */
    @GridToStringInclude
    private ConcurrentMap<UUID, Buffer> bufMappings = new ConcurrentHashMap8<>();

    /** Discovery listener. */
    private final GridLocalEventListener discoLsnr;

    /** Context. */
    private final GridKernalContext ctx;

    /** */
    private final IgniteCacheObjectProcessor cacheObjProc;

    /** */
    private final CacheObjectContext cacheObjCtx;

    /** Communication topic for responses. */
    private final Object topic;

    /** */
    private byte[] topicBytes;

    /** {@code True} if data loader has been cancelled. */
    private volatile boolean cancelled;

    /** Fail counter. */
    private final LongAdder8 failCntr = new LongAdder8();

    /** Active futures of this data loader. */
    @GridToStringInclude
    private final Collection<IgniteInternalFuture<?>> activeFuts = new GridConcurrentHashSet<>();

    /** Closure to remove from active futures. */
    @GridToStringExclude
    private final IgniteInClosure<IgniteInternalFuture<?>> rmvActiveFut = new IgniteInClosure<IgniteInternalFuture<?>>() {
        @Override public void apply(IgniteInternalFuture<?> t) {
            boolean rmv = activeFuts.remove(t);

            assert rmv;

            Throwable err = t.error();

            if (err != null && !(err instanceof IgniteClientDisconnectedCheckedException)) {
                LT.error(log, t.error(), "DataStreamer operation failed.", true);

                failCntr.increment();

                cancelled = true;
            }
        }
    };

    /** Job peer deploy aware. */
    private volatile GridPeerDeployAware jobPda;

    /** Deployment class. */
    private Class<?> depCls;

    /** Future to track loading finish. */
    private final GridFutureAdapter<?> fut;

    /** Public API future to track loading finish. */
    private final IgniteFuture<?> publicFut;

    /** Busy lock. */
    private final GridSpinBusyLock busyLock = new GridSpinBusyLock();

    /** */
    private CacheException disconnectErr;

    /** Closed flag. */
    private final AtomicBoolean closed = new AtomicBoolean();

    /** */
    private volatile long lastFlushTime = U.currentTimeMillis();

    /** */
    private final DelayQueue<DataStreamerImpl<K, V>> flushQ;

    /** */
    private boolean skipStore;

    /** */
    private boolean keepBinary;

    /** */
    private int maxRemapCnt = DFLT_MAX_REMAP_CNT;

    /** Whether a warning at {@link DataStreamerImpl#allowOverwrite()} printed */
    private static boolean isWarningPrinted;

    /** Allows to pause new data processing while failed data processing in progress. */
    private final Semaphore remapSem = new Semaphore(REMAP_SEMAPHORE_PERMISSIONS_COUNT);

    /** */
    private final ConcurrentLinkedDeque<Runnable> dataToRemap = new ConcurrentLinkedDeque<>();

    /** */
    private final AtomicBoolean remapOwning = new AtomicBoolean();

    /**
     * @param ctx Grid kernal context.
     * @param cacheName Cache name.
     * @param flushQ Flush queue.
     */
    public DataStreamerImpl(
        final GridKernalContext ctx,
        @Nullable final String cacheName,
        DelayQueue<DataStreamerImpl<K, V>> flushQ
    ) {
        assert ctx != null;

        this.ctx = ctx;
        this.cacheObjProc = ctx.cacheObjects();

        if (log == null)
            log = U.logger(ctx, logRef, DataStreamerImpl.class);

        CacheConfiguration ccfg = ctx.cache().cacheConfiguration(cacheName);

        try {
            this.cacheObjCtx = ctx.cacheObjects().contextForCache(ccfg);
        }
        catch (IgniteCheckedException e) {
            throw new IgniteException("Failed to initialize cache context.", e);
        }

        this.cacheName = cacheName;
        this.flushQ = flushQ;

        discoLsnr = new GridLocalEventListener() {
            @Override public void onEvent(Event evt) {
                assert evt.type() == EVT_NODE_FAILED || evt.type() == EVT_NODE_LEFT;

                DiscoveryEvent discoEvt = (DiscoveryEvent)evt;

                UUID id = discoEvt.eventNode().id();

                // Remap regular mappings.
                final Buffer buf = bufMappings.remove(id);

                // Only async notification is possible since
                // discovery thread may be trapped otherwise.
                if (buf != null) {
                    waitAffinityAndRun(new Runnable() {
                        @Override public void run() {
                            buf.onNodeLeft();
                        }
                    }, discoEvt.topologyVersion(), true);
                }
            }
        };

        ctx.event().addLocalEventListener(discoLsnr, EVT_NODE_FAILED, EVT_NODE_LEFT);

        // Generate unique topic for this loader.
        topic = TOPIC_DATASTREAM.topic(IgniteUuid.fromUuid(ctx.localNodeId()));

        ctx.io().addMessageListener(topic, new GridMessageListener() {
            @Override public void onMessage(UUID nodeId, Object msg) {
                assert msg instanceof DataStreamerResponse;

                DataStreamerResponse res = (DataStreamerResponse)msg;

                if (log.isDebugEnabled())
                    log.debug("Received data load response: " + res);

                Buffer buf = bufMappings.get(nodeId);

                if (buf != null)
                    buf.onResponse(res, nodeId);

                else if (log.isDebugEnabled())
                    log.debug("Ignoring response since node has left [nodeId=" + nodeId + ", ");
            }
        });

        if (log.isDebugEnabled())
            log.debug("Added response listener within topic: " + topic);

        fut = new DataStreamerFuture(this);

        publicFut = new IgniteCacheFutureImpl<>(fut);

        GridCacheAdapter cache = ctx.cache().internalCache(cacheName);

        if (cache == null) { // Possible, cache is not configured on node.
            assert ccfg != null;

            if (ccfg.getCacheMode() == CacheMode.LOCAL)
                throw new CacheException("Impossible to load Local cache configured remotely.");

            ctx.grid().getOrCreateCache(ccfg);
        }
    }

    /**
     * @param c Closure to run.
     * @param topVer Topology version to wait for.
     * @param async Async flag.
     */
    private void waitAffinityAndRun(final Runnable c, long topVer, boolean async) {
        AffinityTopologyVersion topVer0 = new AffinityTopologyVersion(topVer, 0);

        IgniteInternalFuture<?> fut = ctx.cache().context().exchange().affinityReadyFuture(topVer0);

        if (fut != null && !fut.isDone()) {
            fut.listen(new CI1<IgniteInternalFuture<?>>() {
                @Override public void apply(IgniteInternalFuture<?> fut) {
                    ctx.closure().runLocalSafe(c, true);
                }
            });
        }
        else {
            if (async)
                ctx.closure().runLocalSafe(c, true);
            else
                c.run();
        }
    }

    /**
     * @return Cache object context.
     */
    public CacheObjectContext cacheObjectContext() {
        return cacheObjCtx;
    }

    /**
     * Enters busy lock.
     */
    private void enterBusy() {
        if (!busyLock.enterBusy()) {
            if (disconnectErr != null)
                throw disconnectErr;

            throw new IllegalStateException("Data streamer has been closed.");
        }
        else if (cancelled) {
            busyLock.leaveBusy();

            throw new IllegalStateException("Data streamer has been closed.");
        }
    }

    /**
     * Leaves busy lock.
     */
    private void leaveBusy() {
        busyLock.leaveBusy();
    }

    /** {@inheritDoc} */
    @Override public IgniteFuture<?> future() {
        return publicFut;
    }

    /**
     * @return Internal future.
     */
    public IgniteInternalFuture<?> internalFuture() {
        return fut;
    }

    /** {@inheritDoc} */
    @Override public void deployClass(Class<?> depCls) {
        this.depCls = depCls;
    }

    /** {@inheritDoc} */
    @Override public void receiver(StreamReceiver<K, V> rcvr) {
        A.notNull(rcvr, "rcvr");

        this.rcvr = rcvr;
    }

    /** {@inheritDoc} */
    @Override public boolean allowOverwrite() {
        return rcvr != ISOLATED_UPDATER;
    }

    /** {@inheritDoc} */
    @Override public void allowOverwrite(boolean allow) {
        if (allow == allowOverwrite())
            return;

        ClusterNode node = F.first(ctx.grid().cluster().forCacheNodes(cacheName).nodes());

        if (node == null)
            throw new CacheException("Failed to get node for cache: " + cacheName);

        rcvr = allow ? DataStreamerCacheUpdaters.<K, V>individual() : ISOLATED_UPDATER;
    }

    /** {@inheritDoc} */
    @Override public boolean skipStore() {
        return skipStore;
    }

    /** {@inheritDoc} */
    @Override public void skipStore(boolean skipStore) {
        this.skipStore = skipStore;
    }

    /** {@inheritDoc} */
    @Override public boolean keepBinary() {
        return keepBinary;
    }

    /** {@inheritDoc} */
    @Override public void keepBinary(boolean keepBinary) {
        this.keepBinary = keepBinary;
    }

    /** {@inheritDoc} */
    @Override @Nullable public String cacheName() {
        return cacheName;
    }

    /** {@inheritDoc} */
    @Override public int perNodeBufferSize() {
        return bufSize;
    }

    /** {@inheritDoc} */
    @Override public void perNodeBufferSize(int bufSize) {
        A.ensure(bufSize > 0, "bufSize > 0");

        this.bufSize = bufSize;
    }

    /** {@inheritDoc} */
    @Override public int perNodeParallelOperations() {
        return parallelOps;
    }

    /** {@inheritDoc} */
    @Override public void perNodeParallelOperations(int parallelOps) {
        this.parallelOps = parallelOps;
    }

    /** {@inheritDoc} */
    @Override public void timeout(long timeout) {
        if (timeout < -1 || timeout == 0)
            throw new IllegalArgumentException();

        this.timeout = timeout;
    }

    /** {@inheritDoc} */
    @Override public long timeout() {
        return this.timeout;
    }

    /** {@inheritDoc} */
    @Override public long autoFlushFrequency() {
        return autoFlushFreq;
    }

    /** {@inheritDoc} */
    @Override public void autoFlushFrequency(long autoFlushFreq) {
        A.ensure(autoFlushFreq >= 0, "autoFlushFreq >= 0");

        long old = this.autoFlushFreq;

        if (autoFlushFreq != old) {
            this.autoFlushFreq = autoFlushFreq;

            if (autoFlushFreq != 0 && old == 0)
                flushQ.add(this);
            else if (autoFlushFreq == 0)
                flushQ.remove(this);
        }
    }

    /** {@inheritDoc} */
    @Override public IgniteFuture<?> addData(Map<K, V> entries) throws IllegalStateException {
        A.notNull(entries, "entries");

        return addData(entries.entrySet());
    }

    /** {@inheritDoc} */
    @Override public IgniteFuture<?> addData(Collection<? extends Map.Entry<K, V>> entries) {
        A.notEmpty(entries, "entries");

        checkSecurityPermission(SecurityPermission.CACHE_PUT);

        enterBusy();

        try {
            GridFutureAdapter<Object> resFut = new GridFutureAdapter<>();

            resFut.listen(rmvActiveFut);

            activeFuts.add(resFut);

            Collection<KeyCacheObjectWrapper> keys =
                new GridConcurrentHashSet<>(entries.size(), U.capacity(entries.size()), 1);

            Collection<DataStreamerEntry> entries0 = new ArrayList<>(entries.size());

            for (Map.Entry<K, V> entry : entries) {
                KeyCacheObject key = cacheObjProc.toCacheKeyObject(cacheObjCtx, null, entry.getKey(), true);
                CacheObject val = cacheObjProc.toCacheObject(cacheObjCtx, entry.getValue(), true);

                keys.add(new KeyCacheObjectWrapper(key));

                entries0.add(new DataStreamerEntry(key, val));
            }

            load0(entries0, resFut, keys, 0);

            return new IgniteCacheFutureImpl<>(resFut);
        }
        catch (IgniteDataStreamerTimeoutException e) {
            throw e;
        }
        catch (IgniteException e) {
            return new IgniteFinishedFutureImpl<>(e);
        }
        finally {
            leaveBusy();
        }
    }

    /**
     * @param key Key.
     * @param val Value.
     * @return Future.
     */
    public IgniteFuture<?> addDataInternal(KeyCacheObject key, CacheObject val) {
        return addDataInternal(Collections.singleton(new DataStreamerEntry(key, val)));
    }

    /**
     * @param key Key.
     * @return Future.
     */
    public IgniteFuture<?> removeDataInternal(KeyCacheObject key) {
        return addDataInternal(Collections.singleton(new DataStreamerEntry(key, null)));
    }

    /**
     * @param entries Entries.
     * @return Future.
     */
    public IgniteFuture<?> addDataInternal(Collection<? extends DataStreamerEntry> entries) {
        enterBusy();

        GridFutureAdapter<Object> resFut = new GridFutureAdapter<>();

        try {
            resFut.listen(rmvActiveFut);

            activeFuts.add(resFut);

            Collection<KeyCacheObjectWrapper> keys = null;

            if (entries.size() > 1) {
                keys = new GridConcurrentHashSet<>(entries.size(), U.capacity(entries.size()), 1);

                for (DataStreamerEntry entry : entries)
                    keys.add(new KeyCacheObjectWrapper(entry.getKey()));
            }

            load0(entries, resFut, keys, 0);

            return new IgniteCacheFutureImpl<>(resFut);
        }
        catch (Throwable e) {
            resFut.onDone(e);

            if (e instanceof Error || e instanceof IgniteDataStreamerTimeoutException)
                throw e;

            return new IgniteFinishedFutureImpl<>(e);
        }
        finally {
            leaveBusy();
        }
    }

    /** {@inheritDoc} */
    @Override public IgniteFuture<?> addData(Map.Entry<K, V> entry) {
        A.notNull(entry, "entry");

        return addData(F.asList(entry));
    }

    /** {@inheritDoc} */
    @Override public IgniteFuture<?> addData(K key, V val) {
        A.notNull(key, "key");

        if (val == null)
            checkSecurityPermission(SecurityPermission.CACHE_REMOVE);
        else
            checkSecurityPermission(SecurityPermission.CACHE_PUT);

        KeyCacheObject key0 = cacheObjProc.toCacheKeyObject(cacheObjCtx, null, key, true);
        CacheObject val0 = cacheObjProc.toCacheObject(cacheObjCtx, val, true);

        return addDataInternal(Collections.singleton(new DataStreamerEntry(key0, val0)));
    }

    /** {@inheritDoc} */
    @Override public IgniteFuture<?> removeData(K key) {
        return addData(key, null);
    }

    /**
     * @param ioPlcRslvr IO policy resolver.
     */
    public void ioPolicyResolver(IgniteClosure<ClusterNode, Byte> ioPlcRslvr) {
        this.ioPlcRslvr = ioPlcRslvr;
    }

    /**
     *
     */
    private void acquireRemapSemaphore() throws IgniteInterruptedCheckedException {
        try {
            if (remapSem.availablePermits() != REMAP_SEMAPHORE_PERMISSIONS_COUNT) {
                if (timeout == DFLT_UNLIMIT_TIMEOUT) {
                    // Wait until failed data being processed.
                    remapSem.acquire(REMAP_SEMAPHORE_PERMISSIONS_COUNT);

                    remapSem.release(REMAP_SEMAPHORE_PERMISSIONS_COUNT);
                }
                else {
                    // Wait until failed data being processed.
                    boolean res = remapSem.tryAcquire(REMAP_SEMAPHORE_PERMISSIONS_COUNT, timeout, TimeUnit.MILLISECONDS);

                    if (res)
                        remapSem.release(REMAP_SEMAPHORE_PERMISSIONS_COUNT);
                    else
                        throw new IgniteDataStreamerTimeoutException("Data streamer exceeded timeout " +
                            "while was waiting for failed data resending finished.");
                }
            }
        }
        catch (InterruptedException e) {
            Thread.currentThread().interrupt();

            throw new IgniteInterruptedCheckedException(e);
        }
    }

    /**
     * @param entries Entries.
     * @param resFut Result future.
     * @param activeKeys Active keys.
     * @param remaps Remaps count.
     */
    private void load0(
        Collection<? extends DataStreamerEntry> entries,
        final GridFutureAdapter<Object> resFut,
        @Nullable final Collection<KeyCacheObjectWrapper> activeKeys,
        final int remaps
    ) {
        try {
            assert entries != null;

            final boolean remap = remaps > 0;

            if (!remap) { // Failed data should be processed prior to new data.
                acquireRemapSemaphore();
            }

            if (!isWarningPrinted) {
                synchronized (this) {
                    if (!allowOverwrite() && !isWarningPrinted) {
                        U.warn(log, "Data streamer will not overwrite existing cache entries for better performance " +
                            "(to change, set allowOverwrite to true)");
                    }

                    isWarningPrinted = true;
                }
            }

            Map<ClusterNode, Collection<DataStreamerEntry>> mappings = new HashMap<>();

            boolean initPda = ctx.deploy().enabled() && jobPda == null;

            GridCacheAdapter cache = ctx.cache().internalCache(cacheName);

            if (cache == null)
                throw new IgniteCheckedException("Cache not created or already destroyed.");

            GridCacheContext cctx = cache.context();

            GridCacheGateway gate = null;

            if (!allowOverwrite() && !cctx.isLocal()) { // Cases where cctx required.
                gate = cctx.gate();

                gate.enter();
            }

            try {
                AffinityTopologyVersion topVer = allowOverwrite() || cctx.isLocal() ?
                        ctx.cache().context().exchange().readyAffinityVersion() :
                        cctx.topology().topologyVersion();

                for (DataStreamerEntry entry : entries) {
                    List<ClusterNode> nodes;

                    try {
                        KeyCacheObject key = entry.getKey();

                        assert key != null;

                        if (initPda) {
                            if (cacheObjCtx.addDeploymentInfo())
                                jobPda = new DataStreamerPda(key.value(cacheObjCtx, false),
                                    entry.getValue() != null ? entry.getValue().value(cacheObjCtx, false) : null,
                                    rcvr);
                            else if (rcvr != null)
                                jobPda = new DataStreamerPda(rcvr);

                            initPda = false;
                        }

                        nodes = nodes(key, topVer, cctx);
                    }
                    catch (IgniteCheckedException e) {
                        resFut.onDone(e);

                        return;
                    }

                    if (F.isEmpty(nodes)) {
                        resFut.onDone(new ClusterTopologyException("Failed to map key to node " +
                            "(no nodes with cache found in topology) [infos=" + entries.size() +
                            ", cacheName=" + cacheName + ']'));

                        return;
                    }

                    for (ClusterNode node : nodes) {
                        Collection<DataStreamerEntry> col = mappings.get(node);

                        if (col == null)
                            mappings.put(node, col = new ArrayList<>());

                        col.add(entry);
                    }
                }

                for (final Map.Entry<ClusterNode, Collection<DataStreamerEntry>> e : mappings.entrySet()) {
                    final UUID nodeId = e.getKey().id();

                    Buffer buf = bufMappings.get(nodeId);

                    if (buf == null) {
                        Buffer old = bufMappings.putIfAbsent(nodeId, buf = new Buffer(e.getKey()));

                        if (old != null)
                            buf = old;
                    }

                    final Collection<DataStreamerEntry> entriesForNode = e.getValue();

                    IgniteInClosure<IgniteInternalFuture<?>> lsnr = new IgniteInClosure<IgniteInternalFuture<?>>() {
                        @Override public void apply(IgniteInternalFuture<?> t) {
                            try {
                                t.get();

                                if (activeKeys != null) {
                                    for (DataStreamerEntry e : entriesForNode)
                                        activeKeys.remove(new KeyCacheObjectWrapper(e.getKey()));

                                    if (activeKeys.isEmpty())
                                        resFut.onDone();
                                }
                                else {
                                    assert entriesForNode.size() == 1;

                                    // That has been a single key,
                                    // so complete result future right away.
                                    resFut.onDone();
                                }
                            }
                            catch (IgniteClientDisconnectedCheckedException e1) {
                                if (log.isDebugEnabled())
                                    log.debug("Future finished with disconnect error [nodeId=" + nodeId + ", err=" + e1 + ']');

                                resFut.onDone(e1);
                            }
                            catch (IgniteCheckedException e1) {
                                if (log.isDebugEnabled())
                                    log.debug("Future finished with error [nodeId=" + nodeId + ", err=" + e1 + ']');

                                if (cancelled) {
                                    resFut.onDone(new IgniteCheckedException("Data streamer has been cancelled: " +
                                        DataStreamerImpl.this, e1));
                                }
                                else if (remaps + 1 > maxRemapCnt) {
                                    resFut.onDone(new IgniteCheckedException("Failed to finish operation (too many remaps): "
                                        + remaps, e1));
                                }
                                else {
                                    try {
                                        remapSem.acquire();

                                        final Runnable r = new Runnable() {
                                            @Override public void run() {
                                                try {
                                                    if (cancelled)
                                                        throw new IllegalStateException("DataStreamer closed.");

                                                    load0(entriesForNode, resFut, activeKeys, remaps + 1);
                                                }
                                                catch (Throwable ex) {
                                                    resFut.onDone(
                                                        new IgniteCheckedException("DataStreamer remapping failed. ", ex));
                                                }
                                                finally {
                                                    remapSem.release();
                                                }
                                            }
                                        };

                                        dataToRemap.add(r);

                                        if (!remapOwning.get() && remapOwning.compareAndSet(false, true)) {
                                            ctx.closure().callLocalSafe(new GPC<Boolean>() {
                                                @Override public Boolean call() {
                                                    boolean locked = true;

                                                    while (locked || !dataToRemap.isEmpty()) {
                                                        if (!locked && !remapOwning.compareAndSet(false, true))
                                                            return false;

                                                        try {
                                                            Runnable r = dataToRemap.poll();

                                                            if (r != null)
                                                                r.run();
                                                        }
                                                        finally {
                                                            if (!dataToRemap.isEmpty())
                                                                locked = true;
                                                            else {
                                                                remapOwning.set(false);

                                                                locked = false;
                                                            }
                                                        }
                                                    }

                                                    return true;
                                                }
                                            }, true);
                                        }
                                    }
                                    catch (InterruptedException e2) {
                                        resFut.onDone(e2);
                                    }
                                }
                            }
                        }
                    };

                    final GridFutureAdapter<?> f;

                    try {
                        f = buf.update(entriesForNode, topVer, lsnr, remap);
                    }
                    catch (IgniteInterruptedCheckedException e1) {
                        resFut.onDone(e1);

                        return;
                    }

                    if (ctx.discovery().node(nodeId) == null) {
                        if (bufMappings.remove(nodeId, buf)) {
                            final Buffer buf0 = buf;

                            waitAffinityAndRun(new Runnable() {
                                @Override public void run() {
                                    buf0.onNodeLeft();

                                    if (f != null)
                                        f.onDone(new ClusterTopologyCheckedException("Failed to wait for request completion " +
                                            "(node has left): " + nodeId));
                                }
                            }, ctx.discovery().topologyVersion(), false);
                        }
                    }
                }
            }
            finally {
                if (gate != null)
                    gate.leave();
            }
        }
        catch (Exception ex) {
            resFut.onDone(new IgniteCheckedException("DataStreamer data loading failed.", ex));
        }
    }

    /**
     * @param key Key to map.
     * @param topVer Topology version.
     * @param cctx Context.
     * @return Nodes to send requests to.
     * @throws IgniteCheckedException If failed.
     */
    private List<ClusterNode> nodes(KeyCacheObject key,
        AffinityTopologyVersion topVer,
        GridCacheContext cctx) throws IgniteCheckedException {
        GridAffinityProcessor aff = ctx.affinity();

        List<ClusterNode> res = null;

        if (!allowOverwrite())
            res = cctx.isLocal() ?
                aff.mapKeyToPrimaryAndBackups(cacheName, key, topVer) :
                cctx.topology().nodes(cctx.affinity().partition(key), topVer);
        else {
            ClusterNode node = aff.mapKeyToNode(cacheName, key, topVer);

            if (node != null)
                res = Collections.singletonList(node);
        }

        if (F.isEmpty(res))
            throw new ClusterTopologyServerNotFoundException("Failed to find server node for cache (all affinity " +
                "nodes have left the grid or cache was stopped): " + cacheName);

        return res;
    }

    /**
     * Performs flush.
     *
     * @throws IgniteCheckedException If failed.
     */
    private void doFlush() throws IgniteCheckedException {
        lastFlushTime = U.currentTimeMillis();

        List<IgniteInternalFuture> activeFuts0 = null;

        int doneCnt = 0;

        for (IgniteInternalFuture<?> f : activeFuts) {
            if (!f.isDone()) {
                if (activeFuts0 == null)
                    activeFuts0 = new ArrayList<>((int)(activeFuts.size() * 1.2));

                activeFuts0.add(f);
            }
            else {
                f.get();

                doneCnt++;
            }
        }

        if (activeFuts0 == null || activeFuts0.isEmpty())
            return;

        while (true) {
            Queue<IgniteInternalFuture<?>> q = null;

            for (Buffer buf : bufMappings.values()) {
                IgniteInternalFuture<?> flushFut = buf.flush();

                if (flushFut != null) {
                    if (q == null)
                        q = new ArrayDeque<>(bufMappings.size() * 2);

                    q.add(flushFut);
                }
            }

            if (q != null) {
                assert !q.isEmpty();

                boolean err = false;

                long startTimeMillis = U.currentTimeMillis();

                for (IgniteInternalFuture fut = q.poll(); fut != null; fut = q.poll()) {
                    try {
                        if (timeout == DFLT_UNLIMIT_TIMEOUT)
                            fut.get();
                        else {
                            long timeRemain = timeout - U.currentTimeMillis() + startTimeMillis;

                            if (timeRemain <= 0)
                                throw new IgniteDataStreamerTimeoutException("Data streamer exceeded timeout on flush.");

                            fut.get(timeRemain);
                        }
                    }
                    catch (IgniteClientDisconnectedCheckedException e) {
                        if (log.isDebugEnabled())
                            log.debug("Failed to flush buffer: " + e);

                        throw CU.convertToCacheException(e);
                    }
                    catch (IgniteFutureTimeoutCheckedException e) {
                        if (log.isDebugEnabled())
                            log.debug("Failed to flush buffer: " + e);

                        throw new IgniteDataStreamerTimeoutException("Data streamer exceeded timeout on flush.", e);
                    }
                    catch (IgniteCheckedException e) {
                        if (log.isDebugEnabled())
                            log.debug("Failed to flush buffer: " + e);

                        err = true;
                    }
                }

                if (err)
                    // Remaps needed - flush buffers.
                    continue;
            }

            doneCnt = 0;

            for (int i = 0; i < activeFuts0.size(); i++) {
                IgniteInternalFuture f = activeFuts0.get(i);

                if (f == null)
                    doneCnt++;
                else if (f.isDone()) {
                    f.get();

                    doneCnt++;

                    activeFuts0.set(i, null);
                }
                else
                    break;
            }

            if (doneCnt == activeFuts0.size())
                return;
        }
    }

    /** {@inheritDoc} */
    @SuppressWarnings("ForLoopReplaceableByForEach")
    @Override public void flush() throws CacheException {
        enterBusy();

        try {
            doFlush();
        }
        catch (IgniteCheckedException e) {
            throw CU.convertToCacheException(e);
        }
        finally {
            leaveBusy();
        }
    }

    /**
     * Flushes every internal buffer if buffer was flushed before passed in
     * threshold.
     * <p>
     * Does not wait for result and does not fail on errors assuming that this method
     * should be called periodically.
     */
    @Override public void tryFlush() throws IgniteInterruptedException {
        if (!busyLock.enterBusy())
            return;

        try {
            for (Buffer buf : bufMappings.values())
                buf.flush();

            lastFlushTime = U.currentTimeMillis();
        }
        catch (IgniteInterruptedCheckedException e) {
            throw GridCacheUtils.convertToCacheException(e);
        }
        finally {
            leaveBusy();
        }
    }

    /**
     * @param cancel {@code True} to close with cancellation.
     * @throws CacheException If failed.
     */
    @Override public void close(boolean cancel) throws CacheException {
        try {
            closeEx(cancel);
        }
        catch (IgniteCheckedException e) {
            throw CU.convertToCacheException(e);
        }
    }

    /**
     * @param cancel {@code True} to close with cancellation.
     * @throws IgniteCheckedException If failed.
     */
    public void closeEx(boolean cancel) throws IgniteCheckedException {
        IgniteCheckedException err = closeEx(cancel, null);

        if (err != null)
            throw err; // Throws at close().
    }

    /**
     * @param cancel {@code True} to close with cancellation.
     * @param err Error.
     * @throws IgniteCheckedException If failed.
     */
    private IgniteCheckedException closeEx(boolean cancel, IgniteCheckedException err) throws IgniteCheckedException {
        if (!closed.compareAndSet(false, true))
            return null;

        busyLock.block();

        if (log.isDebugEnabled())
            log.debug("Closing data streamer [ldr=" + this + ", cancel=" + cancel + ']');

        try {
            // Assuming that no methods are called on this loader after this method is called.
            if (cancel) {
                cancelled = true;

                for (Buffer buf : bufMappings.values())
                    buf.cancelAll(err);
            }
            else
                doFlush();

            ctx.event().removeLocalEventListener(discoLsnr);

            ctx.io().removeMessageListener(topic);
        }
        catch (IgniteCheckedException | IgniteDataStreamerTimeoutException e) {
            fut.onDone(e);
            throw e;
        }

        long failed = failCntr.longValue();

        if (failed > 0 && err == null)
            err = new IgniteCheckedException("Some of DataStreamer operations failed [failedCount=" + failed + "]");

        fut.onDone(err);

        return err;
    }

    /**
     * @param reconnectFut Reconnect future.
     * @throws IgniteCheckedException If failed.
     */
    public void onDisconnected(IgniteFuture<?> reconnectFut) throws IgniteCheckedException {
        IgniteClientDisconnectedCheckedException err = new IgniteClientDisconnectedCheckedException(reconnectFut,
            "Data streamer has been closed, client node disconnected.");

        disconnectErr = (CacheException)CU.convertToCacheException(err);

        for (Buffer buf : bufMappings.values())
            buf.cancelAll(err);

        closeEx(true, err);
    }

    /**
     * @return {@code true} If the loader is closed.
     */
    boolean isClosed() {
        return fut.isDone();
    }

    /** {@inheritDoc} */
    @Override public void close() throws CacheException {
        close(false);
    }

    /**
     * @return Max remap count.
     */
    public int maxRemapCount() {
        return maxRemapCnt;
    }

    /**
     * @param maxRemapCnt New max remap count.
     */
    public void maxRemapCount(int maxRemapCnt) {
        this.maxRemapCnt = maxRemapCnt;
    }

    /** {@inheritDoc} */
    @Override public String toString() {
        return S.toString(DataStreamerImpl.class, this);
    }

    /** {@inheritDoc} */
    @Override public long getDelay(TimeUnit unit) {
        return unit.convert(nextFlushTime() - U.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }

    /**
     * @return Next flush time.
     */
    private long nextFlushTime() {
        return lastFlushTime + autoFlushFreq;
    }

    /** {@inheritDoc} */
    @Override public int compareTo(Delayed o) {
        return nextFlushTime() > ((DataStreamerImpl)o).nextFlushTime() ? 1 : -1;
    }

    /**
     * Check permissions for streaming.
     *
     * @param perm Security permission.
     * @throws org.apache.ignite.plugin.security.SecurityException If permissions are not enough for streaming.
     */
    private void checkSecurityPermission(SecurityPermission perm)
        throws org.apache.ignite.plugin.security.SecurityException {
        if (!ctx.security().enabled())
            return;

        ctx.security().authorize(cacheName, perm, null);
    }

    /**
     *
     */
    private class Buffer {
        /** Node. */
        private final ClusterNode node;

        /** Active futures. */
        private final Collection<IgniteInternalFuture<Object>> locFuts;

        /** Buffered entries. */
        private List<DataStreamerEntry> entries;

        /** */
        @GridToStringExclude
        private GridFutureAdapter<Object> curFut;

        /** Local node flag. */
        private final boolean isLocNode;

        /** ID generator. */
        private final AtomicLong idGen = new AtomicLong();

        /** Active futures. */
        private final ConcurrentMap<Long, GridFutureAdapter<Object>> reqs;

        /** */
        private final Semaphore sem;

        /** Batch topology. */
        private AffinityTopologyVersion batchTopVer;

        /** Closure to signal on task finish. */
        @GridToStringExclude
        private final IgniteInClosure<IgniteInternalFuture<Object>> signalC = new IgniteInClosure<IgniteInternalFuture<Object>>() {
            @Override public void apply(IgniteInternalFuture<Object> t) {
                signalTaskFinished(t);
            }
        };

        /**
         * @param node Node.
         */
        Buffer(ClusterNode node) {
            assert node != null;

            this.node = node;

            locFuts = new GridConcurrentHashSet<>();
            reqs = new ConcurrentHashMap8<>();

            // Cache local node flag.
            isLocNode = node.equals(ctx.discovery().localNode());

            entries = newEntries();
            curFut = new GridFutureAdapter<>();
            curFut.listen(signalC);

            sem = new Semaphore(parallelOps);
        }

        /**
         * @param remap Remapping flag.
         */
        private void renewBatch(boolean remap) {
            entries = newEntries();
            curFut = new GridFutureAdapter<>();

            batchTopVer = null;

            if (!remap)
                curFut.listen(signalC);
        }

        /**
         * @param newEntries Infos.
         * @param topVer Topology version.
         * @param lsnr Listener for the operation future.
         * @param remap Remapping flag.
         * @return Future for operation.
         * @throws IgniteInterruptedCheckedException If failed.
         */
        @Nullable GridFutureAdapter<?> update(Iterable<DataStreamerEntry> newEntries,
            AffinityTopologyVersion topVer,
            IgniteInClosure<IgniteInternalFuture<?>> lsnr,
            boolean remap) throws IgniteInterruptedCheckedException {
            List<DataStreamerEntry> entries0 = null;

            GridFutureAdapter<Object> curFut0;

            AffinityTopologyVersion curBatchTopVer;

            synchronized (this) {
                curFut0 = curFut;

                curFut0.listen(lsnr);

                if (batchTopVer == null)
                    batchTopVer = topVer;

                curBatchTopVer = batchTopVer;

                for (DataStreamerEntry entry : newEntries)
                    entries.add(entry);

                if (entries.size() >= bufSize) {
                    entries0 = entries;

                    renewBatch(remap);
                }
            }

            if (!allowOverwrite() && !topVer.equals(curBatchTopVer)) {
                renewBatch(remap);

                curFut0.onDone(null, new IgniteCheckedException("Topology changed during batch preparation." +
                    "[batchTopVer=" + curBatchTopVer + ", topVer=" + topVer + "]"));
            }
            else if (entries0 != null) {
                submit(entries0, curBatchTopVer, curFut0, remap);

                if (cancelled)
                    curFut0.onDone(new IgniteCheckedException("Data streamer has been cancelled: " +
                        DataStreamerImpl.this));
                else if (ctx.clientDisconnected())
                    curFut0.onDone(new IgniteClientDisconnectedCheckedException(ctx.cluster().clientReconnectFuture(),
                        "Client node disconnected."));
            }

            return curFut0;
        }

        /**
         * @return Fresh collection with some space for outgrowth.
         */
        private List<DataStreamerEntry> newEntries() {
            return new ArrayList<>((int)(bufSize * 1.2));
        }

        /**
         * @return Future if any submitted.
         * @throws IgniteInterruptedCheckedException If thread has been interrupted.
         */
        @Nullable IgniteInternalFuture<?> flush() throws IgniteInterruptedCheckedException {
            List<DataStreamerEntry> entries0 = null;
            GridFutureAdapter<Object> curFut0 = null;

            acquireRemapSemaphore();

            synchronized (this) {
                if (!entries.isEmpty()) {
                    entries0 = entries;
                    curFut0 = curFut;

                    entries = newEntries();
                    curFut = new GridFutureAdapter<>();
                    curFut.listen(signalC);
                }
            }

            if (entries0 != null)
                submit(entries0, batchTopVer, curFut0, false);

            // Create compound future for this flush.
            GridCompoundFuture<Object, Object> res = null;

            for (IgniteInternalFuture<Object> f : locFuts) {
                if (res == null)
                    res = new GridCompoundFuture<>();

                res.add(f);
            }

            for (IgniteInternalFuture<Object> f : reqs.values()) {
                if (res == null)
                    res = new GridCompoundFuture<>();

                res.add(f);
            }

            if (res != null)
                res.markInitialized();

            return res;
        }

        /**
         * Increments active tasks count.
         *
         * @throws IgniteInterruptedCheckedException If thread has been interrupted.
         */
        private void incrementActiveTasks() throws IgniteInterruptedCheckedException {
            if (timeout == DFLT_UNLIMIT_TIMEOUT)
                U.acquire(sem);
            else if (!U.tryAcquire(sem, timeout, TimeUnit.MILLISECONDS)) {
                if (log.isDebugEnabled())
                    log.debug("Failed to add parallel operation.");

                throw new IgniteDataStreamerTimeoutException("Data streamer exceeded timeout when starts parallel operation.");
            }
        }

        /**
         * @param f Future that finished.
         */
        private void signalTaskFinished(IgniteInternalFuture<Object> f) {
            assert f != null;

            sem.release();
        }

        /**
         * @param entries Entries.
         * @param reqTopVer Request topology version.
         * @param curFut Current future.
         */
        private void localUpdate(final Collection<DataStreamerEntry> entries,
            final AffinityTopologyVersion reqTopVer,
            final GridFutureAdapter<Object> curFut) {
            try {
                GridCacheContext cctx = ctx.cache().internalCache(cacheName).context();

                final boolean allowOverwrite = allowOverwrite();
                final boolean loc = cctx.isLocal();

                if (!loc && !allowOverwrite)
                    cctx.topology().readLock();

                try {
                    GridDhtTopologyFuture fut = loc ? null : cctx.topologyVersionFuture();

                    AffinityTopologyVersion topVer = loc ? reqTopVer : fut.topologyVersion();

                    if (!allowOverwrite && !topVer.equals(reqTopVer)) {
                        curFut.onDone(new IgniteCheckedException(
                            "DataStreamer will retry data transfer at stable topology. " +
                                "[reqTop=" + reqTopVer + " ,topVer=" + topVer + ", node=local]"));
                    }
                    else if (loc || allowOverwrite || fut.isDone()) {
                        IgniteInternalFuture<Object> callFut = ctx.closure().callLocalSafe(
                            new DataStreamerUpdateJob(
                                ctx,
                                log,
                                cacheName,
                                entries,
                                false,
                                skipStore,
                                keepBinary,
                                rcvr),
                            false);

                        locFuts.add(callFut);

                        final GridFutureAdapter waitFut = (loc || allowOverwrite) ?
                            null :
                            cctx.mvcc().addDataStreamerFuture(topVer);

                        callFut.listen(new IgniteInClosure<IgniteInternalFuture<Object>>() {
                            @Override public void apply(IgniteInternalFuture<Object> t) {
                                try {
                                    boolean rmv = locFuts.remove(t);

                                    assert rmv;

                                    curFut.onDone(t.get());
                                }
                                catch (IgniteCheckedException e) {
                                    curFut.onDone(e);
                                }
                                finally {
                                    if (waitFut != null)
                                        waitFut.onDone();
                                }
                            }
                        });
                    }
                    else {
                        fut.listen(new IgniteInClosure<IgniteInternalFuture<AffinityTopologyVersion>>() {
                            @Override public void apply(IgniteInternalFuture<AffinityTopologyVersion> e) {
                                localUpdate(entries, reqTopVer, curFut);
                            }
                        });
                    }
                }
                finally {
                    if (!loc && !allowOverwrite)
                        cctx.topology().readUnlock();
                }
            }
            catch (Throwable ex) {
                curFut.onDone(new IgniteCheckedException("DataStreamer data handling failed.", ex));
            }
        }

        /**
         * @param entries Entries to submit.
         * @param topVer Topology version.
         * @param curFut Current future.
         * @param remap Remapping flag.
         * @throws IgniteInterruptedCheckedException If interrupted.
         */
        private void submit(final Collection<DataStreamerEntry> entries,
            @Nullable AffinityTopologyVersion topVer,
            final GridFutureAdapter<Object> curFut,
            boolean remap)
            throws IgniteInterruptedCheckedException {
            assert entries != null;
            assert !entries.isEmpty();
            assert curFut != null;

            if (!remap) {
                try {
                    incrementActiveTasks();
                }
                catch (IgniteDataStreamerTimeoutException e) {
                    curFut.onDone(e);

                    throw e;
                }
            }

            IgniteInternalFuture<Object> fut;

            Byte plc = ioPlcRslvr.apply(node);

            if (plc == null)
                plc = PUBLIC_POOL;

            if (isLocNode && plc == GridIoPolicy.PUBLIC_POOL)
                localUpdate(entries, topVer, curFut);
            else {
                try {
                    for (DataStreamerEntry e : entries) {
                        e.getKey().prepareMarshal(cacheObjCtx);

                        CacheObject val = e.getValue();

                        if (val != null)
                            val.prepareMarshal(cacheObjCtx);
                    }

                    if (updaterBytes == null) {
                        assert rcvr != null;

                        updaterBytes = U.marshal(ctx, rcvr);
                    }

                    if (topicBytes == null)
                        topicBytes = U.marshal(ctx, topic);
                }
                catch (IgniteCheckedException e) {
                    U.error(log, "Failed to marshal (request will not be sent).", e);

                    return;
                }

                GridDeployment dep = null;
                GridPeerDeployAware jobPda0 = null;

                jobPda0 = jobPda;

                if (ctx.deploy().enabled() && jobPda0 != null) {
                    try {
                        dep = ctx.deploy().deploy(jobPda0.deployClass(), jobPda0.classLoader());

                        GridCacheAdapter<Object, Object> cache = ctx.cache().internalCache(cacheName);

                        if (cache != null)
                            cache.context().deploy().onEnter();
                    }
                    catch (IgniteCheckedException e) {
                        U.error(log, "Failed to deploy class (request will not be sent): " + jobPda0.deployClass(), e);

                        return;
                    }

                    if (dep == null)
                        U.warn(log, "Failed to deploy class (request will be sent): " + jobPda0.deployClass());
                }

                long reqId = idGen.incrementAndGet();

                fut = curFut;

                reqs.put(reqId, (GridFutureAdapter<Object>)fut);

                if (topVer == null)
                    topVer = ctx.cache().context().exchange().readyAffinityVersion();

                DataStreamerRequest req = new DataStreamerRequest(
                    reqId,
                    topicBytes,
                    cacheName,
                    updaterBytes,
                    entries,
                    true,
                    skipStore,
                    keepBinary,
                    dep != null ? dep.deployMode() : null,
                    dep != null ? jobPda0.deployClass().getName() : null,
                    dep != null ? dep.userVersion() : null,
                    dep != null ? dep.participants() : null,
                    dep != null ? dep.classLoaderId() : null,
                    dep == null,
                    topVer);

                try {
                    ctx.io().send(node, TOPIC_DATASTREAM, req, plc);

                    if (log.isDebugEnabled())
                        log.debug("Sent request to node [nodeId=" + node.id() + ", req=" + req + ']');
                }
                catch (IgniteCheckedException e) {
                    GridFutureAdapter<Object> fut0 = ((GridFutureAdapter<Object>)fut);

                    try {
                        if (ctx.discovery().alive(node) && ctx.discovery().pingNode(node.id()))
                            fut0.onDone(e);
                        else
                            fut0.onDone(new ClusterTopologyCheckedException("Failed to send request (node has left): "
                                + node.id()));
                    }
                    catch (IgniteClientDisconnectedCheckedException e0) {
                        fut0.onDone(e0);
                    }
                }
            }
        }

        /**
         *
         */
        void onNodeLeft() {
            assert !isLocNode;
            assert bufMappings.get(node.id()) != this;

            if (log.isDebugEnabled())
                log.debug("Forcibly completing futures (node has left): " + node.id());

            Exception e = new ClusterTopologyCheckedException("Failed to wait for request completion " +
                "(node has left): " + node.id());

            for (GridFutureAdapter<Object> f : reqs.values())
                f.onDone(e);

            // Make sure to complete current future.
            GridFutureAdapter<Object> curFut0;

            synchronized (this) {
                curFut0 = curFut;
            }

            curFut0.onDone(e);
        }

        /**
         * @param res Response.
         * @param nodeId Node id.
         */
        void onResponse(DataStreamerResponse res, UUID nodeId) {
            if (log.isDebugEnabled())
                log.debug("Received data load response: " + res);

            GridFutureAdapter<?> f = reqs.remove(res.requestId());

            if (f == null) {
                if (log.isDebugEnabled())
                    log.debug("Future for request has not been found: " + res.requestId());

                return;
            }

            Throwable err = null;

            byte[] errBytes = res.errorBytes();

            if (errBytes != null) {
                try {
                    GridPeerDeployAware jobPda0 = jobPda;

                    err = new IgniteCheckedException("DataStreamer request failed [node=" + nodeId + "]",
                        (Throwable)U.unmarshal(ctx,
                            errBytes,
                            U.resolveClassLoader(jobPda0 != null ? jobPda0.classLoader() : null, ctx.config())));
                }
                catch (IgniteCheckedException e) {
                    f.onDone(null, new IgniteCheckedException("Failed to unmarshal response.", e));

                    return;
                }
            }

            f.onDone(null, err);

            if (log.isDebugEnabled())
                log.debug("Finished future [fut=" + f + ", reqId=" + res.requestId() + ", err=" + err + ']');
        }

        /**
         * @param err Error.
         */
        void cancelAll(@Nullable IgniteCheckedException err) {
            if (err == null)
                err = new IgniteCheckedException("Data streamer has been cancelled: " + DataStreamerImpl.this);

            for (IgniteInternalFuture<?> f : locFuts) {
                try {
                    f.cancel();
                }
                catch (IgniteCheckedException e) {
                    U.error(log, "Failed to cancel mini-future.", e);
                }
            }

            for (GridFutureAdapter<?> f : reqs.values())
                f.onDone(err);
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            int size;

            synchronized (this) {
                size = entries.size();
            }

            return S.toString(Buffer.class, this,
                "entriesCnt", size,
                "locFutsSize", locFuts.size(),
                "reqsSize", reqs.size());
        }
    }

    /**
     * Data streamer peer-deploy aware.
     */
    private class DataStreamerPda implements GridPeerDeployAware {
        /** */
        private static final long serialVersionUID = 0L;

        /** Deploy class. */
        private Class<?> cls;

        /** Class loader. */
        private ClassLoader ldr;

        /** Collection of objects to detect deploy class and class loader. */
        private Collection<Object> objs;

        /**
         * Constructs data streamer peer-deploy aware.
         *
         * @param objs Collection of objects to detect deploy class and class loader.
         */
        private DataStreamerPda(Object... objs) {
            this.objs = Arrays.asList(objs);
        }

        /** {@inheritDoc} */
        @Override public Class<?> deployClass() {
            if (cls == null) {
                Class<?> cls0 = null;

                if (depCls != null)
                    cls0 = depCls;
                else {
                    for (Iterator<Object> it = objs.iterator(); (cls0 == null || U.isJdk(cls0)) && it.hasNext(); ) {
                        Object o = it.next();

                        if (o != null)
                            cls0 = U.detectClass(o);
                    }

                    if (cls0 == null || U.isJdk(cls0))
                        cls0 = DataStreamerImpl.class;
                }

                assert cls0 != null : "Failed to detect deploy class [objs=" + objs + ']';

                cls = cls0;
            }

            return cls;
        }

        /** {@inheritDoc} */
        @Override public ClassLoader classLoader() {
            if (ldr == null) {
                ClassLoader ldr0 = deployClass().getClassLoader();

                // Safety.
                if (ldr0 == null)
                    ldr0 = U.gridClassLoader();

                assert ldr0 != null : "Failed to detect classloader [objs=" + objs + ']';

                ldr = ldr0;
            }

            return ldr;
        }
    }

    /**
     * Isolated receiver which only loads entry initial value.
     */
    protected static class IsolatedUpdater implements StreamReceiver<KeyCacheObject, CacheObject>,
        DataStreamerCacheUpdaters.InternalUpdater {
        /** */
        private static final long serialVersionUID = 0L;

        /** {@inheritDoc} */
        @Override public void receive(IgniteCache<KeyCacheObject, CacheObject> cache,
            Collection<Map.Entry<KeyCacheObject, CacheObject>> entries) {
            IgniteCacheProxy<KeyCacheObject, CacheObject> proxy = (IgniteCacheProxy<KeyCacheObject, CacheObject>)cache;

            GridCacheAdapter<KeyCacheObject, CacheObject> internalCache = proxy.context().cache();

            if (internalCache.isNear())
                internalCache = internalCache.context().near().dht();

            GridCacheContext cctx = internalCache.context();

            AffinityTopologyVersion topVer = cctx.isLocal() ?
                cctx.affinity().affinityTopologyVersion() :
                cctx.shared().exchange().readyAffinityVersion();

            GridCacheVersion ver = cctx.versions().isolatedStreamerVersion();

            long ttl = CU.TTL_ETERNAL;
            long expiryTime = CU.EXPIRE_TIME_ETERNAL;

            ExpiryPolicy plc = cctx.expiry();

            for (Entry<KeyCacheObject, CacheObject> e : entries) {
                try {
                    e.getKey().finishUnmarshal(cctx.cacheObjectContext(), cctx.deploy().globalLoader());

                    GridCacheEntryEx entry = internalCache.entryEx(e.getKey(), topVer);

                    if (plc != null) {
                        ttl = CU.toTtl(plc.getExpiryForCreation());

                        if (ttl == CU.TTL_ZERO)
                            continue;
                        else if (ttl == CU.TTL_NOT_CHANGED)
                            ttl = 0;

                        expiryTime = CU.toExpireTime(ttl);
                    }

                    boolean primary = cctx.affinity().primary(cctx.localNode(), entry.key(), topVer);

                    entry.initialValue(e.getValue(),
                        ver,
                        ttl,
                        expiryTime,
                        false,
                        topVer,
                        primary ? GridDrType.DR_LOAD : GridDrType.DR_PRELOAD,
                        false);

                    cctx.evicts().touch(entry, topVer);

                    CU.unwindEvicts(cctx);

                    entry.onUnlock();
                }
                catch (GridDhtInvalidPartitionException | GridCacheEntryRemovedException ignored) {
                    // No-op.
                }
                catch (IgniteCheckedException ex) {
                    IgniteLogger log = cache.unwrap(Ignite.class).log();

                    U.error(log, "Failed to set initial value for cache entry: " + e, ex);
                }
            }
        }
    }

    /**
     * Default IO policy resolver.
     */
    private static class DefaultIoPolicyResolver implements IgniteClosure<ClusterNode, Byte> {
        /** */
        private static final long serialVersionUID = 0L;

        /** {@inheritDoc} */
        @Override public Byte apply(ClusterNode gridNode) {
            return PUBLIC_POOL;
        }
    }

    /**
     * Key object wrapper. Using identity equals prevents slow down in case of hash code collision.
     */
    private static class KeyCacheObjectWrapper {
        /** key object */
        private final KeyCacheObject key;

        /**
         * Constructor
         *
         * @param key key object
         */
        KeyCacheObjectWrapper(KeyCacheObject key) {
            assert key != null;

            this.key = key;
        }

        /** {@inheritDoc} */
        @Override public boolean equals(Object o) {
            return o instanceof KeyCacheObjectWrapper && this.key == ((KeyCacheObjectWrapper)o).key;
        }

        /** {@inheritDoc} */
        @Override public int hashCode() {
            return key.hashCode();
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(KeyCacheObjectWrapper.class, this);
        }
    }
}