Class PartitionAwareClusterPublisherManager<K,V>

java.lang.Object

org.infinispan.reactive.publisher.impl.ClusterPublisherManagerImpl<K,V>

org.infinispan.reactive.publisher.impl.PartitionAwareClusterPublisherManager<K,V>

All Implemented Interfaces:

ClusterPublisherManager<K,V>

public class PartitionAwareClusterPublisherManager<K,V>
extends ClusterPublisherManagerImpl<K,V>

Cluster stream manager that also pays attention to partition status and properly closes iterators and throws exceptions when the availability mode changes.

Nested Class Summary

Nested classes/interfaces inherited from class ClusterPublisherManagerImpl

ClusterPublisherManagerImpl.SegmentAwarePublisherImpl<I,R>, ClusterPublisherManagerImpl.TopologySegmentAwarePublisherImpl

Field Summary

Fields

Modifier and Type	Field	Description
protected ComponentRef<Cache<?,?>>	cache
protected final PartitionAwareClusterPublisherManager<K,V>.org.infinispan.reactive.publisher.impl.PartitionAwareClusterPublisherManager.PartitionListener	listener

Fields inherited from class ClusterPublisherManagerImpl

log, rpcOptions

Constructor Summary

Constructors

Constructor	Description
PartitionAwareClusterPublisherManager()

Method Summary

Modifier and Type	Method	Description
<R> SegmentPublisherSupplier<R>	entryPublisher(IntSet segments, Set<K> keysToInclude, InvocationContext invocationContext, long explicitFlags, DeliveryGuarantee deliveryGuarantee, int batchSize, Function<? super org.reactivestreams.Publisher<CacheEntry<K,V>>, ? extends org.reactivestreams.Publisher<R>> transformer)	Performs the given transformer on data in the cache, resulting in multiple values.
org.reactivestreams.Publisher<SegmentPublisherSupplier.Notification<CacheEntry<K,V>>>	entryPublisherForTopology(int topologyId, int batchSize, Map<Address,IntSet> targets)	This method is similar to ClusterPublisherManager.entryPublisher(IntSet, Set, InvocationContext, long, DeliveryGuarantee, int, Function) except it is designed to be used solely for iterating over entries for given targets with DeliveryGuarantee.AT_MOST_ONCE semantics since it cannot iterate using any other topology.
<R> CompletionStage<R>	entryReduction(boolean parallelPublisher, IntSet segments, Set<K> keysToInclude, InvocationContext ctx, long explicitFlags, DeliveryGuarantee deliveryGuarantee, Function<? super org.reactivestreams.Publisher<CacheEntry<K,V>>, ? extends CompletionStage<R>> transformer, Function<? super org.reactivestreams.Publisher<R>, ? extends CompletionStage<R>> finalizer)	Performs the given transformer and finalizer on data in the cache, resulting in a single value.
<R> SegmentPublisherSupplier<R>	keyPublisher(IntSet segments, Set<K> keysToInclude, InvocationContext invocationContext, long explicitFlags, DeliveryGuarantee deliveryGuarantee, int batchSize, Function<? super org.reactivestreams.Publisher<K>, ? extends org.reactivestreams.Publisher<R>> transformer)	Same as ClusterPublisherManager.entryPublisher(IntSet, Set, InvocationContext, long, DeliveryGuarantee, int, Function) except that the source publisher provided to the transformer is made up of keys only.
<R> CompletionStage<R>	keyReduction(boolean parallelPublisher, IntSet segments, Set<K> keysToInclude, InvocationContext ctx, long explicitFlags, DeliveryGuarantee deliveryGuarantee, Function<? super org.reactivestreams.Publisher<K>, ? extends CompletionStage<R>> transformer, Function<? super org.reactivestreams.Publisher<R>, ? extends CompletionStage<R>> finalizer)	Same as ClusterPublisherManager.entryReduction(boolean, IntSet, Set, InvocationContext, long, DeliveryGuarantee, Function, Function) except that the source publisher provided to the transformer is made up of keys only.
CompletionStage<Long>	sizePublisher(IntSet segments, InvocationContext ctx, long flags)
void	start()

