PagedFluxBase<T,P> Class

public PagedFluxBase(Supplier<Mono

> firstPageRetriever)

Creates an instance of PagedFluxBase<T,P> that consists of only a single page. This constructor takes a Supplier that return the single page of T.

Code sample

// A supplier that fetches the first page of data from source/service
 Supplier<Mono<PagedResponse<Integer>>> firstPageRetrieverFunction = () -> getFirstPage();

 PagedFluxBase<Integer, PagedResponse<Integer>> pagedFluxBaseInstance =
     new PagedFluxBase<>(firstPageRetrieverFunction,
         nextPageRetriever);

Parameters:

public PagedFluxBase(Supplier<Mono

> firstPageRetriever, Function<String,Mono

> nextPageRetriever)

Creates an instance of PagedFluxBase<T,P>. The constructor takes a Supplier and Function. The Supplier returns the first page of T, the Function retrieves subsequent pages of T.

Code sample

// A supplier that fetches the first page of data from source/service
 Supplier<Mono<PagedResponse<Integer>>> firstPageRetriever = () -> getFirstPage();

 // A function that fetches subsequent pages of data from source/service given a continuation token
 Function<String, Mono<PagedResponse<Integer>>> nextPageRetriever =
     continuationToken -> getNextPage(continuationToken);

 PagedFluxBase<Integer, PagedResponse<Integer>> pagedFluxBase = new PagedFluxBase<>(firstPageRetriever,
     nextPageRetriever);

Parameters:

public Flux

byPage()

Creates a Flux of PagedResponse<T> starting from the first page.

Code sample

// Start processing the results from first page
 pagedFluxBase.byPage()
     .log()
     .doOnSubscribe(ignoredVal -> System.out.println(
         "Subscribed to paged flux processing pages starting from first page"))
     .subscribe(page -> System.out.printf("Processing page containing item values: %s%n",
         page.getElements().stream().map(String::valueOf).collect(Collectors.joining(", "))),
         error -> System.err.println("An error occurred: " + error),
         () -> System.out.println("Processing complete."));

Overrides:

Returns:

public Flux

byPage(String continuationToken)

Creates a Flux of PagedResponse<T> starting from the next page associated with the given continuation token. To start from first page, use byPage() instead.

Code sample

// Start processing the results from a page associated with the continuation token
 String continuationToken = getContinuationToken();
 pagedFluxBase.byPage(continuationToken)
     .log()
     .doOnSubscribe(ignoredVal -> System.out.println(
         "Subscribed to paged flux processing page starting from " + continuationToken))
     .subscribe(page -> System.out.printf("Processing page containing item values: %s%n",
         page.getElements().stream().map(String::valueOf).collect(Collectors.joining(", "))),
         error -> System.err.println("An error occurred: " + error),
         () -> System.out.println("Processing complete."));

Overrides:

Parameters:

Returns:

public void subscribe(CoreSubscriber<? super T> coreSubscriber)

Subscribe to consume all items of type T in the sequence respectively. This is recommended for most common scenarios. This will seamlessly fetch next page when required and provide with a Flux of items.

Code sample

pagedFluxBase.subscribe(new BaseSubscriber<Integer>() {
     @Override
     protected void hookOnSubscribe(Subscription subscription) {
         System.out.println("Subscribed to paged flux processing items");
         super.hookOnSubscribe(subscription);
     }

     @Override
     protected void hookOnNext(Integer value) {
         System.out.println("Processing item with value: " + value);
     }

     @Override
     protected void hookOnComplete() {
         System.out.println("Processing complete.");
     }
 });

Overrides:

Parameters: