Original - the original type of the underlying stream - normally CacheEntry or ObjectR - The type of the streampublic class DistributedCacheStream<Original,R> extends AbstractCacheStream<Original,R,Stream<R>,CacheStream<R>> implements CacheStream<R>
CacheStream that provides support for lazily distributing stream methods to appropriate
nodesStream.Builder<T>BaseCacheStream.SegmentCompletionListenercloseRunnable, cpm, distributedBatchSize, dm, executor, includeLoader, intermediateOperations, invocationContext, iteratorOperation, keyPartitioner, keysToFilter, localAddress, parallel, parallelDistribution, partition, registry, rehashAware, segmentCompletionListener, segmentsToFilter, stateTransferLock, supplier, timeout, timeoutUnit, toKeyFunction| Modifier | Constructor and Description |
|---|---|
protected |
DistributedCacheStream(AbstractCacheStream other)
This constructor is to be used only when a user calls a map or flat map method changing back to a regular
Stream from an IntStream, DoubleStream etc.
|
|
DistributedCacheStream(Address localAddress,
boolean parallel,
DistributionManager dm,
InvocationContext ctx,
Supplier<CacheStream<R>> supplier,
boolean includeLoader,
int distributedBatchSize,
Executor executor,
ComponentRegistry registry,
Function<? super Original,?> toKeyFunction)
Standard constructor requiring all pertinent information to properly utilize a distributed cache stream
|
| Modifier and Type | Method and Description |
|---|---|
boolean |
allMatch(Predicate<? super R> predicate) |
boolean |
anyMatch(Predicate<? super R> predicate) |
<R1,A> R1 |
collect(Collector<? super R,A,R1> collector) |
<R1> R1 |
collect(Supplier<R1> supplier,
BiConsumer<R1,? super R> accumulator,
BiConsumer<R1,R1> combiner)
Note: this method doesn't pay attention to ordering constraints and any sorting performed on the stream will
be ignored by this terminal operator.
|
long |
count() |
CacheStream<R> |
disableRehashAware()
Disables tracking of rehash events that could occur to the underlying cache.
|
CacheStream<R> |
distinct() |
CacheStream<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.
|
protected DistributedDoubleCacheStream |
doubleCacheStream() |
CacheStream<R> |
filter(Predicate<? super R> predicate) |
CacheStream<R> |
filterKeys(Set<?> keys)
Filters which entries are returned by only returning ones that map to the given key.
|
CacheStream<R> |
filterKeySegments(IntSet segments)
Filters which entries are returned by what segment they are present in.
|
CacheStream<R> |
filterKeySegments(Set<Integer> 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) |
protected Log |
getLog() |
protected DistributedIntCacheStream |
intCacheStream() |
Iterator<R> |
iterator() |
CacheStream<R> |
limit(long maxSize) |
protected DistributedLongCacheStream |
longCacheStream() |
<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> |
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) |
CacheStream<R> |
segmentCompletionListener(BaseCacheStream.SegmentCompletionListener listener)
Allows registration of a segment completion listener that is notified when a segment has completed
processing.
|
CacheStream<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() |
protected static <R> Supplier<CacheStream<R>> |
supplierStreamCast(Supplier supplier) |
CacheStream<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) |
protected CacheStream<R> |
unwrap() |
addIntermediateOperation, addIntermediateOperation, addIntermediateOperationMap, close, composeWithExceptions, isParallel, isPrimaryOwner, onClose, parallel, sequential, unorderedclone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, waitallMatch, anyMatch, collect, collect, collect, filter, flatMap, flatMapToDouble, flatMapToInt, flatMapToLong, forEach, forEach, map, mapToDouble, mapToInt, mapToLong, max, min, noneMatch, onClose, parallel, peek, reduce, reduce, reduce, sequential, sorted, toArray, unorderedbuilder, concat, empty, generate, iterate, of, ofclose, isParallelpublic DistributedCacheStream(Address localAddress, boolean parallel, DistributionManager dm, InvocationContext ctx, Supplier<CacheStream<R>> supplier, boolean includeLoader, int distributedBatchSize, Executor executor, ComponentRegistry registry, Function<? super Original,?> toKeyFunction)
localAddress - the local address for this nodeparallel - whether or not this stream is paralleldm - the distribution manager to find out what keys map wherectx - the invocation context when this stream is createdsupplier - a supplier of local cache stream instances.includeLoader - whether or not a cache loader should be utilized for these operationsdistributedBatchSize - default size of distributed batchesexecutor - executor to be used for certain operations that require async processing (ie. iterator)registry - component registry to wire objects withtoKeyFunction - function that can be applied to an object in the stream to convert it to a key or null if it
is a key already. This variable is used to tell also if the underlying stream contains
entries or not by this value being non nullprotected DistributedCacheStream(AbstractCacheStream other)
other - other instance of AbstractCacheStream to copy details fromprotected static <R> Supplier<CacheStream<R>> supplierStreamCast(Supplier supplier)
protected Log getLog()
getLog in class AbstractCacheStream<Original,R,Stream<R>,CacheStream<R>>protected CacheStream<R> unwrap()
unwrap in class AbstractCacheStream<Original,R,Stream<R>,CacheStream<R>>public CacheStream<R> filter(Predicate<? super R> predicate)
CacheStreampublic <R1> CacheStream<R1> map(Function<? super R,? extends R1> mapper)
CacheStreampublic IntCacheStream mapToInt(ToIntFunction<? super R> mapper)
CacheStreampublic LongCacheStream mapToLong(ToLongFunction<? super R> mapper)
CacheStreampublic DoubleCacheStream mapToDouble(ToDoubleFunction<? super R> mapper)
CacheStreammapToDouble in interface Stream<R>mapToDouble in interface CacheStream<R>mapper - a non-interfering, stateless
function to apply to each elementpublic <R1> CacheStream<R1> flatMap(Function<? super R,? extends Stream<? extends R1>> mapper)
CacheStreampublic IntCacheStream flatMapToInt(Function<? super R,? extends IntStream> mapper)
CacheStreamflatMapToInt in interface Stream<R>flatMapToInt in interface CacheStream<R>public LongCacheStream flatMapToLong(Function<? super R,? extends LongStream> mapper)
CacheStreamflatMapToLong in interface Stream<R>flatMapToLong in interface CacheStream<R>public DoubleCacheStream flatMapToDouble(Function<? super R,? extends DoubleStream> mapper)
CacheStreamflatMapToDouble in interface Stream<R>flatMapToDouble in interface CacheStream<R>public CacheStream<R> distinct()
CacheStreamThis 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.
public CacheStream<R> sorted()
CacheStreamThis 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.
public CacheStream<R> sorted(Comparator<? super R> comparator)
CacheStreamThis 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.
public CacheStream<R> peek(Consumer<? super R> action)
CacheStreampublic CacheStream<R> limit(long maxSize)
CacheStreamThis 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.
public CacheStream<R> skip(long n)
CacheStreamThis 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.
public R reduce(R identity, BinaryOperator<R> accumulator)
public Optional<R> reduce(BinaryOperator<R> accumulator)
public <U> U reduce(U identity,
BiFunction<U,? super R,U> accumulator,
BinaryOperator<U> combiner)
public <R1> R1 collect(Supplier<R1> supplier, BiConsumer<R1,? super R> accumulator, BiConsumer<R1,R1> combiner)
collect(Collector) method making sure the
Collector.Characteristics.UNORDERED property is not set.public <R1,A> R1 collect(Collector<? super R,A,R1> collector)
CacheStreamNote 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.
collect in interface Stream<R>collect in interface CacheStream<R>R1 - collected typeA - intermediate collected type if applicableCacheCollectorspublic Optional<R> min(Comparator<? super R> comparator)
public Optional<R> max(Comparator<? super R> comparator)
public Iterator<R> iterator()
CacheStreamUsage 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 org.infinispan.CacheStream.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.
iterator in interface BaseStream<R,Stream<R>>iterator in interface CacheStream<R>public Spliterator<R> spliterator()
CacheStreamUsage 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.
spliterator in interface BaseStream<R,Stream<R>>spliterator in interface CacheStream<R>public void forEach(Consumer<? super R> action)
CacheStreamThis 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..
public <K,V> void forEach(BiConsumer<Cache<K,V>,? super R> action)
CacheStreamCacheStream.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.
forEach in interface CacheStream<R>K - key type of the cacheV - value type of the cacheaction - consumer to be ran for each element in the streampublic void forEachOrdered(Consumer<? super R> action)
forEachOrdered in interface Stream<R>public <A> A[] toArray(IntFunction<A[]> generator)
public CacheStream<R> sequentialDistribution()
CacheStreamParallel distribution is enabled by default except for CacheStream.iterator() and
CacheStream.spliterator()
sequentialDistribution in interface BaseCacheStream<R,Stream<R>>sequentialDistribution in interface CacheStream<R>public CacheStream<R> parallelDistribution()
BaseCacheStreamParallel distribution is enabled by default except for CacheStream.iterator() and
CacheStream.spliterator()
parallelDistribution in interface BaseCacheStream<R,Stream<R>>parallelDistribution in interface CacheStream<R>public CacheStream<R> filterKeySegments(Set<Integer> segments)
CacheStreamCacheStream.filter(Predicate) method as this can control what nodes are
asked for data and what entries are read from the underlying CacheStore if present.filterKeySegments in interface BaseCacheStream<R,Stream<R>>filterKeySegments in interface CacheStream<R>segments - The segments to use for this stream operation. Any segments not in this set will be ignored.public CacheStream<R> filterKeySegments(IntSet segments)
CacheStreamCacheStream.filter(Predicate) method as this can control what nodes are
asked for data and what entries are read from the underlying CacheStore if present.filterKeySegments in interface BaseCacheStream<R,Stream<R>>filterKeySegments in interface CacheStream<R>segments - The segments to use for this stream operation. Any segments not in this set will be ignored.public CacheStream<R> filterKeys(Set<?> keys)
CacheStreamCacheStream.filter(Predicate) if the filter is holding references to the same
keys.filterKeys in interface BaseCacheStream<R,Stream<R>>filterKeys in interface CacheStream<R>keys - The keys that this stream will only operate on.public CacheStream<R> distributedBatchSize(int batchSize)
CacheStreamCacheStream.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.
distributedBatchSize in interface BaseCacheStream<R,Stream<R>>distributedBatchSize in interface CacheStream<R>batchSize - The size of each batch. This defaults to the state transfer chunk size.public CacheStream<R> segmentCompletionListener(BaseCacheStream.SegmentCompletionListener listener)
CacheStreamThis 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.
segmentCompletionListener in interface BaseCacheStream<R,Stream<R>>segmentCompletionListener in interface CacheStream<R>listener - The listener that will be called back as segments are completed.public CacheStream<R> disableRehashAware()
CacheStreamMost 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.
disableRehashAware in interface BaseCacheStream<R,Stream<R>>disableRehashAware in interface CacheStream<R>public CacheStream<R> timeout(long timeout, TimeUnit unit)
CacheStreamIf 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.
timeout in interface BaseCacheStream<R,Stream<R>>timeout in interface CacheStream<R>timeout - the maximum time to waitunit - the time unit of the timeout argumentprotected DistributedIntCacheStream intCacheStream()
protected DistributedDoubleCacheStream doubleCacheStream()
protected DistributedLongCacheStream longCacheStream()
Copyright © 2021 JBoss by Red Hat. All rights reserved.