org.infinispan.stream.impl.local

Class LocalIntCacheStream

java.lang.Object

org.infinispan.stream.impl.local.AbstractLocalCacheStream<Integer,IntStream,IntCacheStream>

org.infinispan.stream.impl.local.LocalIntCacheStream

All Implemented Interfaces:

AutoCloseable, BaseStream<Integer,IntStream>, IntStream, BaseCacheStream<Integer,IntStream>, IntCacheStream

public class LocalIntCacheStream
extends AbstractLocalCacheStream<Integer,IntStream,IntCacheStream>
implements IntCacheStream

IntStream that wraps a given stream to allow for additional functionality such as injection of values into various operations

Nested Class Summary

Nested classes/interfaces inherited from class org.infinispan.stream.impl.local.AbstractLocalCacheStream

AbstractLocalCacheStream.StreamSupplier<T,S extends BaseStream<T,S>>

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

IntStream.Builder

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

BaseCacheStream.SegmentCompletionListener

Field Summary

Fields inherited from class org.infinispan.stream.impl.local.AbstractLocalCacheStream

intermediateOperations, keysToFilter, onCloseRunnables, parallel, registry, segmentsToFilter, streamSupplier

Constructor Summary

Constructors

Constructor and Description
LocalIntCacheStream(AbstractLocalCacheStream.StreamSupplier<Integer,IntStream> streamSupplier, boolean parallel, ComponentRegistry registry)

Method Summary

Modifier and Type	Method and Description
boolean	allMatch(IntPredicate predicate)
boolean	anyMatch(IntPredicate predicate)
DoubleCacheStream	asDoubleStream()
LongCacheStream	asLongStream()
OptionalDouble	average()
LocalCacheStream<Integer>	boxed()
<R> R	collect(Supplier<R> supplier, ObjIntConsumer<R> accumulator, BiConsumer<R,R> combiner)
long	count()
LocalIntCacheStream	disableRehashAware() Disables tracking of rehash events that could occur to the underlying cache.
LocalIntCacheStream	distinct()
LocalIntCacheStream	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.
LocalIntCacheStream	filter(IntPredicate predicate)
LocalIntCacheStream	filterKeys(Set<?> keys) Filters which entries are returned by only returning ones that map to the given key.
LocalIntCacheStream	filterKeySegments(IntSet segments) Filters which entries are returned by what segment they are present in.
LocalIntCacheStream	filterKeySegments(Set<Integer> segments) Filters which entries are returned by what segment they are present in.
OptionalInt	findAny()
OptionalInt	findFirst()
LocalIntCacheStream	flatMap(IntFunction<? extends IntStream> mapper)
void	forEach(IntConsumer action)
<K,V> void	forEach(ObjIntConsumer<Cache<K,V>> action) Same as IntStream.forEach(IntConsumer) except that it takes an ObjIntConsumer that provides access to the underlying Cache that is backing this stream.
void	forEachOrdered(IntConsumer action)
PrimitiveIterator.OfInt	iterator()
LocalIntCacheStream	limit(long maxSize)
LocalIntCacheStream	map(IntUnaryOperator mapper)
LocalDoubleCacheStream	mapToDouble(IntToDoubleFunction mapper)
LocalLongCacheStream	mapToLong(IntToLongFunction mapper)
<U> LocalCacheStream<U>	mapToObj(IntFunction<? extends U> mapper)
OptionalInt	max()
OptionalInt	min()
boolean	noneMatch(IntPredicate predicate)
LocalIntCacheStream	parallelDistribution() This would enable sending requests to all other remote nodes when a terminal operator is performed.
LocalIntCacheStream	peek(IntConsumer action)
OptionalInt	reduce(IntBinaryOperator op)
int	reduce(int identity, IntBinaryOperator op)
LocalIntCacheStream	segmentCompletionListener(BaseCacheStream.SegmentCompletionListener listener) Allows registration of a segment completion listener that is notified when a segment has completed processing.
LocalIntCacheStream	sequentialDistribution() This would disable sending requests to all other remote nodes compared to one at a time.
LocalIntCacheStream	skip(long n)
LocalIntCacheStream	sorted()
Spliterator.OfInt	spliterator()
int	sum()
IntSummaryStatistics	summaryStatistics()
LocalIntCacheStream	timeout(long timeout, TimeUnit unit) Sets a given time to wait for a remote operation to respond by.
int[]	toArray()

