org.apache.ignite.internal.processors.datastreamer

## Class DataStreamerImpl<K,V>

- java.lang.Object
- - org.apache.ignite.internal.processors.datastreamer.DataStreamerImpl<K,V>

All Implemented Interfaces:
  AutoCloseable, Comparable<Delayed>, Delayed, IgniteDataStreamer<K,V>

    public class DataStreamerImpl<K,V>
    extends Object
    implements IgniteDataStreamer<K,V>, Delayed

Data streamer implementation.

### Field Summary

### Fields inherited from interface org.apache.ignite.IgniteDataStreamer

`DFLT_MAX_PARALLEL_OPS, DFLT_PER_NODE_BUFFER_SIZE`

### Constructor Summary

Constructors
Constructor and Description
`DataStreamerImpl(GridKernalContext ctx, String cacheName, DelayQueue<DataStreamerImpl<K,V>> flushQ)`

### Method Summary

Methods
`IgniteFuture<?>` `addData(Collection<? extends Map.Entry<K,V>> entries)` Adds data for streaming on remote node.
`IgniteFuture<?>` `addData(K key, V val)` Adds data for streaming on remote node.
`IgniteFuture<?>` `addData(Map.Entry<K,V> entry)` Adds data for streaming on remote node.
`IgniteFuture<?>` `addData(Map<K,V> entries)` Adds data for streaming on remote node.
`IgniteFuture<?>` `addDataInternal(Collection<? extends DataStreamerEntry> entries)`
`IgniteFuture<?>` `addDataInternal(KeyCacheObject key, CacheObject val)`
`boolean` `allowOverwrite()` Gets flag enabling overwriting existing values in cache.
`void` `allowOverwrite(boolean allow)` Sets flag enabling overwriting existing values in cache.
`long` `autoFlushFrequency()` Gets automatic flush frequency.
`void` `autoFlushFrequency(long autoFlushFreq)` Sets automatic flush frequency.
`String` `cacheName()` Name of cache to stream data to.
`CacheObjectContext` `cacheObjectContext()`
`void` `close()` Closes data streamer.
`void` `close(boolean cancel)` Streams any remaining data and closes this streamer.
`void` `closeEx(boolean cancel)`
`void` `closeEx(boolean cancel, IgniteCheckedException err)`
`int` `compareTo(Delayed o)`
`void` `deployClass(Class<?> depCls)` Optional deploy class for peer deployment.
`void` `flush()` Streams any remaining data, but doesn't close the streamer.
`IgniteFuture<?>` `future()` Gets future for this streaming process.
`long` `getDelay(TimeUnit unit)`
`IgniteInternalFuture<?>` `internalFuture()`
`void` `ioPolicyResolver(IgniteClosure<ClusterNode,Byte> ioPlcRslvr)`
`boolean` `keepBinary()` Gets flag indicating that objects should be kept in binary format when passed to the stream receiver.
`void` `keepBinary(boolean keepBinary)` Sets flag indicating that objects should be kept in binary format when passes to the steam receiver.
`int` `maxRemapCount()`
`void` `maxRemapCount(int maxRemapCnt)`
`void` `onDisconnected(IgniteFuture<?> reconnectFut)`
`int` `perNodeBufferSize()` Gets size of per node key-value pairs buffer.
`void` `perNodeBufferSize(int bufSize)` Sets size of per node key-value pairs buffer.
`int` `perNodeParallelOperations()` Gets maximum number of parallel stream operations for a single node.
`void` `perNodeParallelOperations(int parallelOps)` Sets maximum number of parallel stream operations for a single node.
`void` `receiver(StreamReceiver<K,V> rcvr)` Sets custom stream receiver to this data streamer.
`IgniteFuture<?>` `removeData(K key)` Adds key for removal on remote node.
`IgniteFuture<?>` `removeDataInternal(KeyCacheObject key)`
`boolean` `skipStore()` Gets flag indicating that write-through behavior should be disabled for data streaming.
`void` `skipStore(boolean skipStore)` Sets flag indicating that write-through behavior should be disabled for data streaming.
`String` `toString()`
`void` `tryFlush()` Flushes every internal buffer if buffer was flushed before passed in threshold.

### Methods inherited from class java.lang.Object

`clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait`

### Constructor Detail

#### DataStreamerImpl

    public DataStreamerImpl(GridKernalContext ctx,
                    @Nullable
                    String cacheName,
                    DelayQueue<DataStreamerImpl<K,V>> flushQ)

