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

Modifier and Type	Method and Description
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)
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()
int	maxRemapCount()
void	maxRemapCount(int maxRemapCnt)
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.

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.

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.

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>