Merge pull request #16 from florian-tirard/bugfix/stream-fetch-size

fix: Use param fetchSize in stream query method
MAIF · Sep 17, 2024 · 8bbcc8e · 8bbcc8e
2 parents a4cb86a + a8072f5
commit 8bbcc8e
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/jooq-async-jdbc/src/main/java/fr/maif/jooq/jdbc/JdbcPgAsyncTransaction.java b/jooq-async-jdbc/src/main/java/fr/maif/jooq/jdbc/JdbcPgAsyncTransaction.java
@@ -69,7 +69,7 @@ public CompletionStage<Tuple0> rollback() {
     @Override
     public <Q extends Record> Publisher<QueryResult> stream(Integer fetchSize, Function<DSLContext, ? extends ResultQuery<Q>> queryFunction) {
         return Flux
-                .fromIterable(() -> queryFunction.apply(client).stream().iterator())
+                .fromIterable(() -> queryFunction.apply(client).fetchSize(fetchSize).stream().iterator())
                 .publishOn(Schedulers.parallel())
                 .map(JooqQueryResult::new);
     }

diff --git a/jooq-async-reactive/src/main/java/fr/maif/jooq/reactive/ReactivePgAsyncTransaction.java b/jooq-async-reactive/src/main/java/fr/maif/jooq/reactive/ReactivePgAsyncTransaction.java
@@ -60,7 +60,7 @@ public <Q extends Record> Flux<QueryResult> stream(Integer fetchSize, Function<D
                         cursor -> Mono.just(List.<QueryResult>empty())
                         .expand(results -> {
                             if (first.getAndSet(false) || cursor.hasMore()) {
-                                return Mono.fromCompletionStage(cursor.read(500).map(rs ->
+                                return Mono.fromCompletionStage(cursor.read(fetchSize).map(rs ->
                                         List.ofAll(rs)
                                                 .map(ReactiveRowQueryResult::new)
                                                 .map(r -> (QueryResult)r)