ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [36/40] incubator-ignite git commit: renaming wip
Date Thu, 12 Mar 2015 13:08:29 GMT
renaming wip


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/60a1a481
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/60a1a481
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/60a1a481

Branch: refs/heads/sprint-2
Commit: 60a1a481781c0a8c6533489c83adac82ec2ed4f6
Parents: b2c679e
Author: Yakov Zhdanov <yzhdanov@gridgain.com>
Authored: Wed Mar 11 17:50:32 2015 +0300
Committer: Yakov Zhdanov <yzhdanov@gridgain.com>
Committed: Wed Mar 11 17:50:32 2015 +0300

----------------------------------------------------------------------
 .../ignite/codegen/MessageCodeGenerator.java    |    2 +-
 .../ignite/internal/GridKernalContext.java      |    2 +-
 .../ignite/internal/GridKernalContextImpl.java  |    2 +-
 .../apache/ignite/internal/IgniteKernal.java    |    2 +-
 .../communication/GridIoMessageFactory.java     |    2 +-
 .../processors/cache/GridCacheAdapter.java      |    2 +-
 .../GridDistributedCacheAdapter.java            |    2 +-
 .../version/GridCacheRawVersionedEntry.java     |    2 +-
 .../datastream/DataStreamerCacheUpdaters.java   |  206 ---
 .../datastream/DataStreamerEntry.java           |  170 ---
 .../datastream/DataStreamerFuture.java          |   60 -
 .../processors/datastream/DataStreamerImpl.java | 1405 ------------------
 .../datastream/DataStreamerProcessor.java       |  307 ----
 .../datastream/DataStreamerRequest.java         |  451 ------
 .../datastream/DataStreamerResponse.java        |  166 ---
 .../datastream/DataStreamerUpdateJob.java       |  150 --
 .../internal/processors/datastream/package.html |   24 -
 .../datastreamer/DataStreamerCacheUpdaters.java |  206 +++
 .../datastreamer/DataStreamerEntry.java         |  170 +++
 .../datastreamer/DataStreamerFuture.java        |   60 +
 .../datastreamer/DataStreamerImpl.java          | 1405 ++++++++++++++++++
 .../datastreamer/DataStreamerProcessor.java     |  307 ++++
 .../datastreamer/DataStreamerRequest.java       |  451 ++++++
 .../datastreamer/DataStreamerResponse.java      |  166 +++
 .../datastreamer/DataStreamerUpdateJob.java     |  150 ++
 .../processors/datastreamer/package.html        |   24 +
 .../dr/IgniteDrDataStreamerCacheUpdater.java    |    2 +-
 .../processors/igfs/IgfsDataManager.java        |    2 +-
 .../datastream/DataStreamerImplSelfTest.java    |  205 ---
 .../DataStreamerProcessorSelfTest.java          |  970 ------------
 .../IgniteDataStreamerPerformanceTest.java      |  197 ---
 .../datastreamer/DataStreamerImplSelfTest.java  |  205 +++
 .../DataStreamerProcessorSelfTest.java          |  970 ++++++++++++
 .../IgniteDataStreamerPerformanceTest.java      |  197 +++
 .../ignite/testsuites/IgniteCacheTestSuite.java |    2 +-
 35 files changed, 4322 insertions(+), 4322 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
----------------------------------------------------------------------
diff --git a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
index 07200f5..5037b8a 100644
--- a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
+++ b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
@@ -18,7 +18,7 @@
 package org.apache.ignite.codegen;
 
 import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.processors.datastream.*;
+import org.apache.ignite.internal.processors.datastreamer.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 import org.apache.ignite.lang.*;
 import org.apache.ignite.plugin.extensions.communication.*;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
index 8b6eaa9..f780676 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
@@ -36,7 +36,7 @@ import org.apache.ignite.internal.processors.clock.*;
 import org.apache.ignite.internal.processors.closure.*;
 import org.apache.ignite.internal.processors.cluster.*;
 import org.apache.ignite.internal.processors.continuous.*;
-import org.apache.ignite.internal.processors.datastream.*;
+import org.apache.ignite.internal.processors.datastreamer.*;
 import org.apache.ignite.internal.processors.datastructures.*;
 import org.apache.ignite.internal.processors.hadoop.*;
 import org.apache.ignite.internal.processors.igfs.*;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
index c066c94..362a472 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
@@ -38,7 +38,7 @@ import org.apache.ignite.internal.processors.clock.*;
 import org.apache.ignite.internal.processors.closure.*;
 import org.apache.ignite.internal.processors.cluster.*;
 import org.apache.ignite.internal.processors.continuous.*;
-import org.apache.ignite.internal.processors.datastream.*;
+import org.apache.ignite.internal.processors.datastreamer.*;
 import org.apache.ignite.internal.processors.datastructures.*;
 import org.apache.ignite.internal.processors.hadoop.*;
 import org.apache.ignite.internal.processors.igfs.*;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index 92e4ced..c0b5225 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -42,7 +42,7 @@ import org.apache.ignite.internal.processors.clock.*;
 import org.apache.ignite.internal.processors.closure.*;
 import org.apache.ignite.internal.processors.cluster.*;
 import org.apache.ignite.internal.processors.continuous.*;
-import org.apache.ignite.internal.processors.datastream.*;
+import org.apache.ignite.internal.processors.datastreamer.*;
 import org.apache.ignite.internal.processors.datastructures.*;
 import org.apache.ignite.internal.processors.hadoop.*;
 import org.apache.ignite.internal.processors.job.*;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
