public class LockedStreamImpl<K,V> extends Object implements LockedStream<K,V>
LockManager before and after executing the various
code.
This implementation doesn't work properly when using an optimistic transactional cache. Care should be made to prevent that usage if possible.
BaseCacheStream.SegmentCompletionListener| Constructor and Description |
|---|
LockedStreamImpl(CacheStream<CacheEntry<K,V>> realStream,
long time,
TimeUnit unit) |
| Modifier and Type | Method and Description |
|---|---|
void |
close() |
LockedStream<K,V> |
disableRehashAware()
Disables tracking of rehash events that could occur to the underlying cache.
|
LockedStream<K,V> |
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.
|
LockedStream<K,V> |
filter(Predicate<? super CacheEntry<K,V>> predicate)
Returns a locked stream consisting of the elements of this stream that match
the given predicate.
|
LockedStream<K,V> |
filterKeys(Set<?> keys)
Filters which entries are returned by only returning ones that map to the given key.
|
LockedStream<K,V> |
filterKeySegments(IntSet segments)
Filters which entries are returned by what segment they are present in.
|
LockedStream<K,V> |
filterKeySegments(Set<Integer> segments)
Filters which entries are returned by what segment they are present in.
|
void |
forEach(BiConsumer<Cache<K,V>,? super CacheEntry<K,V>> biConsumer)
Performs an action for each element of this stream on the primary owner of the given key.
|
<R> Map<K,R> |
invokeAll(BiFunction<Cache<K,V>,? super CacheEntry<K,V>,R> biFunction)
Performs a BiFunction for each element of this stream on the primary owner of each entry returning
a value.
|
boolean |
isParallel() |
Iterator<CacheEntry<K,V>> |
iterator()
This method is not supported when using a
LockedStream |
LockedStream<K,V> |
onClose(Runnable closeHandler) |
LockedStream<K,V> |
parallel() |
LockedStream<K,V> |
parallelDistribution()
This would enable sending requests to all other remote nodes when a terminal operator is performed.
|
LockedStream |
segmentCompletionListener(BaseCacheStream.SegmentCompletionListener listener)
This method is not supported when using a
LockedStream |
LockedStream<K,V> |
sequential() |
LockedStream<K,V> |
sequentialDistribution()
This would disable sending requests to all other remote nodes compared to one at a time.
|
Spliterator<CacheEntry<K,V>> |
spliterator()
This method is not supported when using a
LockedStream |
LockedStream |
timeout(long timeout,
TimeUnit unit)
Sets the timeout for the acquisition of the lock for each entry.
|
LockedStream<K,V> |
unordered() |
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, waitfilter, forEach, invokeAllpublic LockedStreamImpl(CacheStream<CacheEntry<K,V>> realStream, long time, TimeUnit unit)
public LockedStream<K,V> filter(Predicate<? super CacheEntry<K,V>> predicate)
LockedStreamThis filter is after the lock is acquired for the given key. This way the filter will see the same value as the consumer is given.
filter in interface LockedStream<K,V>predicate - predicatepublic void forEach(BiConsumer<Cache<K,V>,? super CacheEntry<K,V>> biConsumer)
LockedStream
This method is performed while holding exclusive lock over the given entry and will be released
only after the consumer has completed. In the function, entry.setValue(newValue) is equivalent to
cache.put(entry.getKey(), newValue).
If using pessimistic transactions this lock is not held using a transaction and thus the user can start a transaction in this consumer which also must be completed before returning. A transaction can be started in the consumer and if done it will share the same lock used to obtain the key.
Remember that if you are using an explicit transaction or an async method that these must be completed before the consumer returns to guarantee that they are operating within the scope of the lock for the given key. Failure to do so will lead into possible inconsistency as they will be performing operations without the proper locking.
Some methods on the provided cache may not work as expected. These include
Cache.putForExternalRead(Object, Object), AdvancedCache.lock(Object[]),
AdvancedCache.lock(Collection), and AdvancedCache.removeGroup(String).
If these methods are used inside of the Consumer on the cache it will throw a IllegalStateException.
This is due to possible interactions with locks while using these commands.
forEach in interface LockedStream<K,V>biConsumer - the biConsumer to run for each entry under their lockpublic <R> Map<K,R> invokeAll(BiFunction<Cache<K,V>,? super CacheEntry<K,V>,R> biFunction)
LockedStream
This method is currently marked as Experimental since this method returns a Map and requires blocking.
This operation could take a deal of time and as such should be done using an asynchronous API. Most likely
this return type will be changed to use some sort of asynchronous return value. This method is here until
this can be implemented.
This BiFunction is invoked while holding an exclusive lock over the given entry that will be released
only after the function has completed. In the function, entry.setValue(newValue) is equivalent to
cache.put(entry.getKey(), newValue).
If using pessimistic transactions this lock is not held using a transaction and thus the user can start a transaction in this consumer which also must be completed before returning. A transaction can be started in the biFunction and if done it will share the same lock used to obtain the key.
Remember if you are using an explicit transaction or an async method that these must be completed before the consumer returns to guarantee that they are operating within the scope of the lock for the given key. Failure to do so will lead into possible inconsistency as they will be performing operations without the proper locking.
Some methods on the provided cache may not work as expected. These include
Cache.putForExternalRead(Object, Object), AdvancedCache.lock(Object[]),
AdvancedCache.lock(Collection), and AdvancedCache.removeGroup(String).
If these methods are used inside of the Consumer on the cache it will throw a IllegalStateException.
This is due to possible interactions with locks while using these commands.
invokeAll in interface LockedStream<K,V>R - the return typebiFunction - the biFunction to run for each entry under their lockpublic LockedStream<K,V> sequentialDistribution()
LockedStreamParallel distribution is enabled by default except for CacheStream.iterator() and
CacheStream.spliterator()
sequentialDistribution in interface BaseCacheStream<CacheEntry<K,V>,LockedStream<K,V>>sequentialDistribution in interface LockedStream<K,V>public LockedStream<K,V> parallelDistribution()
LockedStreamParallel distribution is enabled by default except for CacheStream.iterator() and
CacheStream.spliterator()
parallelDistribution in interface BaseCacheStream<CacheEntry<K,V>,LockedStream<K,V>>parallelDistribution in interface LockedStream<K,V>public LockedStream<K,V> filterKeySegments(Set<Integer> segments)
LockedStreamCacheStream.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<CacheEntry<K,V>,LockedStream<K,V>>filterKeySegments in interface LockedStream<K,V>segments - The segments to use for this stream operation. Any segments not in this set will be ignored.public LockedStream<K,V> filterKeySegments(IntSet segments)
LockedStreamCacheStream.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<CacheEntry<K,V>,LockedStream<K,V>>filterKeySegments in interface LockedStream<K,V>segments - The segments to use for this stream operation. Any segments not in this set will be ignored.public LockedStream<K,V> filterKeys(Set<?> keys)
LockedStreamCacheStream.filter(Predicate) if the filter is holding references to the same
keys.filterKeys in interface BaseCacheStream<CacheEntry<K,V>,LockedStream<K,V>>filterKeys in interface LockedStream<K,V>keys - The keys that this stream will only operate on.public LockedStream<K,V> distributedBatchSize(int batchSize)
LockedStreamCacheStream.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<CacheEntry<K,V>,LockedStream<K,V>>distributedBatchSize in interface LockedStream<K,V>batchSize - The size of each batch. This defaults to the state transfer chunk size.public LockedStream segmentCompletionListener(BaseCacheStream.SegmentCompletionListener listener)
LockedStreamLockedStreamsegmentCompletionListener in interface BaseCacheStream<CacheEntry<K,V>,LockedStream<K,V>>segmentCompletionListener in interface LockedStream<K,V>listener - The listener that will be called back as segments are completed.public LockedStream<K,V> disableRehashAware()
LockedStreamMost 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<CacheEntry<K,V>,LockedStream<K,V>>disableRehashAware in interface LockedStream<K,V>public LockedStream timeout(long timeout, TimeUnit unit)
LockedStreamtimeout in interface BaseCacheStream<CacheEntry<K,V>,LockedStream<K,V>>timeout in interface LockedStream<K,V>timeout - the maximum time to waitunit - the time unit of the timeout argumentpublic Iterator<CacheEntry<K,V>> iterator()
LockedStreamLockedStreamiterator in interface BaseStream<CacheEntry<K,V>,LockedStream<K,V>>iterator in interface LockedStream<K,V>public Spliterator<CacheEntry<K,V>> spliterator()
LockedStreamLockedStreamspliterator in interface BaseStream<CacheEntry<K,V>,LockedStream<K,V>>spliterator in interface LockedStream<K,V>public boolean isParallel()
isParallel in interface BaseStream<CacheEntry<K,V>,LockedStream<K,V>>public LockedStream<K,V> sequential()
sequential in interface BaseStream<CacheEntry<K,V>,LockedStream<K,V>>public LockedStream<K,V> parallel()
parallel in interface BaseStream<CacheEntry<K,V>,LockedStream<K,V>>public LockedStream<K,V> unordered()
unordered in interface BaseStream<CacheEntry<K,V>,LockedStream<K,V>>public LockedStream<K,V> onClose(Runnable closeHandler)
onClose in interface BaseStream<CacheEntry<K,V>,LockedStream<K,V>>public void close()
close in interface AutoCloseableclose in interface BaseStream<CacheEntry<K,V>,LockedStream<K,V>>Copyright © 2021 JBoss by Red Hat. All rights reserved.