PollerFlux<T,U> Class

Reference

Package:: com.azure.core.util.polling

Maven Artifact:: com.azure:azure-core:1.48.0

java.lang.Object
- reactor.core.publisher.Flux
- - com.azure.core.util.polling.PollerFlux<T,U>

Type Parameters

T: The type of poll response value.

U: The type of the final result of long-running operation.

public final class PollerFlux<T,U>
extends Flux<AsyncPollResponse<T,U>>

A Flux that simplifies the task of executing long-running operations against an Azure service. A subscription to PollerFlux<T,U> initiates a long-running operation and polls the status until it completes.

Code samples

Instantiating and subscribing to PollerFlux

LocalDateTime timeToReturnFinalResponse = LocalDateTime.now().plus(Duration.ofMillis(800));

 // Create poller instance
 PollerFlux<String, String> poller = new PollerFlux<>(Duration.ofMillis(100),
     (context) -> Mono.empty(),
     // Define your custom poll operation
     (context) ->  {
         if (LocalDateTime.now().isBefore(timeToReturnFinalResponse)) {
             System.out.println("Returning intermediate response.");
             return Mono.just(new PollResponse<>(LongRunningOperationStatus.IN_PROGRESS,
                     "Operation in progress."));
         } else {
             System.out.println("Returning final response.");
             return Mono.just(new PollResponse<>(LongRunningOperationStatus.SUCCESSFULLY_COMPLETED,
                     "Operation completed."));
         }
     },
     (activationResponse, context) -> Mono.error(new RuntimeException("Cancellation is not supported")),
     (context) -> Mono.just("Final Output"));

 // Listen to poll responses
 poller.subscribe(response -> {
     // Process poll response
     System.out.printf("Got response. Status: %s, Value: %s%n", response.getStatus(), response.getValue());
 });
 // Do something else

Asynchronously wait for polling to complete and then retrieve the final result

LocalDateTime timeToReturnFinalResponse = LocalDateTime.now().plus(Duration.ofMinutes(5));

 // Create poller instance
 PollerFlux<String, String> poller = new PollerFlux<>(Duration.ofMillis(100),
     (context) -> Mono.empty(),
     (context) ->  {
         if (LocalDateTime.now().isBefore(timeToReturnFinalResponse)) {
             System.out.println("Returning intermediate response.");
             return Mono.just(new PollResponse<>(LongRunningOperationStatus.IN_PROGRESS,
                     "Operation in progress."));
         } else {
             System.out.println("Returning final response.");
             return Mono.just(new PollResponse<>(LongRunningOperationStatus.SUCCESSFULLY_COMPLETED,
                     "Operation completed."));
         }
     },
     (activationResponse, context) -> Mono.just("FromServer:OperationIsCancelled"),
     (context) -> Mono.just("FromServer:FinalOutput"));

 poller.take(Duration.ofMinutes(30))
         .last()
         .flatMap(asyncPollResponse -> {
             if (asyncPollResponse.getStatus() == LongRunningOperationStatus.SUCCESSFULLY_COMPLETED) {
                 // operation completed successfully, retrieving final result.
                 return asyncPollResponse
                         .getFinalResult();
             } else {
                 return Mono.error(new RuntimeException("polling completed unsuccessfully with status:"
                         + asyncPollResponse.getStatus()));
             }
         }).block();

Block for polling to complete and then retrieve the final result

AsyncPollResponse<String, String> terminalResponse = pollerFlux.blockLast();
 System.out.printf("Polling complete. Final Status: %s", terminalResponse.getStatus());
 if (terminalResponse.getStatus() == LongRunningOperationStatus.SUCCESSFULLY_COMPLETED) {
     String finalResult = terminalResponse.getFinalResult().block();
     System.out.printf("Polling complete. Final Status: %s", finalResult);
 }

Asynchronously poll until poller receives matching status