index 7be89cf..762507a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
@@ -34,7 +34,7 @@ import org.apache.ignite.internal.processors.cache.transactions.*;
 import org.apache.ignite.internal.processors.cache.version.*;
 import org.apache.ignite.internal.processors.clock.*;
 import org.apache.ignite.internal.processors.continuous.*;
-import org.apache.ignite.internal.processors.datastream.*;
+import org.apache.ignite.internal.processors.datastreamer.*;
 import org.apache.ignite.internal.processors.igfs.*;
 import org.apache.ignite.internal.processors.rest.handlers.task.*;
 import org.apache.ignite.internal.processors.streamer.*;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index ff5e58c..2e16475 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -32,7 +32,7 @@ import org.apache.ignite.internal.processors.cache.dr.*;
 import org.apache.ignite.internal.processors.cache.query.*;
 import org.apache.ignite.internal.processors.cache.transactions.*;
 import org.apache.ignite.internal.processors.cache.version.*;
-import org.apache.ignite.internal.processors.datastream.*;
+import org.apache.ignite.internal.processors.datastreamer.*;
 import org.apache.ignite.internal.processors.dr.*;
 import org.apache.ignite.internal.processors.task.*;
 import org.apache.ignite.internal.transactions.*;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
index 3d85590..792fc47 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
@@ -26,7 +26,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.*;
 import org.apache.ignite.internal.processors.cache.distributed.near.*;
 import org.apache.ignite.internal.processors.cache.transactions.*;
 import org.apache.ignite.internal.processors.cache.version.*;
-import org.apache.ignite.internal.processors.datastream.*;
+import org.apache.ignite.internal.processors.datastreamer.*;
 import org.apache.ignite.internal.processors.task.*;
 import org.apache.ignite.internal.util.future.*;
 import org.apache.ignite.internal.util.typedef.internal.*;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheRawVersionedEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheRawVersionedEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheRawVersionedEntry.java
index 9e73e1b..986515c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheRawVersionedEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheRawVersionedEntry.java
@@ -20,7 +20,7 @@ package org.apache.ignite.internal.processors.cache.version;
 import org.apache.ignite.*;
 import org.apache.ignite.internal.*;
 import org.apache.ignite.internal.processors.cache.*;
-import org.apache.ignite.internal.processors.datastream.*;
+import org.apache.ignite.internal.processors.datastreamer.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 import org.apache.ignite.marshaller.*;
 import org.apache.ignite.plugin.extensions.communication.*;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerCacheUpdaters.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerCacheUpdaters.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerCacheUpdaters.java
