org.infinispan.stream.impl

Class DistributedLongCacheStream<Original>

java.lang.Object

org.infinispan.stream.impl.AbstractCacheStream<Original,Long,LongStream,LongCacheStream>

org.infinispan.stream.impl.DistributedLongCacheStream<Original>

Type Parameters:

Original - original stream type

All Implemented Interfaces:

AutoCloseable, BaseStream<Long,LongStream>, LongStream, BaseCacheStream<Long,LongStream>, LongCacheStream

public class DistributedLongCacheStream<Original>
extends AbstractCacheStream<Original,Long,LongStream,LongCacheStream>
implements LongCacheStream

Implementation of LongStream that utilizes a lazily evaluated distributed back end execution. Note this class is only able to be created using CacheStream.mapToInt(ToIntFunction) or similar methods from the CacheStream interface.

Nested Class Summary

Nested classes/interfaces inherited from interface java.util.stream.LongStream

LongStream.Builder

Nested classes/interfaces inherited from interface org.infinispan.BaseCacheStream

BaseCacheStream.SegmentCompletionListener

Field Summary

Fields inherited from class org.infinispan.stream.impl.AbstractCacheStream

closeRunnable, cpm, distributedBatchSize, dm, executor, includeLoader, intermediateOperations, invocationContext, iteratorOperation, keyPartitioner, keysToFilter, localAddress, parallel, parallelDistribution, partition, registry, rehashAware, segmentCompletionListener, segmentsToFilter, stateTransferLock, supplier, timeout, timeoutUnit, toKeyFunction

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	DistributedLongCacheStream(AbstractCacheStream other) This constructor is to be used only when a user calls a map or flat map method changing to an IntStream from a CacheStream, Stream, DoubleStream, IntStream etc.

Method Summary