Methods inherited from class ClusterPublisherManagerImpl

entryComposedType, entryPublisherForTopology, stop

Methods inherited from class Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

listener

protected final PartitionAwareClusterPublisherManager<K,V>.org.infinispan.reactive.publisher.impl.PartitionAwareClusterPublisherManager.PartitionListener listener

cache

protected ComponentRef<Cache<?,?>> cache

Constructor Details

PartitionAwareClusterPublisherManager

public PartitionAwareClusterPublisherManager()

Method Details

start

public void start()

Overrides:

start in class ClusterPublisherManagerImpl<K,V>

keyReduction

public <R> CompletionStage<R> keyReduction(boolean parallelPublisher,
 IntSet segments,
 Set<K> keysToInclude,
 InvocationContext ctx,
 long explicitFlags,
 DeliveryGuarantee deliveryGuarantee,
 Function<? super org.reactivestreams.Publisher<K>, ? extends CompletionStage<R>> transformer,
 Function<? super org.reactivestreams.Publisher<R>, ? extends CompletionStage<R>> finalizer)

Description copied from interface: ClusterPublisherManager

Same as ClusterPublisherManager.entryReduction(boolean, IntSet, Set, InvocationContext, long, DeliveryGuarantee, Function, Function) except that the source publisher provided to the transformer is made up of keys only.

Specified by:

keyReduction in interface ClusterPublisherManager<K,V>

Overrides:

keyReduction in class ClusterPublisherManagerImpl<K,V>

Type Parameters:

R - return value type

Returns:

CompletionStage that contains the resulting value when complete

entryReduction

public <R> CompletionStage<R> entryReduction(boolean parallelPublisher,
 IntSet segments,
 Set<K> keysToInclude,
 InvocationContext ctx,
 long explicitFlags,
 DeliveryGuarantee deliveryGuarantee,
 Function<? super org.reactivestreams.Publisher<CacheEntry<K,V>>, ? extends CompletionStage<R>> transformer,
 Function<? super org.reactivestreams.Publisher<R>, ? extends CompletionStage<R>> finalizer)

Description copied from interface: ClusterPublisherManager

Performs the given transformer and finalizer on data in the cache, resulting in a single value. Depending on the deliveryGuarantee the transformer may be invoked 1..numSegments times. It could be that the transformer is invoked for every segment and produces a result. All of these results are then fed into the finalizer to produce a final result. If publisher is parallel the finalizer will be invoked on each node to ensure there is only a single result per node.

If the provided transformer internally uses a reduction with a default value, that value must be its identity value. This is the same as can be seen at Stream.reduce(Object, BinaryOperator). Then as long as the finalizer can handle the identity value it will be properly reduced.

Specified by:

entryReduction in interface ClusterPublisherManager<K,V>

Overrides:

entryReduction in class ClusterPublisherManagerImpl<K,V>

Type Parameters:

R - return value type

Parameters:

parallelPublisher - Whether on each node the publisher should be parallelized remotely and locally

segments - determines what entries should be evaluated by only using ones that map to the given segments (if null assumes all segments)

keysToInclude - set of keys that should only be used (if null all entries for the given segments will be evaluated)

ctx - context of the invoking operation, context entries override the values in the cache (may be null)

explicitFlags - cache flags, which are passed to KeySetCommand or EntrySetCommand

deliveryGuarantee - delivery guarantee for given entries

transformer - reduces the given publisher of data eventually into a single value. Must not be null.

finalizer - reduces all of the single values produced by the transformer or this finalizer into one final value. Must not be null.

Returns:

CompletionStage that contains the resulting value when complete

keyPublisher

public <R> SegmentPublisherSupplier<R> keyPublisher(IntSet segments,
 Set<K> keysToInclude,
 InvocationContext invocationContext,
 long explicitFlags,
 DeliveryGuarantee deliveryGuarantee,
 int batchSize,
 Function<? super org.reactivestreams.Publisher<K>, ? extends org.reactivestreams.Publisher<R>> transformer)