Parameters:
  `ctx` - Grid kernal context.
  `cacheName` - Cache name.
  `flushQ` - Flush queue.

### Method Detail

#### cacheObjectContext

    public CacheObjectContext cacheObjectContext()

Returns:
  Cache object context.

#### future

    public IgniteFuture<?> future()

Gets future for this streaming process. This future completes whenever method `IgniteDataStreamer.close(boolean)` completes. By attaching listeners to this future it is possible to get asynchronous notifications for completion of this streaming process.

Specified by:
  `future` in interface `IgniteDataStreamer<K,V>`
Returns:
  Future for this streaming process.

#### internalFuture

    public IgniteInternalFuture<?> internalFuture()

Returns:
  Internal future.

#### deployClass

    public void deployClass(Class<?> depCls)

Optional deploy class for peer deployment. All classes added by a data streamer must be class-loadable from the same class-loader. Ignite will make the best effort to detect the most suitable class-loader for data loading. However, in complex cases, where compound or deeply nested class-loaders are used, it is best to specify a deploy class which can be any class loaded by the class-loader for given data.

Specified by:
  `deployClass` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `depCls` - Any class loaded by the class-loader for given data.

#### receiver

    public void receiver(StreamReceiver<K,V> rcvr)

Sets custom stream receiver to this data streamer.

Specified by:
  `receiver` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `rcvr` - Stream receiver.

#### allowOverwrite

    public boolean allowOverwrite()

Gets flag enabling overwriting existing values in cache. Data streamer will perform better if this flag is disabled.

This flag is disabled by default (default is `false`).

Specified by:
  `allowOverwrite` in interface `IgniteDataStreamer<K,V>`
Returns:
  `True` if overwriting is allowed, `false` otherwise..

#### allowOverwrite

    public void allowOverwrite(boolean allow)

Sets flag enabling overwriting existing values in cache. Data streamer will perform better if this flag is disabled.

Should not be used when custom cache receiver set using `IgniteDataStreamer.receiver(StreamReceiver)` method.

Note that when this flag is `false`, updates will not be propagated to the cache store (i.e. `IgniteDataStreamer.skipStore()` flag will be set to `true` implicitly).

This flag is disabled by default (default is `false`).

Specified by:
  `allowOverwrite` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `allow` - Flag value.

#### skipStore

    public boolean skipStore()

Gets flag indicating that write-through behavior should be disabled for data streaming. Default is `false`.

Specified by:
  `skipStore` in interface `IgniteDataStreamer<K,V>`
Returns:
  Skip store flag.

#### skipStore

    public void skipStore(boolean skipStore)

Sets flag indicating that write-through behavior should be disabled for data streaming. Default is `false`.

Specified by:
  `skipStore` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `skipStore` - Skip store flag.

#### keepBinary

    public boolean keepBinary()

Gets flag indicating that objects should be kept in binary format when passed to the stream receiver. Default is `false`.

Specified by:
  `keepBinary` in interface `IgniteDataStreamer<K,V>`
Returns:
  Skip store flag.

#### keepBinary

    public void keepBinary(boolean keepBinary)

Sets flag indicating that objects should be kept in binary format when passes to the steam receiver. Default is `false`.

Specified by:
  `keepBinary` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `keepBinary` - Keep binary flag.

#### cacheName

    @Nullable
    public String cacheName()

Name of cache to stream data to.

Specified by:
  `cacheName` in interface `IgniteDataStreamer<K,V>`
Returns:
  Cache name or `null` for default cache.

#### perNodeBufferSize

    public int perNodeBufferSize()

Gets size of per node key-value pairs buffer.

Specified by:
  `perNodeBufferSize` in interface `IgniteDataStreamer<K,V>`
Returns:
  Per node buffer size.

#### perNodeBufferSize

    public void perNodeBufferSize(int bufSize)

Sets size of per node key-value pairs buffer.

This method should be called prior to `IgniteDataStreamer.addData(Object, Object)` call.

If not provided, default value is `IgniteDataStreamer.DFLT_PER_NODE_BUFFER_SIZE`.

Specified by:
  `perNodeBufferSize` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `bufSize` - Per node buffer size.

#### perNodeParallelOperations

    public int perNodeParallelOperations()

Gets maximum number of parallel stream operations for a single node.

Specified by:
  `perNodeParallelOperations` in interface `IgniteDataStreamer<K,V>`
Returns:
  Maximum number of parallel stream operations for a single node.

#### perNodeParallelOperations

    public void perNodeParallelOperations(int parallelOps)