Modifier and Type	Method and Description
boolean	allMatch(LongPredicate predicate)
boolean	anyMatch(LongPredicate predicate)
DoubleCacheStream	asDoubleStream()
OptionalDouble	average()
CacheStream<Long>	boxed()
protected <R> DistributedCacheStream<Original,R>	cacheStream()
<R> R	collect(Supplier<R> supplier, ObjLongConsumer<R> accumulator, BiConsumer<R,R> combiner)
long	count()
LongCacheStream	disableRehashAware() Disables tracking of rehash events that could occur to the underlying cache.
LongCacheStream	distinct()
LongCacheStream	distributedBatchSize(int batchSize) Controls how many keys are returned from a remote node when using a stream terminal operation with a distributed cache to back this stream.
protected DistributedDoubleCacheStream<Original>	doubleCacheStream()
LongCacheStream	filter(LongPredicate predicate)
LongCacheStream	filter(SerializableLongPredicate predicate) Same as LongCacheStream.filter(LongPredicate) except that the LongPredicate must also implement Serializable.
LongCacheStream	filterKeys(Set<?> keys) Filters which entries are returned by only returning ones that map to the given key.
LongCacheStream	filterKeySegments(IntSet segments) Filters which entries are returned by what segment they are present in.
LongCacheStream	filterKeySegments(Set<Integer> segments) Filters which entries are returned by what segment they are present in.
OptionalLong	findAny()
OptionalLong	findFirst()
LongCacheStream	flatMap(LongFunction<? extends LongStream> mapper)
LongCacheStream	flatMap(SerializableLongFunction<? extends LongStream> mapper) Same as LongCacheStream.flatMap(LongFunction) except that the LongFunction must also implement Serializable.
void	forEach(LongConsumer action)
<K,V> void	forEach(ObjLongConsumer<Cache<K,V>> action) Same as LongStream.forEach(LongConsumer) except that it takes an ObjLongConsumer that provides access to the underlying Cache that is backing this stream.
void	forEach(SerializableLongConsumer action) Same as LongStream.forEach(LongConsumer) except that the LongConsumer must also implement Serializable.
<K,V> void	forEach(SerializableObjLongConsumer<Cache<K,V>> action) Same as LongCacheStream.forEach(ObjLongConsumer) except that the BiConsumer must also implement Serializable
void	forEachOrdered(LongConsumer action)
protected Log	getLog()
protected DistributedIntCacheStream<Original>	intCacheStream()
PrimitiveIterator.OfLong	iterator()
LongCacheStream	limit(long maxSize)
LongCacheStream	map(LongUnaryOperator mapper)
LongCacheStream	map(SerializableLongUnaryOperator mapper) Same as LongCacheStream.map(LongUnaryOperator) except that the LongUnaryOperator must also implement Serializable.
DoubleCacheStream	mapToDouble(LongToDoubleFunction mapper)
DoubleCacheStream	mapToDouble(SerializableLongToDoubleFunction mapper) Same as LongCacheStream.mapToDouble(LongToDoubleFunction) except that the LongToLongFunction must also implement Serializable.
IntCacheStream	mapToInt(LongToIntFunction mapper)
IntCacheStream	mapToInt(SerializableLongToIntFunction mapper) Same as LongCacheStream.mapToInt(LongToIntFunction) except that the LongToIntFunction must also implement Serializable.
<U> CacheStream<U>	mapToObj(LongFunction<? extends U> mapper)
<U> CacheStream<U>	mapToObj(SerializableLongFunction<? extends U> mapper) Same as LongCacheStream.mapToObj(LongFunction) except that the LongFunction must also implement Serializable.
OptionalLong	max()
OptionalLong	min()
boolean	noneMatch(LongPredicate predicate)
LongCacheStream	parallelDistribution() This would enable sending requests to all other remote nodes when a terminal operator is performed.
LongCacheStream	peek(LongConsumer action)
LongCacheStream	peek(SerializableLongConsumer action) Same as LongCacheStream.flatMap(LongFunction) except that the LongFunction must also implement Serializable.
OptionalLong	reduce(LongBinaryOperator op)
long	reduce(long identity, LongBinaryOperator op)
LongCacheStream	segmentCompletionListener(BaseCacheStream.SegmentCompletionListener listener) Allows registration of a segment completion listener that is notified when a segment has completed processing.
LongCacheStream	sequentialDistribution() This would disable sending requests to all other remote nodes compared to one at a time.
LongCacheStream	skip(long n)
LongCacheStream	sorted()
Spliterator.OfLong	spliterator()
long	sum()
LongSummaryStatistics	summaryStatistics()
LongCacheStream	timeout(long timeout, TimeUnit unit) Sets a given time to wait for a remote operation to respond by.
long[]	toArray()
protected LongCacheStream	unwrap()

Methods inherited from class org.infinispan.stream.impl.AbstractCacheStream

addIntermediateOperation, addIntermediateOperation, addIntermediateOperationMap, close, composeWithExceptions, isParallel, isPrimaryOwner, onClose, parallel, sequential, unordered

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.infinispan.LongCacheStream

allMatch, anyMatch, collect, noneMatch, onClose, parallel, reduce, reduce, sequential, unordered

Methods inherited from interface java.util.stream.LongStream

builder, concat, empty, generate, iterate, of, of, range, rangeClosed

Methods inherited from interface java.util.stream.BaseStream

close, isParallel

Constructor Detail

DistributedLongCacheStream

protected DistributedLongCacheStream(AbstractCacheStream other)

This constructor is to be used only when a user calls a map or flat map method changing to an IntStream from a CacheStream, Stream, DoubleStream, IntStream etc.

Parameters:

other - other instance of AbstractCacheStream to copy details from

Method Detail

getLog

protected Log getLog()

Specified by:

getLog in class AbstractCacheStream<Original,Long,LongStream,LongCacheStream>

unwrap

protected LongCacheStream unwrap()

Specified by:

unwrap in class AbstractCacheStream<Original,Long,LongStream,LongCacheStream>

filter

public LongCacheStream filter(LongPredicate predicate)

Description copied from interface: LongCacheStream

Specified by:

filter in interface LongStream

Specified by:

filter in interface LongCacheStream

Returns:

the new cache long stream

filter

public LongCacheStream filter(SerializableLongPredicate predicate)