Description copied from interface: ClusterPublisherManager

Same as ClusterPublisherManager.entryPublisher(IntSet, Set, InvocationContext, long, DeliveryGuarantee, int, Function) except that the source publisher provided to the transformer is made up of keys only.

Specified by:

keyPublisher in interface ClusterPublisherManager<K,V>

Overrides:

keyPublisher in class ClusterPublisherManagerImpl<K,V>

Type Parameters:

R - return value type

Returns:

Publisher that when subscribed to will return the results and notify of segment completion if necessary

entryPublisher

public <R> SegmentPublisherSupplier<R> entryPublisher(IntSet segments,
 Set<K> keysToInclude,
 InvocationContext invocationContext,
 long explicitFlags,
 DeliveryGuarantee deliveryGuarantee,
 int batchSize,
 Function<? super org.reactivestreams.Publisher<CacheEntry<K,V>>, ? extends org.reactivestreams.Publisher<R>> transformer)

Description copied from interface: ClusterPublisherManager

Performs the given transformer on data in the cache, resulting in multiple values. If a single value is desired, the user should use ClusterPublisherManager.entryReduction(boolean, IntSet, Set, InvocationContext, long, DeliveryGuarantee, Function, Function) instead as it can optimize some things. Depending on the deliveryGuarantee the transformer may be invoked 1..numSegments times per node. Results from a given node will retrieve values up to batchSize values until some are consumed.

For example when using RxJava and using an intermediate operation such as Flowable.switchIfEmpty(Publisher) this can add elements if the given Publisher is empty, and it is very possible that a segment may not have entries and therefore may add the elements the switched Publisher returns multiple times.

Methods that add elements to the returned Publisher are fine as long as they are tied to a specific entry, for example Flowable.flatMap(io.reactivex.rxjava3.functions.Function) which can reproduce the same elements when provided the same input entry from the cache.

Specified by:

entryPublisher in interface ClusterPublisherManager<K,V>

Overrides:

entryPublisher in class ClusterPublisherManagerImpl<K,V>

Type Parameters:

R - return value type

Parameters:

segments - determines what entries should be evaluated by only using ones that map to the given segments (if null assumes all segments)

keysToInclude - set of keys that should only be used (if null all entries for the given segments will be evaluated)

invocationContext - context of the invoking operation, context entries override the values in the cache (may be null)

explicitFlags - cache flags

deliveryGuarantee - delivery guarantee for given entries

batchSize - how many entries to be returned at a given time

transformer - transform the given stream of data into something else (requires non null)

Returns:

Publisher that when subscribed to will return the results and notify of segment completion if necessary

sizePublisher

public CompletionStage<Long> sizePublisher(IntSet segments,
 InvocationContext ctx,
 long flags)

Specified by:

sizePublisher in interface ClusterPublisherManager<K,V>

Overrides:

sizePublisher in class ClusterPublisherManagerImpl<K,V>

entryPublisherForTopology

public org.reactivestreams.Publisher<SegmentPublisherSupplier.Notification<CacheEntry<K,V>>> entryPublisherForTopology(int topologyId,
 int batchSize,
 Map<Address,IntSet> targets)

Description copied from interface: ClusterPublisherManager

This method is similar to ClusterPublisherManager.entryPublisher(IntSet, Set, InvocationContext, long, DeliveryGuarantee, int, Function) except it is designed to be used solely for iterating over entries for given targets with DeliveryGuarantee.AT_MOST_ONCE semantics since it cannot iterate using any other topology.

Specified by:

entryPublisherForTopology in interface ClusterPublisherManager<K,V>

Overrides:

entryPublisherForTopology in class ClusterPublisherManagerImpl<K,V>

Parameters:

batchSize - how many entries to be returned at a given time

targets - the address and segment targets to use explicitly

Returns:

a publisher that will publish entries