Methods inherited from class org.infinispan.stream.impl.local.AbstractLocalCacheStream

close, createStream, isParallel, 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.IntCacheStream

allMatch, anyMatch, collect, filter, flatMap, forEach, forEach, map, mapToDouble, mapToLong, mapToObj, noneMatch, onClose, parallel, peek, reduce, reduce, sequential, unordered

Methods inherited from interface java.util.stream.IntStream

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

Methods inherited from interface java.util.stream.BaseStream

close, isParallel

Constructor Detail

LocalIntCacheStream

public LocalIntCacheStream(AbstractLocalCacheStream.StreamSupplier<Integer,IntStream> streamSupplier,
                           boolean parallel,
                           ComponentRegistry registry)

Method Detail

filter

public LocalIntCacheStream filter(IntPredicate predicate)

Description copied from interface: IntCacheStream

Specified by:

filter in interface IntStream

Specified by:

filter in interface IntCacheStream

Returns:

the new cache int stream

map

public LocalIntCacheStream map(IntUnaryOperator mapper)

Description copied from interface: IntCacheStream

Specified by:

map in interface IntStream

Specified by:

map in interface IntCacheStream

Returns:

the new cache int stream

mapToObj

public <U> LocalCacheStream<U> mapToObj(IntFunction<? extends U> mapper)

Description copied from interface: IntCacheStream

Specified by:

mapToObj in interface IntStream

Specified by:

mapToObj in interface IntCacheStream

Returns:

the new cache stream

mapToLong

public LocalLongCacheStream mapToLong(IntToLongFunction mapper)

Description copied from interface: IntCacheStream

Specified by:

mapToLong in interface IntStream

Specified by:

mapToLong in interface IntCacheStream

Returns:

the new cache long stream

mapToDouble

public LocalDoubleCacheStream mapToDouble(IntToDoubleFunction mapper)

Description copied from interface: IntCacheStream

Specified by:

mapToDouble in interface IntStream

Specified by:

mapToDouble in interface IntCacheStream

Returns:

the new cache double stream

flatMap

public LocalIntCacheStream flatMap(IntFunction<? extends IntStream> mapper)

Description copied from interface: IntCacheStream

Specified by:

flatMap in interface IntStream

Specified by:

flatMap in interface IntCacheStream

Returns:

the new cache int stream

distinct

public LocalIntCacheStream distinct()

Description copied from interface: IntCacheStream

Specified by:

distinct in interface IntStream

Specified by:

distinct in interface IntCacheStream

Returns:

the new cache int stream

sorted

public LocalIntCacheStream sorted()

Description copied from interface: IntCacheStream

Specified by:

sorted in interface IntStream

Specified by:

sorted in interface IntCacheStream

Returns:

the new cache int stream

peek

public LocalIntCacheStream peek(IntConsumer action)

Description copied from interface: IntCacheStream

Specified by:

peek in interface IntStream

Specified by:

peek in interface IntCacheStream

Returns:

the new cache int stream

limit

public LocalIntCacheStream limit(long maxSize)

Description copied from interface: IntCacheStream

Specified by:

limit in interface IntStream

Specified by:

limit in interface IntCacheStream

Returns:

the new cache int stream

skip

public LocalIntCacheStream skip(long n)

Description copied from interface: IntCacheStream

Specified by:

skip in interface IntStream

Specified by:

skip in interface IntCacheStream

Returns:

the new cache int stream

forEach

public void forEach(IntConsumer action)

Specified by:

forEach in interface IntStream

forEach

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

Description copied from interface: IntCacheStream

Same as IntStream.forEach(IntConsumer) except that it takes an ObjIntConsumer 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 IntCacheStream

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(IntConsumer action)

Specified by:

forEachOrdered in interface IntStream

toArray

public int[] toArray()

Specified by:

toArray in interface IntStream

reduce

public int reduce(int identity,
                  IntBinaryOperator op)

Specified by:

reduce in interface IntStream

reduce

public OptionalInt reduce(IntBinaryOperator op)

Specified by:

reduce in interface IntStream

collect

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

Specified by:

collect in interface IntStream

sum

public int sum()

Specified by:

sum in interface IntStream

min

public OptionalInt min()

Specified by:

min in interface IntStream

max

public OptionalInt max()

Specified by:

max in interface IntStream

count

public long count()

Specified by:

count in interface IntStream

average

public OptionalDouble average()

Specified by:

average in interface IntStream

summaryStatistics

public IntSummaryStatistics summaryStatistics()

Specified by:

summaryStatistics in interface IntStream

anyMatch

public boolean anyMatch(IntPredicate predicate)

Specified by:

anyMatch in interface IntStream

allMatch

public boolean allMatch(IntPredicate predicate)

Specified by:

allMatch in interface IntStream

noneMatch

public boolean noneMatch(IntPredicate predicate)

Specified by:

noneMatch in interface IntStream

findFirst

public OptionalInt findFirst()

Specified by:

findFirst in interface IntStream

findAny

public OptionalInt findAny()

Specified by:

findAny in interface IntStream

asLongStream

public LongCacheStream asLongStream()

Description copied from interface: IntCacheStream

Specified by:

asLongStream in interface IntStream

Specified by:

asLongStream in interface IntCacheStream

Returns:

the cache long stream

asDoubleStream

public DoubleCacheStream asDoubleStream()

Description copied from interface: IntCacheStream

Specified by:

asDoubleStream in interface IntStream

Specified by:

asDoubleStream in interface IntCacheStream

Returns:

the cache double stream

boxed

public LocalCacheStream<Integer> boxed()

Description copied from interface: IntCacheStream

Specified by:

boxed in interface IntStream

Specified by:

boxed in interface IntCacheStream

Returns:

the new cache stream containing integers

iterator

public PrimitiveIterator.OfInt iterator()

Specified by:

iterator in interface BaseStream<Integer,IntStream>

Specified by:

iterator in interface IntStream

spliterator

public Spliterator.OfInt spliterator()

Specified by:

spliterator in interface BaseStream<Integer,IntStream>

Specified by:

spliterator in interface IntStream

sequentialDistribution

public LocalIntCacheStream sequentialDistribution()

Description copied from interface: IntCacheStream

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<Integer,IntStream>

Specified by:

sequentialDistribution in interface IntCacheStream

Returns:

a stream with parallel distribution disabled.

parallelDistribution

public LocalIntCacheStream 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<Integer,IntStream>

Specified by:

parallelDistribution in interface IntCacheStream

Returns:

a stream with parallel distribution enabled.

filterKeySegments

public LocalIntCacheStream filterKeySegments(Set<Integer> segments)

Description copied from interface: IntCacheStream

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<Integer,IntStream>

Specified by:

filterKeySegments in interface IntCacheStream

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 LocalIntCacheStream 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<Integer,IntStream>

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 LocalIntCacheStream filterKeys(Set<?> keys)

Description copied from interface: IntCacheStream

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<Integer,IntStream>

Specified by:

filterKeys in interface IntCacheStream

Parameters:

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

Returns:

a stream with the keys filtered.

distributedBatchSize

public LocalIntCacheStream distributedBatchSize(int batchSize)

Description copied from interface: IntCacheStream

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<Integer,IntStream>

Specified by:

distributedBatchSize in interface IntCacheStream

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 LocalIntCacheStream segmentCompletionListener(BaseCacheStream.SegmentCompletionListener listener)

Description copied from interface: IntCacheStream

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<Integer,IntStream>

Specified by:

segmentCompletionListener in interface IntCacheStream

Parameters:

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

Returns:

a stream with the listener registered.

disableRehashAware

public LocalIntCacheStream disableRehashAware()

Description copied from interface: IntCacheStream

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<Integer,IntStream>

Specified by:

disableRehashAware in interface IntCacheStream

Returns:

a stream with rehash awareness disabled.

timeout

public LocalIntCacheStream timeout(long timeout,
                                   TimeUnit unit)

Description copied from interface: IntCacheStream

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<Integer,IntStream>

Specified by:

timeout in interface IntCacheStream

Parameters:

timeout - the maximum time to wait

unit - the time unit of the timeout argument

Returns:

a stream with the timeout set