Description copied from interface: LongCacheStream

Same as LongCacheStream.filter(LongPredicate) except that the LongPredicate must also implement Serializable.

This method will be used automatically by lambdas, which prevents users from having to manually cast to a Serializable lambda.

Specified by:

filter in interface LongCacheStream

Parameters:

predicate - a non-interfering, stateless predicate to apply to each element to determine if it should be included

Returns:

the new cache long stream

map

public LongCacheStream map(LongUnaryOperator mapper)

Description copied from interface: LongCacheStream

Specified by:

map in interface LongStream

Specified by:

map in interface LongCacheStream

Returns:

the new cache long stream

map

public LongCacheStream map(SerializableLongUnaryOperator mapper)

Description copied from interface: LongCacheStream

Same as LongCacheStream.map(LongUnaryOperator) except that the LongUnaryOperator must also implement Serializable.

This method will be used automatically by lambdas, which prevents users from having to manually cast to a Serializable lambda.

Specified by:

map in interface LongCacheStream

Parameters:

mapper - a non-interfering, stateless function to apply to each element

Returns:

the new cache long stream

mapToObj

public <U> CacheStream<U> mapToObj(LongFunction<? extends U> mapper)

Description copied from interface: LongCacheStream

Specified by:

mapToObj in interface LongStream

Specified by:

mapToObj in interface LongCacheStream

Returns:

the new cache stream

mapToObj

public <U> CacheStream<U> mapToObj(SerializableLongFunction<? extends U> mapper)

Description copied from interface: LongCacheStream

Same as LongCacheStream.mapToObj(LongFunction) except that the LongFunction must also implement Serializable.

This method will be used automatically by lambdas, which prevents users from having to manually cast to a Serializable lambda.

Specified by:

mapToObj in interface LongCacheStream

Type Parameters:

U - the element type of the new stream

Parameters:

mapper - a non-interfering, stateless function to apply to each element

Returns:

the new cache stream

mapToInt

public IntCacheStream mapToInt(LongToIntFunction mapper)

Description copied from interface: LongCacheStream

Specified by:

mapToInt in interface LongStream

Specified by:

mapToInt in interface LongCacheStream

Returns:

the new cache int stream

mapToInt

public IntCacheStream mapToInt(SerializableLongToIntFunction mapper)

Description copied from interface: LongCacheStream

Same as LongCacheStream.mapToInt(LongToIntFunction) except that the LongToIntFunction must also implement Serializable.

This method will be used automatically by lambdas, which prevents users from having to manually cast to a Serializable lambda.

Specified by:

mapToInt in interface LongCacheStream

Parameters:

mapper - a non-interfering, stateless function to apply to each element

Returns:

the new cache int stream

mapToDouble

public DoubleCacheStream mapToDouble(LongToDoubleFunction mapper)

Description copied from interface: LongCacheStream

Specified by:

mapToDouble in interface LongStream

Specified by:

mapToDouble in interface LongCacheStream

Returns:

the new cache double stream

mapToDouble

public DoubleCacheStream mapToDouble(SerializableLongToDoubleFunction mapper)

Description copied from interface: LongCacheStream

Same as LongCacheStream.mapToDouble(LongToDoubleFunction) except that the LongToLongFunction must also implement Serializable.

This method will be used automatically by lambdas, which prevents users from having to manually cast to a Serializable lambda.

Specified by:

mapToDouble in interface LongCacheStream

Parameters:

mapper - a non-interfering, stateless function to apply to each element

Returns:

the new cache double stream

flatMap

public LongCacheStream flatMap(LongFunction<? extends LongStream> mapper)

Description copied from interface: LongCacheStream

Specified by:

flatMap in interface LongStream

Specified by:

flatMap in interface LongCacheStream

Returns:

the new cache long stream

flatMap

public LongCacheStream flatMap(SerializableLongFunction<? extends LongStream> mapper)

Description copied from interface: LongCacheStream

Same as LongCacheStream.flatMap(LongFunction) except that the LongFunction must also implement Serializable.

