Skip to content

Reduce Prefetch Limit for PagedIterable and IterableStream #6538

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alzimmermsft merged 4 commits into from
Nov 25, 2019
Merged

Reduce Prefetch Limit for PagedIterable and IterableStream #6538

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d72076c
Updated PagedIterable and IterableStream to prefetch 1 instead of the…
alzimmermsft Nov 25, 2019
207c434
Updated to use constant and commented tests
alzimmermsft Nov 25, 2019
45589f0
Merged in master
alzimmermsft Nov 25, 2019
9c866f3
Fixed linting issue
alzimmermsft Nov 25, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions sdk/core/azure-core/src/main/java/com/azure/core/http/rest/PagedFluxBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,7 @@ private Publisher<T> extractAndFetchT(PagedResponse<T> page) {
if (nextPageLink == null) {
return Flux.fromIterable(page.getItems());
}
return Flux.fromIterable(page.getItems()).concatWith(byT(nextPageLink));
return Flux.fromIterable(page.getItems()).concatWith(Flux.defer(() -> byT(nextPageLink)));
}

/**
Expand All @@ -163,6 +163,6 @@ private Publisher<? extends P> extractAndFetchPage(P page) {
if (nextPageLink == null) {
return Flux.just(page);
}
return Flux.just(page).concatWith(byPage(page.getContinuationToken()));
return Flux.just(page).concatWith(Flux.defer(() -> byPage(page.getContinuationToken())));
}
}
14 changes: 10 additions & 4 deletions sdk/core/azure-core/src/main/java/com/azure/core/http/rest/PagedIterableBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,12 @@
* @see IterableStream
*/
public class PagedIterableBase<T, P extends PagedResponse<T>> extends IterableStream<T> {
/*
* This is the default batch size that will be requested when using stream or iterable by page, this will indicate
* to Reactor how many elements should be prefetched before another batch is requested.
*/
private static final int DEFAULT_BATCH_SIZE = 1;

private final PagedFluxBase<T, P> pagedFluxBase;

/**
Expand All @@ -46,7 +52,7 @@ public PagedIterableBase(PagedFluxBase<T, P> pagedFluxBase) {
* @return {@link Stream} of a Response that extends {@link PagedResponse}
*/
public Stream<P> streamByPage() {
return pagedFluxBase.byPage().toStream();
return pagedFluxBase.byPage().toStream(DEFAULT_BATCH_SIZE);
}

/**
Expand All @@ -58,7 +64,7 @@ public Stream<P> streamByPage() {
* with the continuation token
*/
public Stream<P> streamByPage(String continuationToken) {
return pagedFluxBase.byPage(continuationToken).toStream();
return pagedFluxBase.byPage(continuationToken).toStream(DEFAULT_BATCH_SIZE);
}

/**
Expand All @@ -68,7 +74,7 @@ public Stream<P> streamByPage(String continuationToken) {
* @return {@link Iterable} interface
*/
public Iterable<P> iterableByPage() {
return pagedFluxBase.byPage().toIterable();
return pagedFluxBase.byPage().toIterable(DEFAULT_BATCH_SIZE);
}

/**
Expand All @@ -80,6 +86,6 @@ public Iterable<P> iterableByPage() {
* @return {@link Iterable} interface
*/
public Iterable<P> iterableByPage(String continuationToken) {
return pagedFluxBase.byPage(continuationToken).toIterable();
return pagedFluxBase.byPage(continuationToken).toIterable(DEFAULT_BATCH_SIZE);
}
}
10 changes: 8 additions & 2 deletions sdk/core/azure-core/src/main/java/com/azure/core/util/IterableStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,12 @@
* @see Iterable
*/
public class IterableStream<T> implements Iterable<T> {
/*
* This is the default batch size that will be requested when using stream or iterable by page, this will indicate
* to Reactor how many elements should be prefetched before another batch is requested.
*/
private static final int DEFAULT_BATCH_SIZE = 1;

private final ClientLogger logger = new ClientLogger(IterableStream.class);
private final Flux<T> flux;
private final Iterable<T> iterable;
Expand Down Expand Up @@ -66,7 +72,7 @@ public IterableStream(Iterable<T> iterable) {
*/
public Stream<T> stream() {
if (flux != null) {
return flux.toStream();
return flux.toStream(DEFAULT_BATCH_SIZE);
} else if (iterable != null) {
return StreamSupport.stream(iterable.spliterator(), false);
} else {
Expand All @@ -84,7 +90,7 @@ public Stream<T> stream() {
@Override
public Iterator<T> iterator() {
if (flux != null) {
return flux.toIterable().iterator();
return flux.toIterable(DEFAULT_BATCH_SIZE).iterator();
} else if (iterable != null) {
return iterable.iterator();
} else {
Expand Down
Loading