-
Notifications
You must be signed in to change notification settings - Fork 2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Decouple PagedIterable and PagedFlux to Prevent Background Page Reque…
…sts (#15646) Decouples PagedIterable from PagedFlux and resolves an issue where more pages would be requested than expected when using PagedIterable.
- Loading branch information
1 parent
4a2713b
commit af97f53
Showing
15 changed files
with
808 additions
and
243 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
80 changes: 80 additions & 0 deletions
80
...e/azure-core/src/main/java/com/azure/core/util/paging/ContinuablePagedByItemIterable.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,80 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT License. | ||
|
|
||
| package com.azure.core.util.paging; | ||
|
|
||
| import com.azure.core.util.logging.ClientLogger; | ||
|
|
||
| import java.util.Iterator; | ||
| import java.util.Queue; | ||
| import java.util.concurrent.ConcurrentLinkedQueue; | ||
|
|
||
| /** | ||
| * Internal class that is a blocking iterable for {@link ContinuablePagedIterable}. | ||
| * <p> | ||
| * This class retrieves pages from the service in a blocking manner while also respecting the number of items to be | ||
| * retrieved. This functions differently than just wrapping a {@link ContinuablePagedFlux} as this will track the exact | ||
| * number of items emitted and whether the previously retrieve page/pages contain any additional items that could be | ||
| * emitted. | ||
| * | ||
| * @param <C> The continuation token type. | ||
| * @param <T> The item type. | ||
| * @param <P> The page type. | ||
| */ | ||
| final class ContinuablePagedByItemIterable<C, T, P extends ContinuablePage<C, T>> implements Iterable<T> { | ||
| private final PageRetriever<C, P> pageRetriever; | ||
| private final C continuationToken; | ||
| private final Integer preferredPageSize; | ||
|
|
||
| ContinuablePagedByItemIterable(PageRetriever<C, P> pageRetriever, C continuationToken, Integer preferredPageSize) { | ||
| this.pageRetriever = pageRetriever; | ||
| this.continuationToken = continuationToken; | ||
| this.preferredPageSize = preferredPageSize; | ||
| } | ||
|
|
||
| @Override | ||
| public Iterator<T> iterator() { | ||
| return new ContinuablePagedByItemIterator<>(pageRetriever, continuationToken, preferredPageSize); | ||
| } | ||
|
|
||
| private static final class ContinuablePagedByItemIterator<C, T, P extends ContinuablePage<C, T>> | ||
| extends ContinuablePagedByIteratorBase<C, T, P, T> { | ||
| private volatile Queue<Iterator<T>> pages = new ConcurrentLinkedQueue<>(); | ||
| private volatile Iterator<T> currentPage; | ||
|
|
||
| ContinuablePagedByItemIterator(PageRetriever<C, P> pageRetriever, C continuationToken, | ||
| Integer preferredPageSize) { | ||
| super(pageRetriever, new ContinuationState<>(continuationToken), preferredPageSize, | ||
| new ClientLogger(ContinuablePagedByItemIterator.class)); | ||
|
|
||
| requestPage(); | ||
| } | ||
|
|
||
| @Override | ||
| boolean needToRequestPage() { | ||
| return (currentPage == null || !currentPage.hasNext()) && pages.peek() == null; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isNextAvailable() { | ||
| return (currentPage != null && currentPage.hasNext()) || pages.peek() != null; | ||
| } | ||
|
|
||
| @Override | ||
| T getNext() { | ||
| if ((currentPage == null || !currentPage.hasNext()) && pages.peek() != null) { | ||
| currentPage = pages.poll(); | ||
| } | ||
|
|
||
| return currentPage.next(); | ||
| } | ||
|
|
||
| @Override | ||
| void addPage(P page) { | ||
| Iterator<T> pageValues = page.getElements().iterator(); | ||
| if (pageValues.hasNext()) { | ||
| this.pages.add(pageValues); | ||
| } | ||
| } | ||
| } | ||
| } |
107 changes: 107 additions & 0 deletions
107
...e/azure-core/src/main/java/com/azure/core/util/paging/ContinuablePagedByIteratorBase.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,107 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT License. | ||
|
|
||
| package com.azure.core.util.paging; | ||
|
|
||
| import com.azure.core.util.logging.ClientLogger; | ||
|
|
||
| import java.util.Iterator; | ||
| import java.util.NoSuchElementException; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
|
|
||
| /** | ||
| * Internal class that is a blocking iterator base class. | ||
| * <p> | ||
| * This class manages retrieving and maintaining previously retrieve page/pages in a synchronous fashion. It will ensure | ||
| * the minimum number of pages are retrieved from a service by checking if any additional items/pages could be emitted | ||
| * before requesting additional ones from the service. | ||
| * | ||
| * @param <C> The continuation token type. | ||
| * @param <T> The item type. | ||
| * @param <P> The page type. | ||
| * @param <E> The type that the {@link ContinuablePagedIterable} will emit. | ||
| */ | ||
| abstract class ContinuablePagedByIteratorBase<C, T, P extends ContinuablePage<C, T>, E> implements Iterator<E> { | ||
| private final PageRetriever<C, P> pageRetriever; | ||
| private final ContinuationState<C> continuationState; | ||
| private final Integer defaultPageSize; | ||
| private final ClientLogger logger; | ||
|
|
||
| private volatile boolean done; | ||
|
|
||
| ContinuablePagedByIteratorBase(PageRetriever<C, P> pageRetriever, ContinuationState<C> continuationState, | ||
| Integer defaultPageSize, ClientLogger logger) { | ||
| this.continuationState = continuationState; | ||
| this.pageRetriever = pageRetriever; | ||
| this.defaultPageSize = defaultPageSize; | ||
| this.logger = logger; | ||
| } | ||
|
|
||
| @Override | ||
| public E next() { | ||
| if (!hasNext()) { | ||
| throw logger.logExceptionAsError(new NoSuchElementException("Iterator contains no more elements.")); | ||
| } | ||
|
|
||
| return getNext(); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean hasNext() { | ||
| // Request next pages in a loop in case we are returned empty pages for the by item implementation. | ||
| while (!done && needToRequestPage()) { | ||
| requestPage(); | ||
| } | ||
|
|
||
| return isNextAvailable(); | ||
| } | ||
|
|
||
| /* | ||
| * Indicates if a page needs to be requested. | ||
| */ | ||
| abstract boolean needToRequestPage(); | ||
|
|
||
| /* | ||
| * Indicates if another element is available. | ||
| */ | ||
| abstract boolean isNextAvailable(); | ||
|
|
||
| /* | ||
| * Gets the next element to be emitted. | ||
| */ | ||
| abstract E getNext(); | ||
|
|
||
| synchronized void requestPage() { | ||
| /* | ||
| * In the scenario where multiple threads were waiting on synchronization, check that no earlier thread made a | ||
| * request that would satisfy the current element request. Additionally, check to make sure that any earlier | ||
| * requests didn't consume the paged responses to completion. | ||
| */ | ||
| if (isNextAvailable() || done) { | ||
| return; | ||
| } | ||
|
|
||
| AtomicBoolean receivedPages = new AtomicBoolean(false); | ||
| pageRetriever.get(continuationState.getLastContinuationToken(), defaultPageSize) | ||
| .map(page -> { | ||
| receivedPages.set(true); | ||
| addPage(page); | ||
|
|
||
| continuationState.setLastContinuationToken(page.getContinuationToken()); | ||
| this.done = continuationState.isDone(); | ||
|
|
||
| return page; | ||
| }).blockLast(); | ||
|
|
||
| /* | ||
| * In the scenario when the subscription completes without emitting an element indicate we are done by checking | ||
| * if we have any additional elements to return. | ||
| */ | ||
| this.done = done || (!receivedPages.get() && !isNextAvailable()); | ||
| } | ||
|
|
||
| /* | ||
| * Add a page returned by the service and update the continuation state. | ||
| */ | ||
| abstract void addPage(P page); | ||
| } |
Oops, something went wrong.