final Predicate<AsyncPollResponse<String, String>> isComplete = response -> {
     return response.getStatus() != LongRunningOperationStatus.IN_PROGRESS
         && response.getStatus() != LongRunningOperationStatus.NOT_STARTED;
 };

 pollerFlux
     .takeUntil(isComplete)
     .subscribe(completed -> {
         System.out.println("Completed poll response, status: " + completed.getStatus());
     });

Asynchronously cancel the long running operation

LocalDateTime timeToReturnFinalResponse = LocalDateTime.now().plus(Duration.ofMinutes(5));

 // Create poller instance
 PollerFlux<String, String> poller = new PollerFlux<>(Duration.ofMillis(100),
     (context) -> Mono.empty(),
     (context) ->  {
         if (LocalDateTime.now().isBefore(timeToReturnFinalResponse)) {
             System.out.println("Returning intermediate response.");
             return Mono.just(new PollResponse<>(LongRunningOperationStatus.IN_PROGRESS,
                     "Operation in progress."));
         } else {
             System.out.println("Returning final response.");
             return Mono.just(new PollResponse<>(LongRunningOperationStatus.SUCCESSFULLY_COMPLETED,
                     "Operation completed."));
         }
     },
     (activationResponse, context) -> Mono.just("FromServer:OperationIsCancelled"),
     (context) -> Mono.just("FromServer:FinalOutput"));

 // Asynchronously wait 30 minutes to complete the polling, if not completed
 // within in the time then cancel the server operation.
 poller.take(Duration.ofMinutes(30))
         .last()
         .flatMap(asyncPollResponse -> {
             if (!asyncPollResponse.getStatus().isComplete()) {
                 return asyncPollResponse
                         .cancelOperation()
                         .then(Mono.error(new RuntimeException("Operation is cancelled!")));
             } else {
                 return Mono.just(asyncPollResponse);
             }
         }).block();

Instantiating and subscribing to PollerFlux from a known polling strategy

// Create poller instance
 PollerFlux<BinaryData, String> poller = PollerFlux.create(
     Duration.ofMillis(100),
     // pass in your custom activation operation
     () -> Mono.just(new SimpleResponse<Void>(new HttpRequest(
         HttpMethod.POST,
         "http://httpbin.org"),
         202,
         new HttpHeaders().set("Operation-Location", "http://httpbin.org"),
         null)),
     new OperationResourcePollingStrategy<>(new HttpPipelineBuilder().build()),
     TypeReference.createInstance(BinaryData.class),
     TypeReference.createInstance(String.class));

 // Listen to poll responses
 poller.subscribe(response -> {
     // Process poll response
     System.out.printf("Got response. Status: %s, Value: %s%n", response.getStatus(), response.getValue());
 });
 // Do something else

Instantiating and subscribing to PollerFlux from a custom polling strategy

// Create custom polling strategy based on OperationResourcePollingStrategy
 PollingStrategy<BinaryData, String> strategy = new OperationResourcePollingStrategy<BinaryData, String>(
         new HttpPipelineBuilder().build()) {
     // override any interface method to customize the polling behavior
     @Override
     public Mono<PollResponse<BinaryData>> poll(PollingContext<BinaryData> context,
                                                TypeReference<BinaryData> pollResponseType) {
         return Mono.just(new PollResponse<>(
             LongRunningOperationStatus.SUCCESSFULLY_COMPLETED,
             BinaryData.fromString("")));
     }
 };

 // Create poller instance
 PollerFlux<BinaryData, String> poller = PollerFlux.create(
     Duration.ofMillis(100),
     // pass in your custom activation operation
     () -> Mono.just(new SimpleResponse<Void>(new HttpRequest(
         HttpMethod.POST,
         "http://httpbin.org"),
         202,
         new HttpHeaders().set("Operation-Location", "http://httpbin.org"),
         null)),
     strategy,
     TypeReference.createInstance(BinaryData.class),
     TypeReference.createInstance(String.class));

 // Listen to poll responses
 poller.subscribe(response -> {
     // Process poll response
     System.out.printf("Got response. Status: %s, Value: %s%n", response.getStatus(), response.getValue());
 });
 // Do something else