Sets maximum number of parallel stream operations for a single node.

This method should be called prior to `IgniteDataStreamer.addData(Object, Object)` call.

If not provided, default value is `IgniteDataStreamer.DFLT_MAX_PARALLEL_OPS`.

Specified by:
  `perNodeParallelOperations` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `parallelOps` - Maximum number of parallel stream operations for a single node.

#### autoFlushFrequency

    public long autoFlushFrequency()

Gets automatic flush frequency. Essentially, this is the time after which the streamer will make an attempt to submit all data added so far to remote nodes. Note that there is no guarantee that data will be delivered after this concrete attempt (e.g., it can fail when topology is changing), but it won't be lost anyway.

If set to `0`, automatic flush is disabled.

Automatic flush is disabled by default (default value is `0`).

Specified by:
  `autoFlushFrequency` in interface `IgniteDataStreamer<K,V>`
Returns:
  Flush frequency or `0` if automatic flush is disabled.
See Also:
  `IgniteDataStreamer.flush()`

#### autoFlushFrequency

    public void autoFlushFrequency(long autoFlushFreq)

Sets automatic flush frequency. Essentially, this is the time after which the streamer will make an attempt to submit all data added so far to remote nodes. Note that there is no guarantee that data will be delivered after this concrete attempt (e.g., it can fail when topology is changing), but it won't be lost anyway.

If set to `0`, automatic flush is disabled.

Automatic flush is disabled by default (default value is `0`).

Specified by:
  `autoFlushFrequency` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `autoFlushFreq` - Flush frequency or `0` to disable automatic flush.
See Also:
  `IgniteDataStreamer.flush()`

#### addData

    public IgniteFuture<?> addData(Map<K,V> entries)
                            throws IllegalStateException

Adds data for streaming on remote node. This method can be called from multiple threads in parallel to speed up streaming if needed.

Note that streamer will stream data concurrently by multiple internal threads, so the data may get to remote nodes in different order from which it was added to the streamer.

Note: if `IgniteDataStreamer.allowOverwrite()` set to `false` (by default) then data streamer will not overwrite existing cache entries for better performance (to change, set `IgniteDataStreamer.allowOverwrite(boolean)` to `true`)

Specified by:
  `addData` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `entries` - Map to be streamed.
Returns:
  Future for this stream operation.
Throws:
  `IllegalStateException` - If grid has been concurrently stopped or `IgniteDataStreamer.close(boolean)` has already been called on streamer.
See Also:
  `IgniteDataStreamer.allowOverwrite()`

#### addData

    public IgniteFuture<?> addData(Collection<? extends Map.Entry<K,V>> entries)

Adds data for streaming on remote node. This method can be called from multiple threads in parallel to speed up streaming if needed.

Note that streamer will stream data concurrently by multiple internal threads, so the data may get to remote nodes in different order from which it was added to the streamer.

Note: if `IgniteDataStreamer.allowOverwrite()` set to `false` (by default) then data streamer will not overwrite existing cache entries for better performance (to change, set `IgniteDataStreamer.allowOverwrite(boolean)` to `true`)

Specified by:
  `addData` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `entries` - Collection of entries to be streamed.
Returns:
  Future for this stream operation.
See Also:
  `IgniteDataStreamer.allowOverwrite()`

#### addDataInternal

    public IgniteFuture<?> addDataInternal(KeyCacheObject key,
                                  CacheObject val)

Parameters:
  `key` - Key.
  `val` - Value.
Returns:
  Future.

#### removeDataInternal

    public IgniteFuture<?> removeDataInternal(KeyCacheObject key)

Parameters:
  `key` - Key.
Returns:
  Future.

#### addDataInternal

    public IgniteFuture<?> addDataInternal(Collection<? extends DataStreamerEntry> entries)

Parameters:
  `entries` - Entries.
Returns:
  Future.

#### addData

    public IgniteFuture<?> addData(Map.Entry<K,V> entry)

Adds data for streaming on remote node. This method can be called from multiple threads in parallel to speed up streaming if needed.

Note that streamer will stream data concurrently by multiple internal threads, so the data may get to remote nodes in different order from which it was added to the streamer.

Note: if `IgniteDataStreamer.allowOverwrite()` set to `false` (by default) then data streamer will not overwrite existing cache entries for better performance (to change, set `IgniteDataStreamer.allowOverwrite(boolean)` to `true`)

Specified by:
  `addData` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `entry` - Entry.
Returns:
  Future fo this operation.