This method will be used automatically by lambdas, which prevents users from having to manually cast to a Serializable lambda.

Specified by:

flatMap in interface LongCacheStream

Parameters:

mapper - a non-interfering, stateless function to apply to each element which produces a LongStream of new values

Returns:

the new cache long stream

distinct

public LongCacheStream distinct()

Description copied from interface: LongCacheStream

Specified by:

distinct in interface LongStream

Specified by:

distinct in interface LongCacheStream

Returns:

the new cache long stream

sorted

public LongCacheStream sorted()

Description copied from interface: LongCacheStream

Specified by:

sorted in interface LongStream

Specified by:

sorted in interface LongCacheStream

Returns:

the new cache long stream

peek

public LongCacheStream peek(LongConsumer action)

Description copied from interface: LongCacheStream

Specified by:

peek in interface LongStream

Specified by:

peek in interface LongCacheStream

Returns:

the new cache long stream

peek

public LongCacheStream peek(SerializableLongConsumer action)

Description copied from interface: LongCacheStream

Same as LongCacheStream.flatMap(LongFunction) except that the LongFunction must also implement Serializable.

This method will be used automatically by lambdas, which prevents users from having to manually cast to a Serializable lambda.

Specified by:

peek in interface LongCacheStream

Parameters:

action - a non-interfering action to perform on the elements as they are consumed from the stream

Returns:

the new cache long stream

asDoubleStream

public DoubleCacheStream asDoubleStream()

Description copied from interface: LongCacheStream

Specified by:

asDoubleStream in interface LongStream

Specified by:

asDoubleStream in interface LongCacheStream

Returns:

the cache double stream

boxed

public CacheStream<Long> boxed()

Description copied from interface: LongCacheStream

Specified by:

boxed in interface LongStream

Specified by:

boxed in interface LongCacheStream

Returns:

the new cache stream containing longs

limit

public LongCacheStream limit(long maxSize)

Description copied from interface: LongCacheStream

Specified by:

limit in interface LongStream

Specified by:

limit in interface LongCacheStream

Returns:

the new cache long stream

skip

public LongCacheStream skip(long n)

Description copied from interface: LongCacheStream

Specified by:

skip in interface LongStream

Specified by:

skip in interface LongCacheStream

Returns:

the new cache long stream

forEach

public void forEach(LongConsumer action)

Specified by:

forEach in interface LongStream

forEach

public void forEach(SerializableLongConsumer action)

Description copied from interface: LongCacheStream

Same as LongStream.forEach(LongConsumer) except that the LongConsumer must also implement Serializable.

This method will be used automatically by lambdas, which prevents users from having to manually cast to a Serializable lambda.

Specified by:

forEach in interface LongCacheStream

Parameters:

action - a non-interfering action to perform on the elements

forEach

public <K,V> void forEach(ObjLongConsumer<Cache<K,V>> action)

Description copied from interface: LongCacheStream

Same as LongStream.forEach(LongConsumer) except that it takes an ObjLongConsumer that provides access to the underlying Cache that is backing this stream.

Note that the CacheAware interface is not supported for injection using this method as the cache is provided in the consumer directly.

Specified by:

forEach in interface LongCacheStream

Type Parameters:

K - key type of the cache

V - value type of the cache

Parameters:

action - consumer to be ran for each element in the stream

forEach

public <K,V> void forEach(SerializableObjLongConsumer<Cache<K,V>> action)

Description copied from interface: LongCacheStream

Same as LongCacheStream.forEach(ObjLongConsumer) except that the BiConsumer must also implement Serializable

Specified by:

forEach in interface LongCacheStream

Type Parameters:

K - key type of the cache

V - value type of the cache

Parameters:

action - consumer to be ran for each element in the stream

forEachOrdered

public void forEachOrdered(LongConsumer action)

Specified by:

forEachOrdered in interface LongStream

toArray

public long[] toArray()

Specified by:

toArray in interface LongStream

reduce

public long reduce(long identity,
                   LongBinaryOperator op)

Specified by:

reduce in interface LongStream

reduce

