Package org.infinispan.util

Class AbstractDelegatingCacheStream<R>

java.lang.Object

org.infinispan.util.AbstractDelegatingCacheStream<R>

All Implemented Interfaces:

AutoCloseable, BaseStream<R,Stream<R>>, Stream<R>, BaseCacheStream<R,Stream<R>>, CacheStream<R>

public class AbstractDelegatingCacheStream<R>
extends Object
implements CacheStream<R>

Delegate that forwards all the of the method calls to the underlying cache stream. This can be useful to intercept a given method call. Note that primitive stream operations are not intercepted (including those from BaseStream, however defined BaseCacheStream methods would be.

Nested Class Summary

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

BaseCacheStream.SegmentCompletionListener

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

Stream.Builder<T extends Object>

Field Summary

Fields

Modifier and Type	Field	Description
protected CacheStream<?>	underlyingStream

Constructor Summary

Constructors

Constructor	Description
AbstractDelegatingCacheStream(CacheStream<R> stream)

Method Summary

Modifier and Type	Method	Description
boolean	allMatch(Predicate<? super R> predicate)
boolean	anyMatch(Predicate<? super R> predicate)
void	close()
<R1> R1	collect(Supplier<R1> supplier, BiConsumer<R1,? super R> accumulator, BiConsumer<R1,R1> combiner)
<R1,A> R1	collect(Collector<? super R,A,R1> collector)
long	count()
AbstractDelegatingCacheStream<R>	disableRehashAware()	Disables tracking of rehash events that could occur to the underlying cache.
CacheStream<R>	distinct()
AbstractDelegatingCacheStream<R>	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.
CacheStream<R>	filter(Predicate<? super R> predicate)
AbstractDelegatingCacheStream<R>	filterKeys(Set<?> keys)	Filters which entries are returned by only returning ones that map to the given key.
AbstractDelegatingCacheStream<R>	filterKeySegments(Set<Integer> segments)	Filters which entries are returned by what segment they are present in.
AbstractDelegatingCacheStream<R>	filterKeySegments(org.infinispan.commons.util.IntSet segments)	Filters which entries are returned by what segment they are present in.
Optional<R>	findAny()
Optional<R>	findFirst()
<R1> CacheStream<R1>	flatMap(Function<? super R,? extends Stream<? extends R1>> mapper)
DoubleCacheStream	flatMapToDouble(Function<? super R,? extends DoubleStream> mapper)
IntCacheStream	flatMapToInt(Function<? super R,? extends IntStream> mapper)
LongCacheStream	flatMapToLong(Function<? super R,? extends LongStream> mapper)
<K,V> void	forEach(BiConsumer<Cache<K,V>,? super R> action)	Same as CacheStream.forEach(Consumer) except that it takes a BiConsumer that provides access to the underlying Cache that is backing this stream.
void	forEach(Consumer<? super R> action)
void	forEachOrdered(Consumer<? super R> action)
boolean	isParallel()
Iterator<R>	iterator()
CacheStream<R>	limit(long maxSize)
<R1> CacheStream<R1>	map(Function<? super R,? extends R1> mapper)
DoubleCacheStream	mapToDouble(ToDoubleFunction<? super R> mapper)
IntCacheStream	mapToInt(ToIntFunction<? super R> mapper)
LongCacheStream	mapToLong(ToLongFunction<? super R> mapper)
Optional<R>	max(Comparator<? super R> comparator)
Optional<R>	min(Comparator<? super R> comparator)
boolean	noneMatch(Predicate<? super R> predicate)
CacheStream<R>	onClose(Runnable closeHandler)
CacheStream<R>	parallel()
AbstractDelegatingCacheStream<R>	parallelDistribution()	This would enable sending requests to all other remote nodes when a terminal operator is performed.
CacheStream<R>	peek(Consumer<? super R> action)
Optional<R>	reduce(BinaryOperator<R> accumulator)
R	reduce(R identity, BinaryOperator<R> accumulator)
<U> U	reduce(U identity, BiFunction<U,? super R,U> accumulator, BinaryOperator<U> combiner)
AbstractDelegatingCacheStream<R>	segmentCompletionListener(BaseCacheStream.SegmentCompletionListener listener)	Allows registration of a segment completion listener that is notified when a segment has completed processing.
CacheStream<R>	sequential()
AbstractDelegatingCacheStream<R>	sequentialDistribution()	This would disable sending requests to all other remote nodes compared to one at a time.
CacheStream<R>	skip(long n)
CacheStream<R>	sorted()
CacheStream<R>	sorted(Comparator<? super R> comparator)
Spliterator<R>	spliterator()
AbstractDelegatingCacheStream<R>	timeout(long timeout, TimeUnit unit)	Sets a given time to wait for a remote operation to respond by.
Object[]	toArray()
<A> A[]	toArray(IntFunction<A[]> generator)
CacheStream<R>	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.CacheStream

allMatch, anyMatch, collect, collect, collect, filter, flatMap, flatMapToDouble, flatMapToInt, flatMapToLong, forEach, forEach, map, mapToDouble, mapToInt, mapToLong, max, min, noneMatch, peek, reduce, reduce, reduce, sorted, toArray

Methods inherited from interface java.util.stream.Stream

dropWhile, takeWhile

Field Detail

underlyingStream

protected CacheStream<?> underlyingStream

Constructor Detail

AbstractDelegatingCacheStream

public AbstractDelegatingCacheStream(CacheStream<R> stream)

Method Detail

mapToInt

public IntCacheStream mapToInt(ToIntFunction<? super R> mapper)

Description copied from interface: CacheStream

Specified by:

mapToInt in interface CacheStream<R>

Specified by:

mapToInt in interface Stream<R>

Parameters:

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

Returns:

the new int cache stream

mapToLong

public LongCacheStream mapToLong(ToLongFunction<? super R> mapper)

Description copied from interface: CacheStream

Specified by:

mapToLong in interface CacheStream<R>

Specified by:

mapToLong in interface Stream<R>

Parameters:

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

Returns:

the new long cache stream

mapToDouble

public DoubleCacheStream mapToDouble(ToDoubleFunction<? super R> mapper)

Description copied from interface: CacheStream

Specified by:

mapToDouble in interface CacheStream<R>

Specified by:

mapToDouble in interface Stream<R>

Parameters:

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

Returns:

the new double cache stream

flatMapToInt

public IntCacheStream flatMapToInt(Function<? super R,? extends IntStream> mapper)

Description copied from interface: CacheStream

Specified by:

flatMapToInt in interface CacheStream<R>

Specified by:

flatMapToInt in interface Stream<R>

Returns:

the new cache stream

flatMapToLong

public LongCacheStream flatMapToLong(Function<? super R,? extends LongStream> mapper)

Description copied from interface: CacheStream

Specified by:

flatMapToLong in interface CacheStream<R>

Specified by:

flatMapToLong in interface Stream<R>

Returns:

the new cache stream

flatMapToDouble

public DoubleCacheStream flatMapToDouble(Function<? super R,? extends DoubleStream> mapper)

Description copied from interface: CacheStream

Specified by:

flatMapToDouble in interface CacheStream<R>

Specified by:

flatMapToDouble in interface Stream<R>

Returns:

the new cache stream

sequentialDistribution

public AbstractDelegatingCacheStream<R> sequentialDistribution()

Description copied from interface: CacheStream

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<R,Stream<R>>

Specified by:

sequentialDistribution in interface CacheStream<R>

Returns:

a stream with parallel distribution disabled.

parallelDistribution

public AbstractDelegatingCacheStream<R> 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<R,Stream<R>>

Specified by:

parallelDistribution in interface CacheStream<R>

Returns:

a stream with parallel distribution enabled.

filterKeySegments

public AbstractDelegatingCacheStream<R> filterKeySegments(Set<Integer> segments)

Description copied from interface: CacheStream

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<R,Stream<R>>

Specified by:

filterKeySegments in interface CacheStream<R>

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 AbstractDelegatingCacheStream<R> filterKeySegments(org.infinispan.commons.util.IntSet segments)

Description copied from interface: CacheStream

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<R,Stream<R>>

Specified by:

filterKeySegments in interface CacheStream<R>

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

Description copied from interface: CacheStream

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<R,Stream<R>>

Specified by:

filterKeys in interface CacheStream<R>

Parameters:

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

Returns:

a stream with the keys filtered.

distributedBatchSize

public AbstractDelegatingCacheStream<R> distributedBatchSize(int batchSize)

Description copied from interface: CacheStream

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<R,Stream<R>>

Specified by:

distributedBatchSize in interface CacheStream<R>

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

Description copied from interface: CacheStream

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<R,Stream<R>>

Specified by:

segmentCompletionListener in interface CacheStream<R>

Parameters:

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

Returns:

a stream with the listener registered.

disableRehashAware

public AbstractDelegatingCacheStream<R> disableRehashAware()

Description copied from interface: CacheStream

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<R,Stream<R>>

Specified by:

disableRehashAware in interface CacheStream<R>

Returns:

a stream with rehash awareness disabled.

timeout

public AbstractDelegatingCacheStream<R> timeout(long timeout,
                                                TimeUnit unit)

Description copied from interface: CacheStream

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<R,Stream<R>>

Specified by:

timeout in interface CacheStream<R>

Parameters:

timeout - the maximum time to wait

unit - the time unit of the timeout argument

Returns:

a stream with the timeout set

forEach

public void forEach(Consumer<? super R> action)

Description copied from interface: CacheStream

This operation is performed remotely on the node that is the primary owner for the key tied to the entry(s) in this stream.

NOTE: This method while being rehash aware has the lowest consistency of all of the operators. This operation will be performed on every entry at least once in the cluster, as long as the originator doesn't go down while it is being performed. This is due to how the distributed action is performed. Essentially the CacheStream.distributedBatchSize(int) value controls how many elements are processed per node at a time when rehash is enabled. After those are complete the keys are sent to the originator to confirm that those were processed. If that node goes down during/before the response those keys will be processed a second time.

It is possible to have the cache local to each node injected into this instance if the provided Consumer also implements the CacheAware interface. This method will be invoked before the consumer accept() method is invoked.

This method is ran distributed by default with a distributed backing cache. However if you wish for this operation to run locally you can use the stream().iterator().forEachRemaining(action) for a single threaded variant. If you wish to have a parallel variant you can use StreamSupport.stream(Spliterator, boolean) passing in the spliterator from the stream. In either case remember you must close the stream after you are done processing the iterator or spliterator..

Specified by:

forEach in interface CacheStream<R>

Specified by:

forEach in interface Stream<R>

Parameters:

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

forEach

public <K,V> void forEach(BiConsumer<Cache<K,V>,? super R> action)

Description copied from interface: CacheStream

Same as CacheStream.forEach(Consumer) except that it takes a BiConsumer 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 CacheStream<R>

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(Consumer<? super R> action)

Specified by:

forEachOrdered in interface Stream<R>

toArray

public Object[] toArray()

Specified by:

toArray in interface Stream<R>

toArray

public <A> A[] toArray(IntFunction<A[]> generator)

Specified by:

toArray in interface Stream<R>

reduce

public R reduce(R identity,
                BinaryOperator<R> accumulator)

Specified by:

reduce in interface Stream<R>

reduce

public Optional<R> reduce(BinaryOperator<R> accumulator)

Specified by:

reduce in interface Stream<R>

reduce

public <U> U reduce(U identity,
                    BiFunction<U,? super R,U> accumulator,
                    BinaryOperator<U> combiner)

Specified by:

reduce in interface Stream<R>

collect

public <R1> R1 collect(Supplier<R1> supplier,
                       BiConsumer<R1,? super R> accumulator,
                       BiConsumer<R1,R1> combiner)

Specified by:

collect in interface Stream<R>

iterator

public Iterator<R> iterator()

Description copied from interface: CacheStream

Usage of this operator requires closing this stream after you are done with the iterator. The preferred usage is to use a try with resource block on the stream.

This method has special usage with the BaseCacheStream.SegmentCompletionListener in that as entries are retrieved from the next method it will complete segments.

This method obeys the CacheStream.distributedBatchSize(int). Note that when using methods such as CacheStream.flatMap(Function) that you will have possibly more than 1 element mapped to a given key so this doesn't guarantee that many number of entries are returned per batch.

Note that the Iterator.remove() method is only supported if no intermediate operations have been applied to the stream and this is not a stream created from a Cache.values() collection.

Specified by:

iterator in interface BaseStream<R,Stream<R>>

Specified by:

iterator in interface CacheStream<R>

Returns:

the element iterator for this stream

spliterator

public Spliterator<R> spliterator()

Description copied from interface: CacheStream

Usage of this operator requires closing this stream after you are done with the spliterator. The preferred usage is to use a try with resource block on the stream.

Specified by:

spliterator in interface BaseStream<R,Stream<R>>

Specified by:

spliterator in interface CacheStream<R>

Returns:

the element spliterator for this stream

isParallel

public boolean isParallel()

Specified by:

isParallel in interface BaseStream<R,Stream<R>>

sequential

public CacheStream<R> sequential()

Description copied from interface: CacheStream

Specified by:

sequential in interface BaseStream<R,Stream<R>>

Specified by:

sequential in interface CacheStream<R>

Returns:

a sequential cache stream

parallel

public CacheStream<R> parallel()

Description copied from interface: CacheStream

Specified by:

parallel in interface BaseStream<R,Stream<R>>

Specified by:

parallel in interface CacheStream<R>

Returns:

a parallel cache stream

unordered

public CacheStream<R> unordered()

Description copied from interface: CacheStream

Specified by:

unordered in interface BaseStream<R,Stream<R>>

Specified by:

unordered in interface CacheStream<R>

Returns:

an unordered cache stream

onClose

public CacheStream<R> onClose(Runnable closeHandler)

Description copied from interface: CacheStream

Specified by:

onClose in interface BaseStream<R,Stream<R>>

Specified by:

onClose in interface CacheStream<R>

Returns:

a cache stream with the handler applied

close

public void close()

Specified by:

close in interface AutoCloseable

Specified by:

close in interface BaseStream<R,Stream<R>>

sorted

public CacheStream<R> sorted()

Description copied from interface: CacheStream

This operation is performed entirely on the local node irrespective of the backing cache. This operation will act as an intermediate iterator operation requiring data be brought locally for proper behavior. Beware this means it will require having all entries of this cache into memory at one time. This is described in more detail at CacheStream

Any subsequent intermediate operations and the terminal operation are also performed locally.

Specified by:

sorted in interface CacheStream<R>

Specified by:

sorted in interface Stream<R>

Returns:

the new stream

sorted

public CacheStream<R> sorted(Comparator<? super R> comparator)

Description copied from interface: CacheStream

This operation is performed entirely on the local node irrespective of the backing cache. This operation will act as an intermediate iterator operation requiring data be brought locally for proper behavior. Beware this means it will require having all entries of this cache into memory at one time. This is described in more detail at CacheStream

Any subsequent intermediate operations and the terminal operation are then performed locally.

Specified by:

sorted in interface CacheStream<R>

Specified by:

sorted in interface Stream<R>

Parameters:

comparator - the comparator to be used for sorting the elements

Returns:

the new stream

peek

public CacheStream<R> peek(Consumer<? super R> action)

Description copied from interface: CacheStream

Specified by:

peek in interface CacheStream<R>

Specified by:

peek in interface Stream<R>

Parameters:

action - the action to perform on the stream

Returns:

the new stream

limit

public CacheStream<R> limit(long maxSize)

Description copied from interface: CacheStream

This intermediate operation will be performed both remotely and locally to reduce how many elements are sent back from each node. More specifically this operation is applied remotely on each node to only return up to the maxSize value and then the aggregated results are limited once again on the local node.

This operation will act as an intermediate iterator operation requiring data be brought locally for proper behavior. This is described in more detail in the CacheStream documentation

Any subsequent intermediate operations and the terminal operation are then performed locally.

Specified by:

limit in interface CacheStream<R>

Specified by:

limit in interface Stream<R>

Parameters:

maxSize - how many elements to limit this stream to.

Returns:

the new stream

skip

public CacheStream<R> skip(long n)

Description copied from interface: CacheStream

This operation is performed entirely on the local node irrespective of the backing cache. This operation will act as an intermediate iterator operation requiring data be brought locally for proper behavior. This is described in more detail in the CacheStream documentation

Depending on the terminal operator this may or may not require all entries or a subset after skip is applied to be in memory all at once.

Any subsequent intermediate operations and the terminal operation are then performed locally.

Specified by:

skip in interface CacheStream<R>

Specified by:

skip in interface Stream<R>

Parameters:

n - how many elements to skip from this stream

Returns:

the new stream

filter

public CacheStream<R> filter(Predicate<? super R> predicate)

Description copied from interface: CacheStream

Specified by:

filter in interface CacheStream<R>

Specified by:

filter in interface Stream<R>

Returns:

the new cache stream

map

public <R1> CacheStream<R1> map(Function<? super R,? extends R1> mapper)

Description copied from interface: CacheStream

Specified by:

map in interface CacheStream<R>

Specified by:

map in interface Stream<R>

Returns:

the new cache stream

flatMap

public <R1> CacheStream<R1> flatMap(Function<? super R,? extends Stream<? extends R1>> mapper)

Description copied from interface: CacheStream

Specified by:

flatMap in interface CacheStream<R>

Specified by:

flatMap in interface Stream<R>

Returns:

the new cache stream

distinct

public CacheStream<R> distinct()

Description copied from interface: CacheStream

This operation will be invoked both remotely and locally when used with a distributed cache backing this stream. This operation will act as an intermediate iterator operation requiring data be brought locally for proper behavior. This is described in more detail in the CacheStream documentation

This intermediate iterator operation will be performed locally and remotely requiring possibly a subset of all elements to be in memory

Any subsequent intermediate operations and the terminal operation are then performed locally.

Specified by:

distinct in interface CacheStream<R>

Specified by:

distinct in interface Stream<R>

Returns:

the new stream

collect

public <R1,A> R1 collect(Collector<? super R,A,R1> collector)

Description copied from interface: CacheStream

Note when using a distributed backing cache for this stream the collector must be marshallable. This prevents the usage of Collectors class. However you can use the CacheCollectors static factory methods to create a serializable wrapper, which then creates the actual collector lazily after being deserialized. This is useful to use any method from the Collectors class as you would normally. Alternatively, you can call CacheStream.collect(SerializableSupplier) too.

Specified by:

collect in interface CacheStream<R>

Specified by:

collect in interface Stream<R>

Type Parameters:

R1 - collected type

A - intermediate collected type if applicable

Returns:

the collected value

See Also:

CacheCollectors

min

public Optional<R> min(Comparator<? super R> comparator)

Specified by:

min in interface Stream<R>

max

public Optional<R> max(Comparator<? super R> comparator)

Specified by:

max in interface Stream<R>

count

public long count()

Specified by:

count in interface Stream<R>

anyMatch

public boolean anyMatch(Predicate<? super R> predicate)

Specified by:

anyMatch in interface Stream<R>

allMatch

public boolean allMatch(Predicate<? super R> predicate)

Specified by:

allMatch in interface Stream<R>

noneMatch

public boolean noneMatch(Predicate<? super R> predicate)

Specified by:

noneMatch in interface Stream<R>

findFirst

public Optional<R> findFirst()

Specified by:

findFirst in interface Stream<R>

findAny

public Optional<R> findAny()

Specified by:

findAny in interface Stream<R>