See Also:
  `IgniteDataStreamer.allowOverwrite()`

#### addData

    public IgniteFuture<?> addData(K key,
                          V val)

Adds data for streaming on remote node. This method can be called from multiple threads in parallel to speed up streaming if needed.

Note that streamer will stream data concurrently by multiple internal threads, so the data may get to remote nodes in different order from which it was added to the streamer.

Note: if `IgniteDataStreamer.allowOverwrite()` set to `false` (by default) then data streamer will not overwrite existing cache entries for better performance (to change, set `IgniteDataStreamer.allowOverwrite(boolean)` to `true`)

Specified by:
  `addData` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `key` - Key.
  `val` - Value or `null` if respective entry must be removed from cache.
Returns:
  Future fo this operation.
See Also:
  `IgniteDataStreamer.allowOverwrite()`

#### removeData

    public IgniteFuture<?> removeData(K key)

Adds key for removal on remote node. Equivalent to `addData(key, null)`.

Specified by:
  `removeData` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `key` - Key.
Returns:
  Future fo this operation.

#### ioPolicyResolver

    public void ioPolicyResolver(IgniteClosure<ClusterNode,Byte> ioPlcRslvr)

Parameters:
  `ioPlcRslvr` - IO policy resolver.

#### flush

    public void flush()
               throws javax.cache.CacheException

Streams any remaining data, but doesn't close the streamer. Data can be still added after flush is finished. This method blocks and doesn't allow to add any data until all data is streamed.

If another thread is already performing flush, this method will block, wait for another thread to complete flush and exit. If you don't want to wait in this case, use `IgniteDataStreamer.tryFlush()` method.

Specified by:
  `flush` in interface `IgniteDataStreamer<K,V>`
Throws:
  `javax.cache.CacheException` - If failed to map key to node.
See Also:
  `IgniteDataStreamer.tryFlush()`

#### tryFlush

    public void tryFlush()
                  throws IgniteInterruptedException

Flushes every internal buffer if buffer was flushed before passed in threshold.

Does not wait for result and does not fail on errors assuming that this method should be called periodically.

Specified by:
  `tryFlush` in interface `IgniteDataStreamer<K,V>`
Throws:
  `IgniteInterruptedException` - If thread has been interrupted.
See Also:
  `IgniteDataStreamer.flush()`

#### close

    public void close(boolean cancel)
               throws javax.cache.CacheException

Description copied from interface: `IgniteDataStreamer`
Streams any remaining data and closes this streamer.

Specified by:
  `close` in interface `IgniteDataStreamer<K,V>`
Parameters:
  `cancel` - `True` to close with cancellation.
Throws:
  `javax.cache.CacheException` - If failed.

#### closeEx

    public void closeEx(boolean cancel)
                 throws IgniteCheckedException

Parameters:
  `cancel` - `True` to close with cancellation.
Throws:
  `IgniteCheckedException` - If failed.

#### closeEx

    public void closeEx(boolean cancel,
               IgniteCheckedException err)
                 throws IgniteCheckedException

Parameters:
  `cancel` - `True` to close with cancellation.
  `err` - Error.
Throws:
  `IgniteCheckedException` - If failed.

#### onDisconnected

    public void onDisconnected(IgniteFuture<?> reconnectFut)
                        throws IgniteCheckedException

Parameters:
  `reconnectFut` - Reconnect future.
Throws:
  `IgniteCheckedException` - If failed.

#### close

    public void close()
               throws javax.cache.CacheException

Closes data streamer. This method is identical to calling `close(false)` method.

The method is invoked automatically on objects managed by the `try-with-resources` statement.

Specified by:
  `close` in interface `AutoCloseable`
Specified by:
  `close` in interface `IgniteDataStreamer<K,V>`
Throws:
  `javax.cache.CacheException` - If failed to close data streamer.

#### maxRemapCount

    public int maxRemapCount()

Returns:
  Max remap count.

#### maxRemapCount

    public void maxRemapCount(int maxRemapCnt)

Parameters:
  `maxRemapCnt` - New max remap count.

#### toString

    public String toString()

Overrides:
  `toString` in class `Object`

#### getDelay

    public long getDelay(TimeUnit unit)

Specified by:
  `getDelay` in interface `Delayed`

#### compareTo

    public int compareTo(Delayed o)

Specified by:
  `compareTo` in interface `Comparable<Delayed>`

2015 Copyright © Apache Software Foundation

Follow @ApacheIgnite

Ignite Fabric | : ver. 1.6.0
Release Date | : May 18 2016