deleted file mode 100644
index 64f0851..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerCacheUpdaters.java
+++ /dev/null
@@ -1,206 +0,0 @@
-/*
- * 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.datastream;
-
-import org.apache.ignite.*;
-import org.apache.ignite.internal.processors.cache.*;
-import org.apache.ignite.internal.util.typedef.*;
-import org.jetbrains.annotations.*;
-
-import java.util.*;
-
-/**
- * Bundled factory for cache updaters.
- */
-public class DataStreamerCacheUpdaters {
-    /** */
-    private static final IgniteDataStreamer.Updater INDIVIDUAL = new Individual();
-
-    /** */
-    private static final IgniteDataStreamer.Updater BATCHED = new Batched();
-
-    /** */
-    private static final IgniteDataStreamer.Updater BATCHED_SORTED = new BatchedSorted();
-
-    /**
-     * Updates cache using independent {@link org.apache.ignite.cache.GridCache#put(Object, Object, org.apache.ignite.lang.IgnitePredicate[])} and
-     * {@link org.apache.ignite.cache.GridCache#remove(Object, org.apache.ignite.lang.IgnitePredicate[])} operations. Thus it is safe from deadlocks but performance
-     * is not the best.
-     *
-     * @return Single updater.
-     */
-    public static <K, V> IgniteDataStreamer.Updater<K, V> individual() {
-        return INDIVIDUAL;
-    }
-
-    /**
-     * Updates cache using batched methods {@link org.apache.ignite.cache.GridCache#putAll(Map, org.apache.ignite.lang.IgnitePredicate[])} and
-     * {@link org.apache.ignite.cache.GridCache#removeAll(Collection, org.apache.ignite.lang.IgnitePredicate[])}. Can cause deadlocks if the same keys are getting
-     * updated concurrently. Performance is generally better than in {@link #individual()}.
-     *
-     * @return Batched updater.
-     */
-    public static <K, V> IgniteDataStreamer.Updater<K, V> batched() {
-        return BATCHED;
-    }
-
-    /**
-     * Updates cache using batched methods {@link org.apache.ignite.cache.GridCache#putAll(Map, org.apache.ignite.lang.IgnitePredicate[])} and
-     * {@link org.apache.ignite.cache.GridCache#removeAll(Collection, org.apache.ignite.lang.IgnitePredicate[])}. Keys are sorted in natural order and if all updates
-     * use the same rule deadlock can not happen. Performance is generally better than in {@link #individual()}.
-     *
-     * @return Batched sorted updater.
-     */
-    public static <K extends Comparable<?>, V> IgniteDataStreamer.Updater<K, V> batchedSorted() {
-        return BATCHED_SORTED;
-    }
-
-    /**
-     * Updates cache.
-     *
-     * @param cache Cache.
-     * @param rmvCol Keys to remove.
-     * @param putMap Entries to put.
-     * @throws IgniteException If failed.
-     */
-    protected static <K, V> void updateAll(IgniteCache<K, V> cache, @Nullable Set<K> rmvCol,
-        Map<K, V> putMap) {
-        assert rmvCol != null || putMap != null;
-
-        // Here we assume that there are no key duplicates, so the following calls are valid.
-        if (rmvCol != null)
-            ((IgniteCacheProxy<K, V>)cache).removeAll(rmvCol);
-
-        if (putMap != null)
-            cache.putAll(putMap);
-    }
-
-    /**
-     * Simple cache updater implementation. Updates keys one by one thus is not dead lock prone.
-     */
-    private static class Individual<K, V> implements IgniteDataStreamer.Updater<K, V>, InternalUpdater {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** {@inheritDoc} */
-        @Override public void update(IgniteCache<K, V> cache, Collection<Map.Entry<K, V>> entries) {
-            assert cache != null;
-            assert !F.isEmpty(entries);
-
-            for (Map.Entry<K, V> entry : entries) {
-                K key = entry.getKey();
-
-                assert key != null;
-
-                V val = entry.getValue();
-
-                if (val == null)
-                    cache.remove(key);
-                else
-                    cache.put(key, val);
-            }
-        }
-    }
-
-    /**
-     * Batched updater. Updates cache using batch operations thus is dead lock prone.
-     */
-    private static class Batched<K, V> implements IgniteDataStreamer.Updater<K, V>, InternalUpdater {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** {@inheritDoc} */
-        @Override public void update(IgniteCache<K, V> cache, Collection<Map.Entry<K, V>> entries) {
-            assert cache != null;
-            assert !F.isEmpty(entries);
-
-            Map<K, V> putAll = null;
-            Set<K> rmvAll = null;
-
-            for (Map.Entry<K, V> entry : entries) {
-                K key = entry.getKey();
-
-                assert key != null;
-
-                V val = entry.getValue();
-
-                if (val == null) {
-                    if (rmvAll == null)
-                        rmvAll = new HashSet<>();
-
-                    rmvAll.add(key);
-                }
-                else {
-                    if (putAll == null)
-                        putAll = new HashMap<>();
-
-                    putAll.put(key, val);
-                }
-            }
-
-            updateAll(cache, rmvAll, putAll);
-        }
-    }
-
-    /**
-     * Batched updater. Updates cache using batch operations thus is dead lock prone.
-     */
-    private static class BatchedSorted<K, V> implements IgniteDataStreamer.Updater<K, V>, InternalUpdater {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** {@inheritDoc} */
-        @Override public void update(IgniteCache<K, V> cache, Collection<Map.Entry<K, V>> entries) {
-            assert cache != null;
-            assert !F.isEmpty(entries);
-
-            Map<K, V> putAll = null;
-            Set<K> rmvAll = null;
-
-            for (Map.Entry<K, V> entry : entries) {
-                K key = entry.getKey();
-
-                assert key instanceof Comparable;
-
-                V val = entry.getValue();
-
-                if (val == null) {
-                    if (rmvAll == null)
-                        rmvAll = new TreeSet<>();
-
-                    rmvAll.add(key);
-                }
-                else {
-                    if (putAll == null)
-                        putAll = new TreeMap<>();
-
-                    putAll.put(key, val);
-                }
-            }
-
-            updateAll(cache, rmvAll, putAll);
-        }
-    }
-
-    /**
-     * Marker interface for updaters which do not need to unwrap cache objects.
-     */
-    public static interface InternalUpdater {
-        // No-op.
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerEntry.java
deleted file mode 100644
index 2aa11e5..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerEntry.java
+++ /dev/null
@@ -1,170 +0,0 @@
-/*
- * 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.datastream;
-
-import org.apache.ignite.internal.processors.cache.*;
-import org.apache.ignite.internal.util.tostring.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.plugin.extensions.communication.*;
-
-import java.nio.*;
-import java.util.*;
-
-/**
- *
- */
-public class DataStreamerEntry implements Map.Entry<KeyCacheObject, CacheObject>, Message {
-    /** */
-    private static final long serialVersionUID = 0L;
-
-    /** */
-    @GridToStringInclude
-    protected KeyCacheObject key;
-
-    /** */
-    @GridToStringInclude
-    protected CacheObject val;
-
-    /**
-     *
-     */
-    public DataStreamerEntry() {
-        // No-op.
-    }
-
-    /**
-     * @param key Key.
-     * @param val Value.
-     */
-    public DataStreamerEntry(KeyCacheObject key, CacheObject val) {
-        this.key = key;
-        this.val = val;
-    }
-
-    /** {@inheritDoc} */
-    @Override public KeyCacheObject getKey() {
-        return key;
-    }
-
-    /** {@inheritDoc} */
-    @Override public CacheObject getValue() {
-        return val;
-    }
-
-    /** {@inheritDoc} */
-    @Override public CacheObject setValue(CacheObject val) {
-        CacheObject old = this.val;
-
-        this.val = val;
-
-        return old;
-    }
-
-    /**
-     * @param ctx Cache context.
-     * @return Map entry unwrapping internal key and value.
-     */
-    public <K, V> Map.Entry<K, V> toEntry(final GridCacheContext ctx) {
-        return new Map.Entry<K, V>() {
-            @Override public K getKey() {
-                return key.value(ctx.cacheObjectContext(), false);
-            }
-
-            @Override public V setValue(V val) {
-                throw new UnsupportedOperationException();
-            }
-
-            @Override public V getValue() {
-                return val != null ? val.<V>value(ctx.cacheObjectContext(), false) : null;
-            }
-        };
-    }
-
-    /** {@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.writeMessage("key", key))
-                    return false;
-
-                writer.incrementState();
-
-            case 1:
-                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:
-                key = reader.readMessage("key");
-
-                if (!reader.isLastRead())
-                    return false;
-
-                reader.incrementState();
-
-            case 1:
-                val = reader.readMessage("val");
-
-                if (!reader.isLastRead())
-                    return false;
-
-                reader.incrementState();
-
-        }
-
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override public byte directType() {
-        return 95;
-    }
-
-    /** {@inheritDoc} */
-    @Override public byte fieldsCount() {
-        return 2;
-    }
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(DataStreamerEntry.class, this);
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerFuture.java
deleted file mode 100644
index d16887d..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerFuture.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.datastream;
-
-import org.apache.ignite.*;
-import org.apache.ignite.internal.util.future.*;
-import org.apache.ignite.internal.util.tostring.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-
-/**
- * Data streamer future.
- */
-class DataStreamerFuture extends GridFutureAdapter<Object> {
-    /** */
-    private static final long serialVersionUID = 0L;
-
-    /** Data loader. */
-    @GridToStringExclude
-    private DataStreamerImpl dataLdr;
-
-    /**
-     * @param dataLdr Data streamer.
-     */
-    DataStreamerFuture(DataStreamerImpl dataLdr) {
-        assert dataLdr != null;
-
-        this.dataLdr = dataLdr;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean cancel() throws IgniteCheckedException {
-        if (onCancelled()) {
-            dataLdr.closeEx(true);
-
-            return true;
-        }
-
-        return false;
-    }
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(DataStreamerFuture.class, this, super.toString());
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerImpl.java
deleted file mode 100644
index 691c74f..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerImpl.java
+++ /dev/null
@@ -1,1405 +0,0 @@
-/*
- * 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.datastream;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cluster.*;
-import org.apache.ignite.events.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.cluster.*;
-import org.apache.ignite.internal.managers.communication.*;
-import org.apache.ignite.internal.managers.deployment.*;
-import org.apache.ignite.internal.managers.eventstorage.*;
-import org.apache.ignite.internal.processors.affinity.*;
-import org.apache.ignite.internal.processors.cache.*;
-import org.apache.ignite.internal.processors.cache.distributed.dht.*;
-import org.apache.ignite.internal.processors.cache.version.*;
-import org.apache.ignite.internal.processors.cacheobject.*;
-import org.apache.ignite.internal.processors.dr.*;
-import org.apache.ignite.internal.util.*;
-import org.apache.ignite.internal.util.future.*;
-import org.apache.ignite.internal.util.lang.*;
-import org.apache.ignite.internal.util.tostring.*;
-import org.apache.ignite.internal.util.typedef.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.lang.*;
-import org.jdk8.backport.*;
-import org.jetbrains.annotations.*;
-
-import java.util.*;
-import java.util.Map.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.*;
-
-import static org.apache.ignite.events.EventType.*;
-import static org.apache.ignite.internal.GridTopic.*;
-import static org.apache.ignite.internal.managers.communication.GridIoPolicy.*;
-
-/**
- * Data streamer implementation.
- */
-@SuppressWarnings("unchecked")
-public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed {
-    /** Isolated updater. */
-    private static final Updater ISOLATED_UPDATER = new IsolatedUpdater();
-
-    /** Cache updater. */
-    private Updater<K, V> updater = ISOLATED_UPDATER;
-
-    /** */
-    private byte[] updaterBytes;
-
-    /** 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 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;
-
-    /** 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;
-        }
-    };
-
-    /** 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();
-
-    /** 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 int maxRemapCnt = DFLT_MAX_REMAP_CNT;
-
-    /** Whether a warning at {@link DataStreamerImpl#allowOverwrite()} printed */
-    private static boolean isWarningPrinted;
-
-    /**
-     * @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);
-
-        ClusterNode node = F.first(ctx.grid().cluster().forCacheNodes(cacheName).nodes());
-
-        if (node == null)
-            throw new IllegalStateException("Cache doesn't exist: " + cacheName);
-
-        this.cacheObjCtx = ctx.cacheObjects().contextForCache(node, cacheName);
-        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);
-
-                if (buf != null) {
-                    // Only async notification is possible since
-                    // discovery thread may be trapped otherwise.
-                    ctx.closure().callLocalSafe(
-                        new Callable<Object>() {
-                            @Override public Object call() throws Exception {
-                                buf.onNodeLeft();
-
-                                return null;
-                            }
-                        },
-                        true /* system pool */
-                    );
-                }
-            }
-        };
-
-        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);
-
-                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 IgniteFutureImpl<>(fut);
-    }
-
-    /**
-     * @return Cache object context.
-     */
-    public CacheObjectContext cacheObjectContext() {
-        return cacheObjCtx;
-    }
-
-    /**
-     * Enters busy lock.
-     */
-    private void enterBusy() {
-        if (!busyLock.enterBusy())
-            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 updater(Updater<K, V> updater) {
-        A.notNull(updater, "updater");
-
-        this.updater = updater;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean allowOverwrite() {
-        boolean allow = updater != ISOLATED_UPDATER;
-
-        if (!allow && !isWarningPrinted) {
-            synchronized (this) {
-                if (!isWarningPrinted) {
-                    log.warning("Data streamer will not overwrite existing cache entries for better performance " +
-                        "(to change, set allowOverwrite to true)");
-
-                    isWarningPrinted = true;
-                }
-            }
-        }
-
-        return allow;
-    }
-
-    /** {@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 IgniteException("Failed to get node for cache: " + cacheName);
-
-        updater = 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 @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 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");
-
-        enterBusy();
-
-        try {
-            GridFutureAdapter<Object> resFut = new GridFutureAdapter<>();
-
-            resFut.listen(rmvActiveFut);
-
-            activeFuts.add(resFut);
-
-            Collection<KeyCacheObject> keys = null;
-
-            if (entries.size() > 1) {
-                keys = new GridConcurrentHashSet<>(entries.size(), U.capacity(entries.size()), 1);
-
-                for (Map.Entry<K, V> entry : entries)
-                    keys.add(cacheObjProc.toCacheKeyObject(cacheObjCtx, entry.getKey(), true));
-            }
-
-            Collection<? extends DataStreamerEntry> entries0 = F.viewReadOnly(entries, new C1<Entry<K, V>, DataStreamerEntry>() {
-                @Override public DataStreamerEntry apply(Entry<K, V> e) {
-                    KeyCacheObject key = cacheObjProc.toCacheKeyObject(cacheObjCtx, e.getKey(), true);
-                    CacheObject val = cacheObjProc.toCacheObject(cacheObjCtx, e.getValue(), true);
-
-                    return new DataStreamerEntry(key, val);
-                }
-            });
-
-            load0(entries0, resFut, keys, 0);
-
-            return new IgniteFutureImpl<>(resFut);
-        }
-        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<KeyCacheObject> keys = null;
-
-            if (entries.size() > 1) {
-                keys = new GridConcurrentHashSet<>(entries.size(), U.capacity(entries.size()), 1);
-
-                for (DataStreamerEntry entry : entries)
-                    keys.add(entry.getKey());
-            }
-
-            load0(entries, resFut, keys, 0);
-
-            return new IgniteFutureImpl<>(resFut);
-        }
-        catch (Throwable e) {
-            resFut.onDone(e);
-
-            if (e instanceof Error)
-                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");
-
-        KeyCacheObject key0 = cacheObjProc.toCacheKeyObject(cacheObjCtx, 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 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<KeyCacheObject> activeKeys,
-        final int remaps
-    ) {
-        assert entries != null;
-
-        Map<ClusterNode, Collection<DataStreamerEntry>> mappings = new HashMap<>();
-
-        boolean initPda = ctx.deploy().enabled() && jobPda == null;
-
-        for (DataStreamerEntry entry : entries) {
-            List<ClusterNode> nodes;
-
-            try {
-                KeyCacheObject key = entry.getKey();
-
-                assert key != null;
-
-                if (initPda) {
-                    jobPda = new DataStreamerPda(key.value(cacheObjCtx, false),
-                        entry.getValue() != null ? entry.getValue().value(cacheObjCtx, false) : null,
-                        updater);
-
-                    initPda = false;
-                }
-
-                nodes = nodes(key);
-            }
-            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(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 (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
-                            load0(entriesForNode, resFut, activeKeys, remaps + 1);
-                    }
-                }
-            };
-
-            GridFutureAdapter<?> f;
-
-            try {
-                f = buf.update(entriesForNode, lsnr);
-            }
-            catch (IgniteInterruptedCheckedException e1) {
-                resFut.onDone(e1);
-
-                return;
-            }
-
-            if (ctx.discovery().node(nodeId) == null) {
-                if (bufMappings.remove(nodeId, buf))
-                    buf.onNodeLeft();
-
-                if (f != null)
-                    f.onDone(new ClusterTopologyCheckedException("Failed to wait for request completion " +
-                        "(node has left): " + nodeId));
-            }
-        }
-    }
-
-    /**
-     * @param key Key to map.
-     * @return Nodes to send requests to.
-     * @throws IgniteCheckedException If failed.
-     */
-    private List<ClusterNode> nodes(KeyCacheObject key) throws IgniteCheckedException {
-        GridAffinityProcessor aff = ctx.affinity();
-
-        return !allowOverwrite() ? aff.mapKeyToPrimaryAndBackups(cacheName, key) :
-            Collections.singletonList(aff.mapKeyToNode(cacheName, key));
-    }
-
-    /**
-     * 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;
-
-                for (IgniteInternalFuture fut = q.poll(); fut != null; fut = q.poll()) {
-                    try {
-                        fut.get();
-                    }
-                    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 IgniteException {
-        enterBusy();
-
-        try {
-            doFlush();
-        }
-        catch (IgniteCheckedException e) {
-            throw U.convertException(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 U.convertException(e);
-        }
-        finally {
-            leaveBusy();
-        }
-    }
-
-    /**
-     * @param cancel {@code True} to close with cancellation.
-     * @throws IgniteException If failed.
-     */
-    @Override public void close(boolean cancel) throws IgniteException {
-        try {
-            closeEx(cancel);
-        }
-        catch (IgniteCheckedException e) {
-            throw U.convertException(e);
-        }
-    }
-
-    /**
-     * @param cancel {@code True} to close with cancellation.
-     * @throws IgniteCheckedException If failed.
-     */
-    public void closeEx(boolean cancel) throws IgniteCheckedException {
-        if (!closed.compareAndSet(false, true))
-            return;
-
-        busyLock.block();
-
-        if (log.isDebugEnabled())
-            log.debug("Closing data streamer [ldr=" + this + ", cancel=" + cancel + ']');
-
-        IgniteCheckedException e = null;
-
-        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();
-            }
-            else
-                doFlush();
-
-            ctx.event().removeLocalEventListener(discoLsnr);
-
-            ctx.io().removeMessageListener(topic);
-        }
-        catch (IgniteCheckedException e0) {
-            e = e0;
-        }
-
-        fut.onDone(null, e);
-
-        if (e != null)
-            throw e;
-    }
-
-    /**
-     * @return {@code true} If the loader is closed.
-     */
-    boolean isClosed() {
-        return fut.isDone();
-    }
-
-    /** {@inheritDoc} */
-    @Override public void close() throws IgniteException {
-        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;
-    }
-
-    /**
-     *
-     */
-    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;
-
-        /** 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 newEntries Infos.
-         * @param lsnr Listener for the operation future.
-         * @throws IgniteInterruptedCheckedException If failed.
-         * @return Future for operation.
-         */
-        @Nullable GridFutureAdapter<?> update(Iterable<DataStreamerEntry> newEntries,
-            IgniteInClosure<IgniteInternalFuture<?>> lsnr) throws IgniteInterruptedCheckedException {
-            List<DataStreamerEntry> entries0 = null;
-            GridFutureAdapter<Object> curFut0;
-
-            synchronized (this) {
-                curFut0 = curFut;
-
-                curFut0.listen(lsnr);
-
-                for (DataStreamerEntry entry : newEntries)
-                    entries.add(entry);
-
-                if (entries.size() >= bufSize) {
-                    entries0 = entries;
-
-                    entries = newEntries();
-                    curFut = new GridFutureAdapter<>();
-                    curFut.listen(signalC);
-                }
-            }
-
-            if (entries0 != null) {
-                submit(entries0, curFut0);
-
-                if (cancelled)
-                    curFut0.onDone(new IgniteCheckedException("Data streamer has been cancelled: " + DataStreamerImpl.this));
-            }
-
-            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;
-
-            synchronized (this) {
-                if (!entries.isEmpty()) {
-                    entries0 = entries;
-                    curFut0 = curFut;
-
-                    entries = newEntries();
-                    curFut = new GridFutureAdapter<>();
-                    curFut.listen(signalC);
-                }
-            }
-
-            if (entries0 != null)
-                submit(entries0, curFut0);
-
-            // 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 {
-            U.acquire(sem);
-        }
-
-        /**
-         * @param f Future that finished.
-         */
-        private void signalTaskFinished(IgniteInternalFuture<Object> f) {
-            assert f != null;
-
-            sem.release();
-        }
-
-        /**
-         * @param entries Entries to submit.
-         * @param curFut Current future.
-         * @throws IgniteInterruptedCheckedException If interrupted.
-         */
-        private void submit(final Collection<DataStreamerEntry> entries, final GridFutureAdapter<Object> curFut)
-            throws IgniteInterruptedCheckedException {
-            assert entries != null;
-            assert !entries.isEmpty();
-            assert curFut != null;
-
-            incrementActiveTasks();
-
-            IgniteInternalFuture<Object> fut;
-
-            if (isLocNode) {
-                fut = ctx.closure().callLocalSafe(
-                    new DataStreamerUpdateJob(ctx, log, cacheName, entries, false, skipStore, updater), false);
-
-                locFuts.add(fut);
-
-                fut.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);
-                        }
-                    }
-                });
-            }
-            else {
-                try {
-                    for (DataStreamerEntry e : entries) {
-                        e.getKey().prepareMarshal(cacheObjCtx);
-
-                        CacheObject val = e.getValue();
-
-                        if (val != null)
-                            val.prepareMarshal(cacheObjCtx);
-                    }
-
-                    if (updaterBytes == null) {
-                        assert updater != null;
-
-                        updaterBytes = ctx.config().getMarshaller().marshal(updater);
-                    }
-
-                    if (topicBytes == null)
-                        topicBytes = ctx.config().getMarshaller().marshal(topic);
-                }
-                catch (IgniteCheckedException e) {
-                    U.error(log, "Failed to marshal (request will not be sent).", e);
-
-                    return;
-                }
-
-                GridDeployment dep = null;
-                GridPeerDeployAware jobPda0 = null;
-
-                if (ctx.deploy().enabled()) {
-                    try {
-                        jobPda0 = jobPda;
-
-                        assert jobPda0 != null;
-
-                        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);
-
-                DataStreamerRequest req = new DataStreamerRequest(
-                    reqId,
-                    topicBytes,
-                    cacheName,
-                    updaterBytes,
-                    entries,
-                    true,
-                    skipStore,
-                    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);
-
-                try {
-                    ctx.io().send(node, TOPIC_DATASTREAM, req, PUBLIC_POOL);
-
-                    if (log.isDebugEnabled())
-                        log.debug("Sent request to node [nodeId=" + node.id() + ", req=" + req + ']');
-                }
-                catch (IgniteCheckedException e) {
-                    if (ctx.discovery().alive(node) && ctx.discovery().pingNode(node.id()))
-                        ((GridFutureAdapter<Object>)fut).onDone(e);
-                    else
-                        ((GridFutureAdapter<Object>)fut).onDone(new ClusterTopologyCheckedException("Failed to send " +
-                            "request (node has left): " + node.id()));
-                }
-            }
-        }
-
-        /**
-         *
-         */
-        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.
-         */
-        void onResponse(DataStreamerResponse res) {
-            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 = ctx.config().getMarshaller().unmarshal(
-                        errBytes,
-                        jobPda0 != null ? jobPda0.classLoader() : U.gridClassLoader());
-                }
-                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 + ']');
-        }
-
-        /**
-         *
-         */
-        void cancelAll() {
-            IgniteCheckedException 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 updater which only loads entry initial value.
-     */
-    private static class IsolatedUpdater implements Updater<KeyCacheObject, CacheObject>,
-        DataStreamerCacheUpdaters.InternalUpdater {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** {@inheritDoc} */
-        @Override public void update(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();
-
-            long topVer = cctx.affinity().affinityTopologyVersion();
-
-            GridCacheVersion ver = cctx.versions().next(topVer);
-
-            for (Map.Entry<KeyCacheObject, CacheObject> e : entries) {
-                try {
-                    e.getKey().finishUnmarshal(cctx.cacheObjectContext(), cctx.deploy().globalLoader());
-
-                    GridCacheEntryEx entry = internalCache.entryEx(e.getKey(), topVer);
-
-                    entry.unswap(true, false);
-
-                    entry.initialValue(e.getValue(),
-                        ver,
-                        CU.TTL_ETERNAL,
-                        CU.EXPIRE_TIME_ETERNAL,
-                        false,
-                        topVer,
-                        GridDrType.DR_LOAD);
-
-                    cctx.evicts().touch(entry, topVer);
-                }
-                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);
-                }
-            }
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerProcessor.java
deleted file mode 100644
index ee12f7a..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerProcessor.java
+++ /dev/null
@@ -1,307 +0,0 @@
-/*
- * 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.datastream;
-
-import org.apache.ignite.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.managers.communication.*;
-import org.apache.ignite.internal.managers.deployment.*;
-import org.apache.ignite.internal.processors.*;
-import org.apache.ignite.internal.util.*;
-import org.apache.ignite.internal.util.typedef.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.internal.util.worker.*;
-import org.apache.ignite.marshaller.*;
-import org.apache.ignite.thread.*;
-import org.jetbrains.annotations.*;
-
-import java.util.*;
-import java.util.concurrent.*;
-
-import static org.apache.ignite.internal.GridTopic.*;
-import static org.apache.ignite.internal.managers.communication.GridIoPolicy.*;
-
-/**
- *
- */
-public class DataStreamerProcessor<K, V> extends GridProcessorAdapter {
-    /** Loaders map (access is not supposed to be highly concurrent). */
-    private Collection<DataStreamerImpl> ldrs = new GridConcurrentHashSet<>();
-
-    /** Busy lock. */
-    private final GridSpinBusyLock busyLock = new GridSpinBusyLock();
-
-    /** Flushing thread. */
-    private Thread flusher;
-
-    /** */
-    private final DelayQueue<DataStreamerImpl<K, V>> flushQ = new DelayQueue<>();
-
-    /** Marshaller. */
-    private final Marshaller marsh;
-
-    /**
-     * @param ctx Kernal context.
-     */
-    public DataStreamerProcessor(GridKernalContext ctx) {
-        super(ctx);
-
-        ctx.io().addMessageListener(TOPIC_DATASTREAM, new GridMessageListener() {
-            @Override public void onMessage(UUID nodeId, Object msg) {
-                assert msg instanceof DataStreamerRequest;
-
-                processRequest(nodeId, (DataStreamerRequest)msg);
-            }
-        });
-
-        marsh = ctx.config().getMarshaller();
-    }
-
-    /** {@inheritDoc} */
-    @Override public void start() throws IgniteCheckedException {
-        if (ctx.config().isDaemon())
-            return;
-
-        flusher = new IgniteThread(new GridWorker(ctx.gridName(), "grid-data-loader-flusher", log) {
-            @Override protected void body() throws InterruptedException {
-                while (!isCancelled()) {
-                    DataStreamerImpl<K, V> ldr = flushQ.take();
-
-                    if (!busyLock.enterBusy())
-                        return;
-
-                    try {
-                        if (ldr.isClosed())
-                            continue;
-
-                        ldr.tryFlush();
-
-                        flushQ.offer(ldr);
-                    }
-                    finally {
-                        busyLock.leaveBusy();
-                    }
-                }
-            }
-        });
-
-        flusher.start();
-
-        if (log.isDebugEnabled())
-            log.debug("Started data streamer processor.");
-    }
-
-    /** {@inheritDoc} */
-    @Override public void onKernalStop(boolean cancel) {
-        if (ctx.config().isDaemon())
-            return;
-
-        ctx.io().removeMessageListener(TOPIC_DATASTREAM);
-
-        busyLock.block();
-
-        U.interrupt(flusher);
-        U.join(flusher, log);
-
-        for (DataStreamerImpl<?, ?> ldr : ldrs) {
-            if (log.isDebugEnabled())
-                log.debug("Closing active data streamer on grid stop [ldr=" + ldr + ", cancel=" + cancel + ']');
-
-            try {
-                ldr.closeEx(cancel);
-            }
-            catch (IgniteInterruptedCheckedException e) {
-                U.warn(log, "Interrupted while waiting for completion of the data streamer: " + ldr, e);
-            }
-            catch (IgniteCheckedException e) {
-                U.error(log, "Failed to close data streamer: " + ldr, e);
-            }
-        }
-
-        if (log.isDebugEnabled())
-            log.debug("Stopped data streamer processor.");
-    }
-
-    /**
-     * @param cacheName Cache name ({@code null} for default cache).
-     * @return Data loader.
-     */
-    public DataStreamerImpl<K, V> dataStreamer(@Nullable String cacheName) {
-        if (!busyLock.enterBusy())
-            throw new IllegalStateException("Failed to create data streamer (grid is stopping).");
-
-        try {
-            final DataStreamerImpl<K, V> ldr = new DataStreamerImpl<>(ctx, cacheName, flushQ);
-
-            ldrs.add(ldr);
-
-            ldr.internalFuture().listen(new CI1<IgniteInternalFuture<?>>() {
-                @Override public void apply(IgniteInternalFuture<?> f) {
-                    boolean b = ldrs.remove(ldr);
-
-                    assert b : "Loader has not been added to set: " + ldr;
-
-                    if (log.isDebugEnabled())
-                        log.debug("Loader has been completed: " + ldr);
-                }
-            });
-
-            return ldr;
-        }
-        finally {
-            busyLock.leaveBusy();
-        }
-    }
-
-    /**
-     * @param nodeId Sender ID.
-     * @param req Request.
-     */
-    private void processRequest(UUID nodeId, DataStreamerRequest req) {
-        if (!busyLock.enterBusy()) {
-            if (log.isDebugEnabled())
-                log.debug("Ignoring data load request (node is stopping): " + req);
-
-            return;
-        }
-
-        try {
-            if (log.isDebugEnabled())
-                log.debug("Processing data load request: " + req);
-
-            Object topic;
-
-            try {
-                topic = marsh.unmarshal(req.responseTopicBytes(), null);
-            }
-            catch (IgniteCheckedException e) {
-                U.error(log, "Failed to unmarshal topic from request: " + req, e);
-
-                return;
-            }
-
-            ClassLoader clsLdr;
-
-            if (req.forceLocalDeployment())
-                clsLdr = U.gridClassLoader();
-            else {
-                GridDeployment dep = ctx.deploy().getGlobalDeployment(
-                    req.deploymentMode(),
-                    req.sampleClassName(),
-                    req.sampleClassName(),
-                    req.userVersion(),
-                    nodeId,
-                    req.classLoaderId(),
-                    req.participants(),
-                    null);
-
-                if (dep == null) {
-                    sendResponse(nodeId,
-                        topic,
-                        req.requestId(),
-                        new IgniteCheckedException("Failed to get deployment for request [sndId=" + nodeId +
-                            ", req=" + req + ']'),
-                        false);
-
-                    return;
-                }
-
-                clsLdr = dep.classLoader();
-            }
-
-            IgniteDataStreamer.Updater<K, V> updater;
-
-            try {
-                updater = marsh.unmarshal(req.updaterBytes(), clsLdr);
-            }
-            catch (IgniteCheckedException e) {
-                U.error(log, "Failed to unmarshal message [nodeId=" + nodeId + ", req=" + req + ']', e);
-
-                sendResponse(nodeId, topic, req.requestId(), e, false);
-
-                return;
-            }
-
-            Collection<DataStreamerEntry> col = req.entries();
-
-            DataStreamerUpdateJob job = new DataStreamerUpdateJob(ctx,
-                log,
-                req.cacheName(),
-                col,
-                req.ignoreDeploymentOwnership(),
-                req.skipStore(),
-                updater);
-
-            Exception err = null;
-
-            try {
-                job.call();
-            }
-            catch (Exception e) {
-                U.error(log, "Failed to finish update job.", e);
-
-                err = e;
-            }
-
-            sendResponse(nodeId, topic, req.requestId(), err, req.forceLocalDeployment());
-        }
-        finally {
-            busyLock.leaveBusy();
-        }
-    }
-
-    /**
-     * @param nodeId Node ID.
-     * @param resTopic Response topic.
-     * @param reqId Request ID.
-     * @param err Error.
-     * @param forceLocDep Force local deployment.
-     */
-    private void sendResponse(UUID nodeId, Object resTopic, long reqId, @Nullable Throwable err,
-        boolean forceLocDep) {
-        byte[] errBytes;
-
-        try {
-            errBytes = err != null ? marsh.marshal(err) : null;
-        }
-        catch (IgniteCheckedException e) {
-            U.error(log, "Failed to marshal message.", e);
-
-            return;
-        }
-
-        DataStreamerResponse res = new DataStreamerResponse(reqId, errBytes, forceLocDep);
-
-        try {
-            ctx.io().send(nodeId, resTopic, res, PUBLIC_POOL);
-        }
-        catch (IgniteCheckedException e) {
-            if (ctx.discovery().alive(nodeId))
-                U.error(log, "Failed to respond to node [nodeId=" + nodeId + ", res=" + res + ']', e);
-            else if (log.isDebugEnabled())
-                log.debug("Node has left the grid: " + nodeId);
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public void printMemoryStats() {
-        X.println(">>>");
-        X.println(">>> Data streamer processor memory stats [grid=" + ctx.gridName() + ']');
-        X.println(">>>   ldrsSize: " + ldrs.size());
-    }
-}


Mime
View raw message