Constructor Summary

Constructor	Description
PollerFlux(Duration pollInterval, Function<PollingContext<T>,Mono<T>> activationOperation, Function<PollingContext<T>,Mono<PollResponse<T>>> pollOperation, BiFunction<PollingContext<T>,PollResponse<T>,Mono<T>> cancelOperation, Function<PollingContext<T>,Mono<U>> fetchResultOperation)	Creates PollerFlux.

Method Summary

Modifier and Type	Method and Description
static PollerFlux<T,U>	<T,U>create(Duration pollInterval, Function<PollingContext<T>,Mono<PollResponse<T>>> activationOperation, Function<PollingContext<T>,Mono<PollResponse<T>>> pollOperation, BiFunction<PollingContext<T>,PollResponse<T>,Mono<T>> cancelOperation, Function<PollingContext<T>,Mono<U>> fetchResultOperation) Creates PollerFlux.
static PollerFlux<T,U>	<T,U>create(Duration pollInterval, Supplier<Mono<? extends Response<?>>> initialOperation, PollingStrategy<T,U> strategy, TypeReference<T> pollResponseType, TypeReference<U> resultType) Creates PollerFlux.
static PollerFlux<T,U>	<T,U>error(Exception ex) Creates a PollerFlux instance that returns an error on subscription.
Duration	getPollInterval() Returns the current polling duration for this PollerFlux<T,U> instance.
SyncPoller<T,U>	getSyncPoller() Gets a synchronous blocking poller.
PollerFlux<T,U>	setPollInterval(Duration pollInterval) Sets the poll interval for this poller.
void	subscribe(CoreSubscriber<? super AsyncPollResponse<T,U>> actual)

Methods inherited from java.lang.Object

clone equals finalize getClass hashCode notify notifyAll wait wait wait

Methods inherited from reactor.core.publisher.Flux