public OptionalLong reduce(LongBinaryOperator op)

Specified by:

reduce in interface LongStream

collect

public <R> R collect(Supplier<R> supplier,
                     ObjLongConsumer<R> accumulator,
                     BiConsumer<R,R> combiner)

Specified by:

collect in interface LongStream

sum

public long sum()

Specified by:

sum in interface LongStream

min

public OptionalLong min()

Specified by:

min in interface LongStream

max

public OptionalLong max()

Specified by:

max in interface LongStream

average

public OptionalDouble average()

Specified by:

average in interface LongStream

summaryStatistics

public LongSummaryStatistics summaryStatistics()

Specified by:

summaryStatistics in interface LongStream

anyMatch

public boolean anyMatch(LongPredicate predicate)

Specified by:

anyMatch in interface LongStream

allMatch

public boolean allMatch(LongPredicate predicate)

Specified by:

allMatch in interface LongStream

noneMatch

public boolean noneMatch(LongPredicate predicate)

Specified by:

noneMatch in interface LongStream

findFirst

public OptionalLong findFirst()

Specified by:

findFirst in interface LongStream

findAny

public OptionalLong findAny()

Specified by:

findAny in interface LongStream

iterator

public PrimitiveIterator.OfLong iterator()

Specified by:

iterator in interface BaseStream<Long,LongStream>

Specified by:

iterator in interface LongStream

spliterator

public Spliterator.OfLong spliterator()

Specified by:

spliterator in interface BaseStream<Long,LongStream>

Specified by:

spliterator in interface LongStream

count

public long count()

Specified by:

count in interface LongStream

sequentialDistribution

public LongCacheStream sequentialDistribution()

Description copied from interface: LongCacheStream

This would disable sending requests to all other remote nodes compared to one at a time. This can reduce memory pressure on the originator node at the cost of performance.

Parallel distribution is enabled by default except for CacheStream.iterator() and CacheStream.spliterator()

Specified by:

sequentialDistribution in interface BaseCacheStream<Long,LongStream>

Specified by:

sequentialDistribution in interface LongCacheStream

Returns:

a stream with parallel distribution disabled.

parallelDistribution

public LongCacheStream parallelDistribution()

Description copied from interface: BaseCacheStream

This would enable sending requests to all other remote nodes when a terminal operator is performed. This requires additional overhead as it must process results concurrently from various nodes, but should perform faster in the majority of cases.

Parallel distribution is enabled by default except for CacheStream.iterator() and CacheStream.spliterator()

Specified by:

parallelDistribution in interface BaseCacheStream<Long,LongStream>

Specified by:

parallelDistribution in interface LongCacheStream

Returns:

a stream with parallel distribution enabled.

filterKeySegments

public LongCacheStream filterKeySegments(Set<Integer> segments)

Description copied from interface: LongCacheStream

Filters which entries are returned by what segment they are present in. This method can be substantially more efficient than using a regular CacheStream.filter(Predicate) method as this can control what nodes are asked for data and what entries are read from the underlying CacheStore if present.

Specified by:

filterKeySegments in interface BaseCacheStream<Long,LongStream>

Specified by:

filterKeySegments in interface LongCacheStream

Parameters:

segments - The segments to use for this stream operation. Any segments not in this set will be ignored.

Returns:

a stream with the segments filtered.

filterKeySegments

public LongCacheStream filterKeySegments(IntSet segments)

Description copied from interface: BaseCacheStream

Filters which entries are returned by what segment they are present in. This method can be substantially more efficient than using a regular CacheStream.filter(Predicate) method as this can control what nodes are asked for data and what entries are read from the underlying CacheStore if present.

Specified by:

filterKeySegments in interface BaseCacheStream<Long,LongStream>

Parameters:

segments - The segments to use for this stream operation. Any segments not in this set will be ignored.

Returns:

a stream with the segments filtered.

filterKeys

public LongCacheStream filterKeys(Set<?> keys)

Description copied from interface: LongCacheStream

Filters which entries are returned by only returning ones that map to the given key. This method will be faster than a regular CacheStream.filter(Predicate) if the filter is holding references to the same keys.

