Class PollerFlux<T,U>

java.lang.Object
reactor.core.publisher.Flux<AsyncPollResponse<T,U>>
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.
All Implemented Interfaces:
org.reactivestreams.Publisher<AsyncPollResponse<T,U>>, CorePublisher<AsyncPollResponse<T,U>>

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 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 Details

    • PollerFlux

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

    • create

      public static <T, U> PollerFlux<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. This create 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. The PollResponse holds the result. If the PollResponse from the activationOperation indicate that long running operation is completed then the pollOperation will not be called.
      Type Parameters:
      T - The type of poll response value.
      U - The type of the final result of long running operation.
      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.
      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.
      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.
      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. 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
    • create

      public static <T, U> PollerFlux<T,U> create(Duration pollInterval, Supplier<Mono<? extends Response<?>>> initialOperation, PollingStrategy<T,U> strategy, TypeReference<T> pollResponseType, TypeReference<U> resultType)
      Creates PollerFlux. This create method uses a PollingStrategy to poll the status of a long running operation after the activation operation is invoked. See PollingStrategy for more details of known polling strategies and how to create a custom strategy.
      Type Parameters:
      T - The type of poll response value.
      U - The type of the final result of long running operation.
      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.
      strategy - a known strategy for polling a long running operation in Azure
      pollResponseType - the TypeReference of the response type from a polling call, or BinaryData if raw response body should be kept. This should match the generic parameter PollerFlux.
      resultType - the TypeReference of the final result object to deserialize into, or BinaryData if raw response body should be kept. This should match the generic parameter PollerFlux.
      Returns:
      PollerFlux
    • error

      public static <T, U> PollerFlux<T,U> error(Exception ex)
      Creates a PollerFlux instance that returns an error on subscription.
      Type Parameters:
      T - The type of poll response value.
      U - The type of the final result of long running operation.
      Parameters:
      ex - The exception to be returned on subscription of this PollerFlux.
      Returns:
      A poller flux instance that returns an error without emitting any data.
      See Also:
    • setPollInterval

      public PollerFlux<T,U> 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.
      Throws:
      NullPointerException - if the pollInterval is null.
      IllegalArgumentException - if the pollInterval is zero or negative.
    • getPollInterval

      public Duration getPollInterval()
      Returns the current polling duration for this PollerFlux instance.
      Returns:
      The current polling duration.
    • subscribe

      public void subscribe(CoreSubscriber<? super AsyncPollResponse<T,U>> actual)
      Specified by:
      subscribe in interface CorePublisher<T>
      Specified by:
      subscribe in class Flux<AsyncPollResponse<T,U>>
    • getSyncPoller

      public SyncPoller<T,U> getSyncPoller()
      Returns:
      a synchronous blocking poller.