reduce reduceWith scan scanWith buffer buffer buffer bufferTimeout bufferTimeout cast collect doOnError onErrorContinue onErrorContinue onErrorMap onErrorResume onErrorReturn subscribeWith zip zip first first firstWithSignal firstWithSignal firstWithValue firstWithValue index merge merge merge mergeComparing mergeDelayError mergeOrdered mergePriority mergeSequential mergeSequential mergeSequential mergeSequential mergeSequentialDelayError mergeSequentialDelayError collectMap collectMap collectMultimap collectMultimap groupBy groupBy collectMap collectMultimap groupBy groupBy error zip zip as collect concatMapIterable concatMapIterable doOnDiscard flatMap flatMap flatMapIterable flatMapIterable flatMapSequential flatMapSequential flatMapSequential flatMapSequentialDelayError handle publish publish using using usingWhen usingWhen generate generate combineLatest combineLatest combineLatest combineLatest zip zip zip combineLatest zip combineLatest zip combineLatest zip combineLatest zip combineLatest zip zipWith zipWith zipWithIterable zipWith zipWith zipWithIterable concat concat concat concat concatDelayError concatDelayError concatDelayError concatDelayError create create defer deferContextual deferWithContext empty error error from fromArray fromIterable fromStream fromStream generate just just merge merge merge mergeComparing mergeComparing mergeComparingDelayError mergeOrdered mergeOrdered mergePriority mergePriority mergePriorityDelayError mergeSequential mergeSequential mergeSequentialDelayError never onAssembly onAssembly push push switchOnNext switchOnNext groupJoin join zip withLatestFrom bufferWhen bufferWhen timeout timeout windowWhen delaySubscription ofType sample sampleFirst sampleTimeout sampleTimeout timeout distinct distinct bufferUntilChanged bufferUntilChanged bufferUntilChanged concatMap concatMap concatMapDelayError concatMapDelayError concatMapDelayError distinct distinctUntilChanged distinctUntilChanged flatMap flatMap flatMapDelayError map mapNotNull switchMap switchMap switchOnFirst switchOnFirst then thenMany transform transformDeferred transformDeferredContextual windowUntilChanged windowUntilChanged windowUntilChanged dematerialize all any blockFirst blockFirst blockLast blockLast buffer buffer buffer buffer buffer buffer buffer buffer bufferTimeout bufferTimeout bufferUntil bufferUntil bufferWhile cache cache cache cache cache cache cancelOn checkpoint checkpoint checkpoint collectList collectSortedList collectSortedList concatWith concatWithValues contextWrite contextWrite count defaultIfEmpty delayElements delayElements delaySequence delaySequence delaySubscription delaySubscription delayUntil distinct distinctUntilChanged doAfterTerminate doFinally doFirst doOnCancel doOnComplete doOnEach doOnError doOnError doOnNext doOnRequest doOnSubscribe doOnTerminate elapsed elapsed elementAt elementAt expand expand expandDeep expandDeep filter filterWhen filterWhen getPrefetch hasElement hasElements hide ignoreElements index interval interval interval interval last last limitRate limitRate limitRequest log log log log log log materialize mergeComparingWith mergeOrderedWith mergeWith metrics name next onBackpressureBuffer onBackpressureBuffer onBackpressureBuffer onBackpressureBuffer onBackpressureBuffer onBackpressureBuffer onBackpressureBuffer onBackpressureDrop onBackpressureDrop onBackpressureError onBackpressureLatest onErrorComplete onErrorComplete onErrorComplete onErrorContinue onErrorMap onErrorMap onErrorResume onErrorResume onErrorReturn onErrorReturn onErrorStop onTerminateDetach or parallel parallel parallel publish publish publishNext publishOn publishOn publishOn range reduce repeat repeat repeat repeat repeatWhen replay replay replay replay replay replay retry retry retryWhen sample sampleFirst scan share shareNext single single singleOrEmpty skip skip skip skipLast skipUntil skipUntilOther skipWhile sort sort startWith startWith startWith subscribe subscribe subscribe subscribe subscribe subscribe subscribe subscribe subscribeOn subscribeOn subscriberContext subscriberContext switchIfEmpty tag take take take take takeLast takeUntil takeUntilOther takeWhile then thenEmpty timed timed timeout timeout timeout timeout timestamp timestamp toIterable toIterable toIterable toStream toStream toString window window window window window window window windowTimeout windowTimeout windowTimeout windowTimeout windowUntil windowUntil windowUntil windowWhile windowWhile

Constructor Details

PollerFlux

public PollerFlux(Duration pollInterval, Function<>,Mono> activationOperation, Function<>,Mono<>>> pollOperation, BiFunction<>,PollResponse,Mono> cancelOperation, Function<>,Mono> fetchResultOperation)

Creates PollerFlux.

Parameters:

pollInterval - the polling interval

activationOperation - the activation operation to activate (start) the long-running operation. This operation will be invoked at most once across all subscriptions. This parameter is required. If there is no specific activation work to be done then invocation should return Mono.empty(), this operation will be called with a new PollingContext<T>.

pollOperation - the operation to poll the current state of long-running operation. This parameter is required and the operation will be called with current PollingContext<T>.

cancelOperation - a Function that represents the operation to cancel the long-running operation if service supports cancellation. This parameter is required. If service does not support cancellation then the implementer should return Mono#errorwith an error message indicating absence of cancellation support. The operation will be called with current PollingContext<T>.

fetchResultOperation - a Function that represents the operation to retrieve final result of the long-running operation if service support it. This parameter is required and operation will be called with the current PollingContext<T>. If service does not have an api to fetch final result and if final result is same as final poll response value then implementer can choose to simply return value from provided final poll response.