Specified by:

filterKeys in interface BaseCacheStream<Long,LongStream>

Specified by:

filterKeys in interface LongCacheStream

Parameters:

keys - The keys that this stream will only operate on.

Returns:

a stream with the keys filtered.

distributedBatchSize

public LongCacheStream distributedBatchSize(int batchSize)

Description copied from interface: LongCacheStream

Controls how many keys are returned from a remote node when using a stream terminal operation with a distributed cache to back this stream. This value is ignored when terminal operators that don't track keys are used. Key tracking terminal operators are CacheStream.iterator(), CacheStream.spliterator(), CacheStream.forEach(Consumer). Please see those methods for additional information on how this value may affect them.

This value may be used in the case of a a terminal operator that doesn't track keys if an intermediate operation is performed that requires bringing keys locally to do computations. Examples of such intermediate operations are CacheStream.sorted(), CacheStream.sorted(Comparator), CacheStream.distinct(), CacheStream.limit(long), CacheStream.skip(long)

This value is always ignored when this stream is backed by a cache that is not distributed as all values are already local.

Specified by:

distributedBatchSize in interface BaseCacheStream<Long,LongStream>

Specified by:

distributedBatchSize in interface LongCacheStream

Parameters:

batchSize - The size of each batch. This defaults to the state transfer chunk size.

Returns:

a stream with the batch size updated

segmentCompletionListener

public LongCacheStream segmentCompletionListener(BaseCacheStream.SegmentCompletionListener listener)

Description copied from interface: LongCacheStream

Allows registration of a segment completion listener that is notified when a segment has completed processing. If the terminal operator has a short circuit this listener may never be called.

This method is designed for the sole purpose of use with the CacheStream.iterator() to allow for a user to track completion of segments as they are returned from the iterator. Behavior of other methods is not specified. Please see CacheStream.iterator() for more information.

Multiple listeners may be registered upon multiple invocations of this method. The ordering of notified listeners is not specified.

This is only used if this stream did not invoke BaseCacheStream.disableRehashAware() and has no flat map based operations. If this is done no segments will be notified.

Specified by:

segmentCompletionListener in interface BaseCacheStream<Long,LongStream>

Specified by:

segmentCompletionListener in interface LongCacheStream

Parameters:

listener - The listener that will be called back as segments are completed.

Returns:

a stream with the listener registered.

disableRehashAware

public LongCacheStream disableRehashAware()

Description copied from interface: LongCacheStream

Disables tracking of rehash events that could occur to the underlying cache. If a rehash event occurs while a terminal operation is being performed it is possible for some values that are in the cache to not be found. Note that you will never have an entry duplicated when rehash awareness is disabled, only lost values.

Most terminal operations will run faster with rehash awareness disabled even without a rehash occuring. However if a rehash occurs with this disabled be prepared to possibly receive only a subset of values.

Specified by:

disableRehashAware in interface BaseCacheStream<Long,LongStream>

Specified by:

disableRehashAware in interface LongCacheStream

Returns:

a stream with rehash awareness disabled.

timeout

public LongCacheStream timeout(long timeout,
                               TimeUnit unit)

Description copied from interface: LongCacheStream

Sets a given time to wait for a remote operation to respond by. This timeout does nothing if the terminal operation does not go remote.

If a timeout does occur then a TimeoutException is thrown from the terminal operation invoking thread or on the next call to the Iterator or Spliterator.

Note that if a rehash occurs this timeout value is reset for the subsequent retry if rehash aware is enabled.

Specified by:

timeout in interface BaseCacheStream<Long,LongStream>

Specified by:

timeout in interface LongCacheStream

Parameters:

timeout - the maximum time to wait

unit - the time unit of the timeout argument

Returns:

a stream with the timeout set

cacheStream

protected <R> DistributedCacheStream<Original,R> cacheStream()

doubleCacheStream

protected DistributedDoubleCacheStream<Original> doubleCacheStream()

intCacheStream

protected DistributedIntCacheStream<Original> intCacheStream()