renaming wip

examples/src/main/java/org/apache/ignite/examples/datagrid/CacheDataStreamerExample.java

@@ -66,7 +66,7 @@ public static void main(String[] args) throws IgniteException {
             try (IgniteDataStreamer<Integer, String> stmr = ignite.dataStreamer(CACHE_NAME)) {
                 // Configure loader.
                 stmr.perNodeBufferSize(1024);
-                stmr.perNodeParallelStreamOperations(8);
+                stmr.perNodeParallelOperations(8);
 
                 for (int i = 0; i < ENTRY_COUNT; i++) {
                     stmr.addData(i, Integer.toString(i));

modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java

@@ -140,7 +140,7 @@ public static void main(String[] args) throws Exception {
 
         MessageCodeGenerator gen = new MessageCodeGenerator(srcDir);
 
-        gen.generateAndWrite(IgniteDataLoaderEntry.class);
+        gen.generateAndWrite(DataStreamerEntry.class);
 
 //        gen.generateAndWrite(GridDistributedLockRequest.class);
 //        gen.generateAndWrite(GridDistributedLockResponse.class);

modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java

@@ -52,7 +52,7 @@
  *      value.
  *  </li>
  *  <li>
- *      {@link #perNodeParallelStreamOperations(int)} - sometimes data may be added
+ *      {@link #perNodeParallelOperations(int)} - sometimes data may be added
  *      to the data streamer via {@link #addData(Object, Object)} method faster than it can
  *      be put in cache. In this case, new buffered stream messages are sent to remote nodes
  *      before responses from previous ones are received. This could cause unlimited heap
@@ -102,15 +102,17 @@ public interface IgniteDataStreamer<K, V> extends AutoCloseable {
     public String cacheName();
 
     /**
-     * Gets flag value indicating that this data streamer assumes that there are no other concurrent updates to the cache.
+     * Gets flag value indicating that this data streamer assumes that
+     * there are no other concurrent updates to the cache.
      * Default is {@code false}.
      *
      * @return Flag value.
      */
     public boolean allowOverwrite();
 
     /**
-     * Sets flag indicating that this data streamer should assume that there are no other concurrent updates to the cache.
+     * Sets flag indicating that this data streamer should assume
+     * that there are no other concurrent updates to the cache.
      * Should not be used when custom cache updater set using {@link #updater(IgniteDataStreamer.Updater)} method.
      * Default is {@code false}. When this flag is set, updates will not be propagated to the cache store.
      *
@@ -154,22 +156,22 @@ public interface IgniteDataStreamer<K, V> extends AutoCloseable {
     public void perNodeBufferSize(int bufSize);
 
     /**
-     * Gets maximum number of parallel stream operations for a single node.
+     * Gets maximum number of parallel update operations for a single node.
      *
      * @return Maximum number of parallel stream operations for a single node.
      */
-    public int perNodeParallelStreamOperations();
+    public int perNodeParallelOperations();
 
     /**
-     * Sets maximum number of parallel stream operations for a single node.
+     * Sets maximum number of parallel update operations for a single node.
      * <p>
      * This method should be called prior to {@link #addData(Object, Object)} call.
      * <p>
      * If not provided, default value is {@link #DFLT_MAX_PARALLEL_OPS}.
      *
      * @param parallelOps Maximum number of parallel stream operations for a single node.
      */
-    public void perNodeParallelStreamOperations(int parallelOps);
+    public void perNodeParallelOperations(int parallelOps);
 
     /**
      * Gets automatic flush frequency. Essentially, this is the time after which the
@@ -284,7 +286,7 @@ public IgniteFuture<?> addData(K key, @Nullable V val) throws IgniteException, I
      * @throws IgniteInterruptedException If thread has been interrupted.
      * @throws IllegalStateException If grid has been concurrently stopped or
      *      {@link #close(boolean)} has already been called on streamer.
-     * @see #allowOverwrite()      
+     * @see #allowOverwrite()
      */
     public IgniteFuture<?> addData(Map.Entry<K, V> entry) throws IgniteException, IgniteInterruptedException,
         IllegalStateException;
@@ -298,7 +300,7 @@ public IgniteFuture<?> addData(Map.Entry<K, V> entry) throws IgniteException, Ig
      * the streamer.
      * <p>
      * Note: if {@link IgniteDataStreamer#allowOverwrite()} set to {@code false} (by default)
-     * then data streamer will not overwrite existing cache entries for better performance 
+     * then data streamer will not overwrite existing cache entries for better performance
      * (to change, set {@link IgniteDataStreamer#allowOverwrite(boolean)} to {@code true})
      *
      * @param entries Collection of entries to be streamed.
@@ -318,7 +320,7 @@ public IgniteFuture<?> addData(Map.Entry<K, V> entry) throws IgniteException, Ig
      * the streamer.
      * <p>
      * Note: if {@link IgniteDataStreamer#allowOverwrite()} set to {@code false} (by default)
-     * then data streamer will not overwrite existing cache entries for better performance 
+     * then data streamer will not overwrite existing cache entries for better performance
      * (to change, set {@link IgniteDataStreamer#allowOverwrite(boolean)} to {@code true})
      *
      * @param entries Map to be streamed.

modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java

@@ -247,7 +247,7 @@ public interface GridKernalContext extends Iterable<GridComponent> {
      *
      * @return Data streamer processor.
      */
-    public <K, V> IgniteDataStreamerProcessor<K, V> dataStream();
+    public <K, V> DataStreamerProcessor<K, V> dataStream();
 
     /**
      * Gets file system processor.

modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java

@@ -200,7 +200,7 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
 
     /** */
     @GridToStringInclude
-    private IgniteDataStreamerProcessor dataLdrProc;
+    private DataStreamerProcessor dataLdrProc;
 
     /** */
     @GridToStringInclude
@@ -457,8 +457,8 @@ else if (comp instanceof GridAffinityProcessor)
             affProc = (GridAffinityProcessor)comp;
         else if (comp instanceof GridRestProcessor)
             restProc = (GridRestProcessor)comp;
-        else if (comp instanceof IgniteDataStreamerProcessor)
-            dataLdrProc = (IgniteDataStreamerProcessor)comp;
+        else if (comp instanceof DataStreamerProcessor)
+            dataLdrProc = (DataStreamerProcessor)comp;
         else if (comp instanceof IgfsProcessorAdapter)
             igfsProc = (IgfsProcessorAdapter)comp;
         else if (comp instanceof GridOffHeapProcessor)
@@ -671,8 +671,8 @@ public void addHelper(Object helper) {
 
     /** {@inheritDoc} */
     @SuppressWarnings("unchecked")
-    @Override public <K, V> IgniteDataStreamerProcessor<K, V> dataStream() {
-        return (IgniteDataStreamerProcessor<K, V>)dataLdrProc;
+    @Override public <K, V> DataStreamerProcessor<K, V> dataStream() {
+        return (DataStreamerProcessor<K, V>)dataLdrProc;
     }
 
     /** {@inheritDoc} */

modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java

@@ -69,7 +69,7 @@ public enum GridTopic {
     TOPIC_IGFS,
 
     /** */
-    TOPIC_DATALOAD,
+    TOPIC_DATASTREAM,
 
     /** */
     TOPIC_STREAM,

modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java

@@ -755,7 +755,7 @@ public void start(final IgniteConfiguration cfg,
             startProcessor(new GridTaskProcessor(ctx));
             startProcessor((GridProcessor)SCHEDULE.createOptional(ctx));
             startProcessor(new GridRestProcessor(ctx));
-            startProcessor(new IgniteDataStreamerProcessor(ctx));
+            startProcessor(new DataStreamerProcessor(ctx));
             startProcessor(new GridStreamProcessor(ctx));
             startProcessor((GridProcessor) IGFS.create(ctx, F.isEmpty(cfg.getFileSystemConfiguration())));
             startProcessor(new GridContinuousProcessor(ctx));

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java

@@ -385,12 +385,12 @@ public GridIoMessageFactory(MessageFactory[] ext) {
                 break;
 
             case 62:
-                msg = new GridDataLoadRequest();
+                msg = new DataStreamerRequest();
 
                 break;
 
             case 63:
-                msg = new GridDataLoadResponse();
+                msg = new DataStreamerResponse();
 
                 break;
 
@@ -525,7 +525,7 @@ public GridIoMessageFactory(MessageFactory[] ext) {
                 break;
 
             case 95:
-                msg = new IgniteDataLoaderEntry();
+                msg = new DataStreamerEntry();
 
                 break;
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java

@@ -3732,7 +3732,7 @@ protected void checkJta() throws IgniteCheckedException {
         final ExpiryPolicy plc = plc0 != null ? plc0 : ctx.expiry();
 
         if (ctx.store().isLocalStore()) {
-            IgniteDataStreamerImpl ldr = ctx.kernalContext().dataStream().dataStreamer(ctx.namex());
+            DataStreamerImpl ldr = ctx.kernalContext().dataStream().dataStreamer(ctx.namex());
 
             try {
                 ldr.updater(new IgniteDrDataStreamerCacheUpdater());
@@ -3883,18 +3883,18 @@ public IgniteInternalFuture<?> loadAll(
      * @throws IgniteCheckedException If failed.
      */
     private void localLoadAndUpdate(final Collection<? extends K> keys) throws IgniteCheckedException {
-        try (final IgniteDataStreamerImpl<KeyCacheObject, CacheObject> ldr =
+        try (final DataStreamerImpl<KeyCacheObject, CacheObject> ldr =
                  ctx.kernalContext().<KeyCacheObject, CacheObject>dataStream().dataStreamer(ctx.namex())) {
             ldr.allowOverwrite(true);
             ldr.skipStore(true);
 
-            final Collection<IgniteDataLoaderEntry> col = new ArrayList<>(ldr.perNodeBufferSize());
+            final Collection<DataStreamerEntry> col = new ArrayList<>(ldr.perNodeBufferSize());
 
             Collection<KeyCacheObject> keys0 = ctx.cacheKeysView(keys);
 
             ctx.store().loadAllFromStore(null, keys0, new CIX2<KeyCacheObject, Object>() {
                 @Override public void applyx(KeyCacheObject key, Object val) {
-                    col.add(new IgniteDataLoaderEntry(key, ctx.toCacheObject(val)));
+                    col.add(new DataStreamerEntry(key, ctx.toCacheObject(val)));
 
                     if (col.size() == ldr.perNodeBufferSize()) {
                         ldr.addDataInternal(col);
@@ -3926,7 +3926,7 @@ public void localLoad(Collection<? extends K> keys,
         Collection<KeyCacheObject> keys0 = ctx.cacheKeysView(keys);
 
         if (ctx.store().isLocalStore()) {
-            IgniteDataStreamerImpl ldr = ctx.kernalContext().dataStream().dataStreamer(ctx.namex());
+            DataStreamerImpl ldr = ctx.kernalContext().dataStream().dataStreamer(ctx.namex());
 
             try {
                 ldr.updater(new IgniteDrDataStreamerCacheUpdater());
@@ -5882,7 +5882,7 @@ private class LocalStoreLoadClosure extends CIX3<KeyCacheObject, Object, GridCac
         final Collection<GridCacheRawVersionedEntry> col;
 
         /** */
-        final IgniteDataStreamerImpl<K, V> ldr;
+        final DataStreamerImpl<K, V> ldr;
 
         /** */
         final ExpiryPolicy plc;
@@ -5893,7 +5893,7 @@ private class LocalStoreLoadClosure extends CIX3<KeyCacheObject, Object, GridCac
          * @param plc Optional expiry policy.
          */
         private LocalStoreLoadClosure(@Nullable IgniteBiPredicate<K, V> p,
-            IgniteDataStreamerImpl<K, V> ldr,
+            DataStreamerImpl<K, V> ldr,
             @Nullable ExpiryPolicy plc) {
             this.p = p;
             this.ldr = ldr;

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java

@@ -283,11 +283,11 @@ private GlobalRemoveAllCallable(String cacheName, long topVer) {
                 else
                     dht = (GridDhtCacheAdapter<K, V>)cacheAdapter;
 
-                try (IgniteDataStreamerImpl<KeyCacheObject, Object> dataLdr =
-                         (IgniteDataStreamerImpl)ignite.dataStreamer(cacheName)) {
-                    ((IgniteDataStreamerImpl)dataLdr).maxRemapCount(0);
+                try (DataStreamerImpl<KeyCacheObject, Object> dataLdr =
+                         (DataStreamerImpl)ignite.dataStreamer(cacheName)) {
+                    ((DataStreamerImpl)dataLdr).maxRemapCount(0);
 
-                    dataLdr.updater(IgniteDataStreamerCacheUpdaters.<KeyCacheObject, Object>batched());
+                    dataLdr.updater(DataStreamerCacheUpdaters.<KeyCacheObject, Object>batched());
 
                     for (GridDhtLocalPartition locPart : dht.topology().currentLocalPartitions()) {
                         if (!locPart.isEmpty() && locPart.primary(topVer)) {

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheRawVersionedEntry.java

@@ -32,7 +32,7 @@
 /**
  * Raw versioned entry.
  */
-public class GridCacheRawVersionedEntry<K, V> extends IgniteDataLoaderEntry implements
+public class GridCacheRawVersionedEntry<K, V> extends DataStreamerEntry implements
     GridCacheVersionedEntry<K, V>, GridCacheVersionable, Externalizable {
     /** */
     private static final long serialVersionUID = 0L;

modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/IgniteDataStreamerCacheUpdaters.java → modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerCacheUpdaters.java

@@ -27,7 +27,7 @@
 /**
  * Bundled factory for cache updaters.
  */
-public class IgniteDataStreamerCacheUpdaters {
+public class DataStreamerCacheUpdaters {
     /** */
     private static final IgniteDataStreamer.Updater INDIVIDUAL = new Individual();
 

modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/IgniteDataLoaderEntry.java → modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerEntry.java

@@ -28,7 +28,7 @@
 /**
  *
  */
-public class IgniteDataLoaderEntry implements Map.Entry<KeyCacheObject, CacheObject>, Message {
+public class DataStreamerEntry implements Map.Entry<KeyCacheObject, CacheObject>, Message {
     /** */
     private static final long serialVersionUID = 0L;
 
@@ -43,15 +43,15 @@ public class IgniteDataLoaderEntry implements Map.Entry<KeyCacheObject, CacheObj
     /**
      *
      */
-    public IgniteDataLoaderEntry() {
+    public DataStreamerEntry() {
         // No-op.
     }
 
     /**
      * @param key Key.
      * @param val Value.
      */
-    public IgniteDataLoaderEntry(KeyCacheObject key, CacheObject val) {
+    public DataStreamerEntry(KeyCacheObject key, CacheObject val) {
         this.key = key;
         this.val = val;
     }
@@ -165,6 +165,6 @@ public <K, V> Map.Entry<K, V> toEntry(final GridCacheContext ctx) {
 
     /** {@inheritDoc} */
     @Override public String toString() {
-        return S.toString(IgniteDataLoaderEntry.class, this);
+        return S.toString(DataStreamerEntry.class, this);
     }
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/IgniteDataStreamerFuture.java → modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerFuture.java

@@ -18,34 +18,25 @@
 package org.apache.ignite.internal.processors.datastream;
 
 import org.apache.ignite.*;
-import org.apache.ignite.internal.*;
 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 IgniteDataStreamerFuture extends GridFutureAdapter<Object> {
+class DataStreamerFuture extends GridFutureAdapter<Object> {
     /** */
     private static final long serialVersionUID = 0L;
 
     /** Data loader. */
     @GridToStringExclude
-    private IgniteDataStreamerImpl dataLdr;
+    private DataStreamerImpl dataLdr;
 
     /**
-     * Default constructor for {@link java.io.Externalizable} support.
-     */
-    public IgniteDataStreamerFuture() {
-        // No-op.
-    }
-
-    /**
-     * @param ctx Context.
      * @param dataLdr Data streamer.
      */
-    IgniteDataStreamerFuture(GridKernalContext ctx, IgniteDataStreamerImpl dataLdr) {
+    DataStreamerFuture(DataStreamerImpl dataLdr) {
         assert dataLdr != null;
 
         this.dataLdr = dataLdr;
@@ -64,6 +55,6 @@ public IgniteDataStreamerFuture() {
 
     /** {@inheritDoc} */
     @Override public String toString() {
-        return S.toString(IgniteDataStreamerFuture.class, this, super.toString());
+        return S.toString(DataStreamerFuture.class, this, super.toString());
     }
 }