Method Details

<T,U>create

public static PollerFlux create(Duration pollInterval, Function<>,Mono<>>> activationOperation, Function<>,Mono<>>> pollOperation, BiFunction<>,PollResponse,Mono> cancelOperation, Function<>,Mono> fetchResultOperation)

Creates PollerFlux.

This method differs from the PollerFlux constructor in that the constructor uses an activationOperation which returns a Mono that emits result, the create method uses an activationOperation which returns a Mono that emits PollResponse<T>. The PollResponse<T> holds the result. If the PollResponse<T> from the activationOperation indicate that long-running operation is completed then the pollOperation will not be called.

Parameters:

pollInterval - the polling interval

pollOperation - the operation to poll the current state of long-running operation. This parameter is required and the operation will be called with current PollingContext<T>.

cancelOperation - a Function that represents the operation to cancel the long-running operation if service supports cancellation. This parameter is required. If service does not support cancellation then the implementer should return Mono#error with an error message indicating absence of cancellation support. The operation will be called with current PollingContext<T>.

fetchResultOperation - a Function that represents the operation to retrieve final result of the long-running operation if service support it. This parameter is required and operation will be called current PollingContext<T>. If service does not have an api to fetch final result and if final result is same as final poll response value then implementer can choose to simply return value from provided final poll response.

Returns:

PollerFlux

<T,U>create

public static PollerFlux create(Duration pollInterval, Supplier<>>> initialOperation, PollingStrategy strategy, TypeReference pollResponseType, TypeReference resultType)

Creates PollerFlux.

This method uses a PollingStrategy<T,U> to poll the status of a long-running operation after the activation operation is invoked. See PollingStrategy<T,U> for more details of known polling strategies and how to create a custom strategy.

Parameters:

pollInterval - the polling interval

initialOperation - the activation operation to activate (start) the long-running operation. This operation will be invoked at most once across all subscriptions. This parameter is required. If there is no specific activation work to be done then invocation should return Mono.empty(), this operation will be called with a new PollingContext<T>.

strategy - a known strategy for polling a long-running operation in Azure

pollResponseType - the TypeReference<T> of the response type from a polling call, or BinaryData if raw response body should be kept. This should match the generic parameter U.

resultType - the TypeReference<T> of the final result object to deserialize into, or BinaryData if raw response body should be kept. This should match the generic parameter U.

Returns:

PollerFlux

<T,U>error

public static PollerFlux error(Exception ex)

Creates a PollerFlux instance that returns an error on subscription.

Parameters:

ex - The exception to be returned on subscription of this PollerFlux<T,U>.

Returns:

A poller flux instance that returns an error without emitting any data.

getPollInterval

public Duration getPollInterval()

Returns the current polling duration for this PollerFlux<T,U> instance.

Returns:

The current polling duration.

getSyncPoller

public SyncPoller getSyncPoller()

Gets a synchronous blocking poller.

Returns:

a synchronous blocking poller.

setPollInterval

public PollerFlux setPollInterval(Duration pollInterval)

Sets the poll interval for this poller. The new interval will be used for all subsequent polling operations including the subscriptions that are already in progress.

Parameters:

pollInterval - The new poll interval for this poller.

Returns:

The updated instance of PollerFlux<T,U>.

public void subscribe(CoreSubscriber> actual)

Overrides:

PollerFlux<T,U>.subscribe(CoreSubscriber<? super AsyncPollResponse<T,U>> actual)

Parameters:

actual

PollerFlux<T,U> Class

Constructor Summary

Method Summary

Methods inherited from java.lang.Object

Methods inherited from reactor.core.publisher.Flux

Constructor Details

PollerFlux

Method Details

<T,U>create

<T,U>create

<T,U>error

getPollInterval

getSyncPoller

setPollInterval

Applies to

Feedback

Additional resources