Remove Result.consumedFuture

Author: Michael-A-McMahon

src/main/java/oracle/r2dbc/impl/OracleBatchImpl.java

@@ -22,6 +22,7 @@
 package oracle.r2dbc.impl;
 
 import java.sql.Connection;
+import java.sql.Statement;
 import java.time.Duration;
 import java.util.LinkedList;
 import java.util.Queue;
@@ -114,18 +115,6 @@ public Batch add(String sql) {
    * are executed in the order they were added. Calling this method clears all
    * statements that have been added to the current batch.
    * </p><p>
-   * A {@code Result} emitted by the returned {@code Publisher} must be
-   * <a href="OracleStatementImpl.html#fully-consumed-result">
-   *   fully-consumed
-   * </a>
-   * before the next {@code Result} is emitted. This ensures that a command in
-   * the batch can not be executed while the {@code Result} of a previous
-   * command is consumed concurrently. It is a known limitation of the Oracle
-   * R2DBC Driver that concurrent operations on a single {@code Connection}
-   * will result in blocked threads. Deferring {@code Statement} execution
-   * until full consumption of the previous {@code Statement}'s {@code Result}
-   * is necessary in order to avoid blocked threads.
-   * </p><p>
    * If the execution of any statement in the sequence results in a failure,
    * then the returned publisher emits {@code onError} with an
    * {@link R2dbcException} that describes the failure, and all subsequent
@@ -147,42 +136,8 @@ public Publisher<OracleResultImpl> execute() {
     requireOpenConnection(jdbcConnection);
     Queue<OracleStatementImpl> currentStatements = statements;
     statements = new LinkedList<>();
-    return publishBatch(currentStatements);
-  }
-
-  /**
-   * <p>
-   * Publish a batch of {@code Result}s from {@code statements}. Each
-   * {@code Result} is published serially with the consumption of the
-   * previous {@code Result}.
-   * </p><p>
-   * This method returns an empty {@code Publisher} if {@code statements} is
-   * empty. Otherwise, this method dequeues the next {@code Statement} and
-   * executes it for a {@code Result}. After the {@code Result} has been
-   * fully consumed, this method is invoked recursively to publish the {@code
-   * Result}s of remaining {@code statements}.
-   * </p>
-   * @param statements A batch to executed.
-   * @return {@code Publisher} of {@code statements} {@code Result}s
-   */
-  private static Publisher<OracleResultImpl> publishBatch(
-    Queue<OracleStatementImpl> statements) {
-
-    OracleStatementImpl next = statements.poll();
-
-    if (next != null) {
-      AtomicReference<OracleResultImpl> lastResult =
-        new AtomicReference<>(null);
-
-      return Flux.from(next.execute())
-        .doOnNext(lastResult::set)
-        .concatWith(Flux.defer(() ->
-          Mono.from(lastResult.get().onConsumed())
-            .thenMany(publishBatch(statements))));
-    }
-    else {
-      return Mono.empty();
-    }
+    return Flux.fromIterable(currentStatements)
+      .concatMap(OracleStatementImpl::execute);
   }
 
 }

src/main/java/oracle/r2dbc/impl/OracleResultImpl.java

@@ -36,7 +36,6 @@
 import java.sql.ResultSet;
 import java.sql.SQLWarning;
 import java.util.Objects;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
@@ -82,15 +81,6 @@ abstract class OracleResultImpl implements Result {
    */
   private boolean isPublished = false;
 
-  /**
-   * Future that is completed when this {@code Result} has been
-   * <a href="OracleStatementImpl.html#fully-consumed-result">
-   *   fully-consumed
-   * </a>.
-   */
-  private final CompletableFuture<Void> consumedFuture =
-    new CompletableFuture<>();
-
   /**
    * Reference to a publisher that must be subscribed to after all segments of
    * this result have been consumed. The reference is updated to {@code null}
@@ -125,11 +115,10 @@ abstract <T> Publisher<T> publishSegments(
    * as well.
    * </p><p>
    * When the returned publisher terminates with {@code onComplete},
-   * {@code onError}, or {@code cancel}, the {@link #consumedFuture} is
-   * completed and the {@link #onConsumed} publisher is subscribed to. The
-   * {@code onConsumed} reference is update to {@code null} so that
-   * post-consumption calls to {@link #onConsumed(Publisher)} can detect that
-   * this result is already consumed.
+   * {@code onError}, or {@code cancel}, the {@link #onConsumed} publisher is
+   * subscribed to. The {@code onConsumed} reference is updated to {@code null}
+   * so that post-consumption calls to {@link #onConsumed(Publisher)} can detect
+   * that this result is already consumed.
    * </p><p>
    * The returned {@code Publisher} emits {@code onError} with an
    * {@link R2dbcException} if this {@code Result} has a {@link Message} segment
@@ -151,7 +140,6 @@ private <T extends Segment, U> Publisher<U> publishSegments(
     setPublished();
 
     Mono<U> whenConsumed = Mono.defer(() -> {
-      consumedFuture.complete(null);
       Publisher<Void> consumedPublisher = onConsumed.getAndSet(null);
       return consumedPublisher == null
         ? Mono.empty()
@@ -279,18 +267,6 @@ public OracleResultImpl filter(Predicate<Segment> filter) {
     return new FilteredResult(this, filter);
   }
 
-  /**
-   * Returns a {@code Publisher} that emits {@code onComplete} when this
-   * {@code Result} has been
-   * <a href="OracleStatementImpl.html#fully-consumed-result">
-   *   fully-consumed
-   * </a>.
-   * @return {@code Publisher} of this {@code Result}'s consumption
-   */
-  final Publisher<Void> onConsumed() {
-    return Mono.fromCompletionStage(consumedFuture);
-  }
-
   /**
    * <p>
    * Sets a publisher that is subscribed to when all segments of this result