Package org.infinispan

Interface LockedStream<K,​V>

  • All Superinterfaces:
    AutoCloseable, BaseCacheStream<org.infinispan.container.entries.CacheEntry<K,​V>,​LockedStream<K,​V>>, BaseStream<org.infinispan.container.entries.CacheEntry<K,​V>,​LockedStream<K,​V>>
    public interface LockedStream<K,​V>
extends BaseCacheStream<org.infinispan.container.entries.CacheEntry<K,​V>,​LockedStream<K,​V>>
    Stream that allows for operation upon data solely with side effects by using forEach(BiConsumer) where the BiConsumer is invoked while guaranteeing that the entry being passed is properly locked for the entire duration of the invocation.

    An attempt is made to acquire the lock for an entry using the default LockingConfiguration.lockAcquisitionTimeout() before invoking any operations on it.

    Since:
    9.1
    Author:
    wburns

    • Method Detail

      • filter

        LockedStream<K,​V> filter​(Predicate<? super org.infinispan.container.entries.CacheEntry<K,​V>> predicate)
        Returns a locked stream consisting of the elements of this stream that match the given predicate.

        This 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.

        Parameters:
        predicate - predicate
        Returns:
        a LockedStream with the filter applied

      • filter

        default LockedStream<K,​V> filter​(org.infinispan.util.function.SerializablePredicate<? super org.infinispan.container.entries.CacheEntry<K,​V>> predicate)
        Same as filter(Predicate) except that the Predicate must also implement Serializable

        The compiler will pick this overload for lambda parameters, making them Serializable

        Parameters:
        predicate - the predicate to filter out unwanted entries
        Returns:
        a LockedStream with the filter applied

      • forEach

        void forEach​(BiConsumer<Cache<K,​V>,​? super org.infinispan.container.entries.CacheEntry<K,​V>> biConsumer)
        Performs an action for each element of this stream on the primary owner of the given key.

        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.

        Parameters:
        biConsumer - the biConsumer to run for each entry under their lock

      • forEach

        default void forEach​(org.infinispan.util.function.SerializableBiConsumer<Cache<K,​V>,​? super org.infinispan.container.entries.CacheEntry<K,​V>> biConsumer)
        Same as forEach(BiConsumer) except that the BiConsumer must also implement Serializable

        The compiler will pick this overload for lambda parameters, making them Serializable

        Parameters:
        biConsumer - the biConsumer to run for each entry under their lock

      • invokeAll

        @Experimental
<R> Map<K,​R> invokeAll​(BiFunction<Cache<K,​V>,​? super org.infinispan.container.entries.CacheEntry<K,​V>,​R> biFunction)
        Performs a BiFunction for each element of this stream on the primary owner of each entry returning a value. The returned value from the function will be sent back to the user mapped to the key that generated it, with all of these stored in a map. Both the BiFunction and the returned value must be Serializable in some way. This method will return only after all entries have been processed.

        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.

        Type Parameters:
        R - the return type
        Parameters:
        biFunction - the biFunction to run for each entry under their lock
        Returns:
        a map with each key mapped to the value returned from the bi function

      • invokeAll

        @Experimental
default <R> Map<K,​R> invokeAll​(org.infinispan.util.function.SerializableBiFunction<Cache<K,​V>,​? super org.infinispan.container.entries.CacheEntry<K,​V>,​R> biFunction)
        Same as invokeAll(BiFunction) except that the BiFunction must also implement Serializable

        The compiler will pick this overload for lambda parameters, making them Serializable

        Type Parameters:
        R - the return type
        Parameters:
        biFunction - the biFunction to run for each entry under their lock
        Returns:
        a map with each key mapped to the value returned from the bi function

      • parallelDistribution

        LockedStream<K,​V> parallelDistribution()
        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<K,​V>
        Returns:
        a stream with parallel distribution enabled.

      • filterKeySegments

        LockedStream<K,​V> filterKeySegments​(Set<Integer> segments)
        Deprecated.
        This is to be replaced by filterKeySegments(IntSet)
        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<K,​V>
        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

        LockedStream<K,​V> filterKeySegments​(IntSet segments)
        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<K,​V>
        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

        LockedStream<K,​V> filterKeys​(Set<?> keys)
        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<K,​V>
        Parameters:
        keys - The keys that this stream will only operate on.
        Returns:
        a stream with the keys filtered.

      • disableRehashAware

        LockedStream<K,​V> disableRehashAware()
        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<K,​V>
        Returns:
        a stream with rehash awareness disabled.

      • timeout

        LockedStream<K,​V> timeout​(long time,
                                TimeUnit unit)
        Sets the timeout for the acquisition of the lock for each entry.
        Specified by:
        timeout in interface BaseCacheStream<K,​V>
        Parameters:
        time - the maximum time to wait
        unit - the time unit of the timeout argument
        Returns:
        a LockedStream with the timeout applied