org.infinispan.commons.reactive

Class AbstractAsyncPublisherHandler<Target,Output,InitialResponse,NextResponse>

java.lang.Object

org.infinispan.commons.reactive.AbstractAsyncPublisherHandler<Target,Output,InitialResponse,NextResponse>

All Implemented Interfaces:

io.reactivex.rxjava3.functions.Action, io.reactivex.rxjava3.functions.LongConsumer

public abstract class AbstractAsyncPublisherHandler<Target,Output,InitialResponse,NextResponse>
extends Object
implements io.reactivex.rxjava3.functions.LongConsumer, io.reactivex.rxjava3.functions.Action

Abstract handler that handles request and cancellation of a given non blocking resource. When additional entries are requested via Subscription.request(long) the handler will keep track of the outstanding count and invoke either sendInitialCommand(Object, int) or sendNextCommand(Object, int) depending upon if it is the first request or subsequent. This handler guarantees that there is only one outstanding request and that if the request amount is larger than the batch, it will continue to send new commands one at a time until the request amount has been satisfied.

The handler processes each target via the provided Supplier one by one, exhausting all values from the target, until there are no more targets left or the publishing has been cancelled by the subscriber.

When a command returns successfully either the handleInitialResponse(Object, Object) or handleNextResponse(Object, Object) will be invoked with the response object. The returned entries can then be emitted by invoking the onNext(Object) method for each value. Note that the return of onNext should be checked in case if the Subscriber has cancelled the publishing of values in the middle.

A command may also encounter a Throwable and it is possible to customize what happens by implementing the handleThrowableInResponse(Throwable, Object) method. For example you may want to skip the given

Each request is provided a batchSize argument and the underlying resource should adhere to this, failure to do so may cause an OutOfMemoryError, since entries are only emitted to the Subscriber based on the requested amount, and any additional are enqueued.

This handler also supports Subscription.cancel() by extending the sendCancel(Object) command and the underlying service must be cancelled in an asynchronous and non blocking fashion. Once cancelled this Publisher will not publish additional values or requests.

Note this handler only supports a single Subscriber for the returned Publisher from startPublisher(). Failure to do so can cause multiple requests and unexpected problems.

Field Summary

Fields

Modifier and Type	Field and Description
protected int	batchSize
protected static Log	log
protected Supplier<Target>	supplier
protected static boolean	trace

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	AbstractAsyncPublisherHandler(int batchSize, Supplier<Target> supplier, Target firstTarget)

Method Summary

Modifier and Type	Method and Description
void	accept(long count) This method is invoked every time a new request is sent to the underlying publisher.
protected boolean	checkCancelled() This method returns whether this subscription has been cancelled
protected abstract long	handleInitialResponse(InitialResponse response, Target target)
protected abstract long	handleNextResponse(NextResponse response, Target target)
protected void	handleThrowableInResponse(Throwable t, Target target) Allows any implementor to handle what happens when a Throwable is encountered.
protected boolean	onNext(Output value) Method that should be called for each emitted output value.
void	run() This is invoked when the Subscription is cancelled
protected abstract void	sendCancel(Target target)
protected abstract CompletionStage<InitialResponse>	sendInitialCommand(Target target, int batchSize)
protected abstract CompletionStage<NextResponse>	sendNextCommand(Target target, int batchSize)
org.reactivestreams.Publisher<Output>	startPublisher()
protected void	targetComplete() Method to invoke when a given target is found to have been completed and the next target should be used

Methods inherited from class java.lang.Object

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

Field Detail

log

protected static final Log log

trace

protected static final boolean trace

batchSize

protected final int batchSize

supplier

protected final Supplier<Target> supplier

Constructor Detail

AbstractAsyncPublisherHandler

protected AbstractAsyncPublisherHandler(int batchSize,
                                        Supplier<Target> supplier,
                                        Target firstTarget)

Method Detail

startPublisher

public org.reactivestreams.Publisher<Output> startPublisher()

run

public void run()

This is invoked when the Subscription is cancelled

Specified by:

run in interface io.reactivex.rxjava3.functions.Action

sendCancel

protected abstract void sendCancel(Target target)

sendInitialCommand

protected abstract CompletionStage<InitialResponse> sendInitialCommand(Target target,
                                                                       int batchSize)

sendNextCommand

protected abstract CompletionStage<NextResponse> sendNextCommand(Target target,
                                                                 int batchSize)

handleInitialResponse

protected abstract long handleInitialResponse(InitialResponse response,
                                              Target target)

handleNextResponse

protected abstract long handleNextResponse(NextResponse response,
                                           Target target)

handleThrowableInResponse

protected void handleThrowableInResponse(Throwable t,
                                         Target target)

Allows any implementor to handle what happens when a Throwable is encountered. By default the returned publisher invokes Subscriber.onError(Throwable) and stops processing. It is possible to ignore the throwable and continue processing by invoking accept(long) with a value of 0. It may also be required to reset the currentTarget so it is initialized to the next supplied value.

Parameters:

t - throwable that was encountered

target - the target which was invoked that caused the throwable

accept

public void accept(long count)

This method is invoked every time a new request is sent to the underlying publisher. We need to submit a request if there is not a pending one. Whenever requestedAmount is a number greater than 0, that means we must submit or there is a pending one.

Specified by:

accept in interface io.reactivex.rxjava3.functions.LongConsumer

Parameters:

count - request count

onNext

protected boolean onNext(Output value)

Method that should be called for each emitted output value. The returned boolean is whether the handler should continue publishing values or not.

Parameters:

value - value emit to the publisher

Returns:

whether to continue emitting values

targetComplete

protected void targetComplete()

Method to invoke when a given target is found to have been completed and the next target should be used

checkCancelled

protected boolean checkCancelled()

This method returns whether this subscription has been cancelled