Merge branch 'main' of github.com:oracle/oracle-r2dbc into 84-option-publisher

Author: Michael-A-McMahon

src/main/java/oracle/r2dbc/OracleR2dbcOptions.java

@@ -58,9 +58,11 @@ private OracleR2dbcOptions() {}
    *   ...
    *   .build();
    * </pre>
-   * If this option is not configured, then Oracle R2DBC will use
-   * {@code ForkJoinPool}'s
-   * {@linkplain ForkJoinPool#commonPool() common pool} by default.
+   * If this option is not configured, then Oracle R2DBC will
+   * use the {@linkplain ForkJoinPool#commonPool() common ForkJoinPool} by
+   * default. However, if the common {@code ForkJoinPool} has a maximum pool
+   * size that is potentially zero, then a single-threaded {@code Executor} will
+   * be used by default.
    */
   public static final Option<Executor> EXECUTOR;
 
@@ -370,6 +372,30 @@ private OracleR2dbcOptions() {}
    */
   public static final Option<CharSequence> NET_ENCRYPTION_TYPES;
 
+  /**
+   * Configures the Oracle JDBC Connection used by Oracle R2DBC as specified by:
+   * {@link OracleConnection#CONNECTION_PROPERTY_THIN_NET_AUTHENTICATION_KRB5_CC_NAME}
+   */
+  public static final Option<CharSequence> KERBEROS_CREDENTIAL_CACHE_NAME;
+
+  /**
+   * Configures the Oracle JDBC Connection used by Oracle R2DBC as specified by:
+   * {@link OracleConnection#CONNECTION_PROPERTY_THIN_NET_AUTHENTICATION_KRB5_MUTUAL}
+   */
+  public static final Option<CharSequence> KERBEROS_MUTUAL_AUTHENTICATION;
+
+  /**
+   * Configures the Oracle JDBC Connection used by Oracle R2DBC as specified by:
+   * {@link OracleConnection#CONNECTION_PROPERTY_THIN_NET_AUTHENTICATION_KRB_REALM}
+   */
+  public static final Option<CharSequence> KERBEROS_REALM;
+
+  /**
+   * Configures the Oracle JDBC Connection used by Oracle R2DBC as specified by:
+   * {@link OracleConnection#CONNECTION_PROPERTY_THIN_NET_AUTHENTICATION_KRB_JAAS_LOGIN_MODULE}
+   */
+  public static final Option<CharSequence> KERBEROS_JAAS_LOGIN_MODULE;
+
 
   /** The unmodifiable set of all extended options */
   private static final Set<Option<?>> OPTIONS = Set.of(
@@ -476,7 +502,15 @@ private OracleR2dbcOptions() {}
     NET_ENCRYPTION_LEVEL = Option.valueOf(
       OracleConnection.CONNECTION_PROPERTY_THIN_NET_ENCRYPTION_LEVEL),
     NET_ENCRYPTION_TYPES = Option.valueOf(
-      OracleConnection.CONNECTION_PROPERTY_THIN_NET_ENCRYPTION_TYPES)
+      OracleConnection.CONNECTION_PROPERTY_THIN_NET_ENCRYPTION_TYPES),
+    KERBEROS_CREDENTIAL_CACHE_NAME = Option.valueOf(
+      OracleConnection.CONNECTION_PROPERTY_THIN_NET_AUTHENTICATION_KRB5_CC_NAME),
+    KERBEROS_MUTUAL_AUTHENTICATION = Option.valueOf(
+      OracleConnection.CONNECTION_PROPERTY_THIN_NET_AUTHENTICATION_KRB5_MUTUAL),
+    KERBEROS_REALM = Option.valueOf(
+      OracleConnection.CONNECTION_PROPERTY_THIN_NET_AUTHENTICATION_KRB_REALM),
+    KERBEROS_JAAS_LOGIN_MODULE = Option.valueOf(
+      OracleConnection.CONNECTION_PROPERTY_THIN_NET_AUTHENTICATION_KRB_JAAS_LOGIN_MODULE)
   );
 
   /**

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

@@ -102,6 +102,20 @@
  */
 final class OracleConnectionFactoryImpl implements ConnectionFactory {
 
+  /**
+   * <p>
+   * The default executor when {@link OracleR2dbcOptions#EXECUTOR} is not
+   * configured. It will use the common {@code ForkJoinPool}, unless it has
+   * a maximum pool size of 0. See:
+   * https://github.com/oracle/oracle-r2dbc/issues/129
+   * </p>
+   */
+  private static final Executor DEFAULT_EXECUTOR =
+    "0".equals(System.getProperty(
+      "java.util.concurrent.ForkJoinPool.common.parallelism"))
+      ? new ForkJoinPool(1)
+      : ForkJoinPool.commonPool();
+
   /** JDBC data source that this factory uses to open connections */
   private final DataSource dataSource;
 
@@ -200,7 +214,7 @@ final class OracleConnectionFactoryImpl implements ConnectionFactory {
 
     Object executor = options.getValue(OracleR2dbcOptions.EXECUTOR);
     if (executor == null) {
-      this.executor = ForkJoinPool.commonPool();
+      this.executor = DEFAULT_EXECUTOR;
     }
     else if (executor instanceof Executor) {
       this.executor = (Executor) executor;
@@ -267,4 +281,5 @@ public Publisher<Connection> create() {
   public ConnectionFactoryMetadata getMetadata() {
     return OracleConnectionFactoryMetadataImpl.INSTANCE;
   }
+
 }

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

@@ -925,14 +925,11 @@ void addDependent() {
     protected final <T extends Segment, U> Publisher<U> mapSegments(
       Class<T> segmentType, Function<? super T, U> segmentMapper) {
 
-      @SuppressWarnings("unchecked")
-      Publisher<U> removeDependent = (Publisher<U>) dependentCounter.decrement();
+      Publisher<Void> removeDependent = dependentCounter.decrement();
 
-      return Flux.concatDelayError(
+      return Publishers.concatTerminal(
         mapDependentSegments(segmentType, segmentMapper),
-        removeDependent)
-        .doOnCancel(() ->
-          Mono.from(removeDependent).subscribe());
+        removeDependent);
     }
 
     /**

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

@@ -1026,18 +1026,16 @@ private JdbcStatement(PreparedStatement preparedStatement, Object[] binds) {
      */
     final Publisher<OracleResultImpl> execute() {
 
-      Mono<OracleResultImpl> deallocate =
-        Mono.from(deallocate()).cast(OracleResultImpl.class);
+      Publisher<Void> deallocate = deallocate();
 
-      return Flux.concatDelayError(
+      return Publishers.concatTerminal(
         Mono.from(bind())
           .thenMany(executeJdbc())
           .map(this::getWarnings)
           .onErrorResume(R2dbcException.class, r2dbcException ->
             Mono.just(createErrorResult(r2dbcException)))
           .doOnNext(OracleResultImpl::addDependent),
-          deallocate)
-        .doOnCancel(deallocate::subscribe);
+        deallocate);
     }
 
     /**

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

@@ -0,0 +1,78 @@
+/*
+  Copyright (c) 2020, 2022, Oracle and/or its affiliates.
+
+  This software is dual-licensed to you under the Universal Permissive License
+  (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl or Apache License
+  2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose
+  either license.
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+     https://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+*/
+
+package oracle.r2dbc.impl;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * Factory methods that create a {@code Publisher}. These methods cover special
+ * cases which are not already supported by Project Reactor.
+ */
+class Publishers {
+
+  private Publishers() {}
+
+  /**
+   * A publisher that immediately emits onNext and onComplete to subscribers
+   */
+  private static final Publisher<Object> COMPLETED_PUBLISHER =
+    Mono.just(new Object());
+
+  /**
+   * <p>
+   * Returns a publisher that emits the concatenated signals of a
+   * {@code publisher} and {@code onTerminationPublisher}. If the
+   * {@code onTerminationPublisher} emits an error, it will suppress any error
+   * emitted by the first {@code publisher}. If a subscription to the returned
+   * publisher is cancelled, the {@code onTerminationPublisher} is subscribed to
+   * but it can not emit any error through the cancelled subscription.
+   * </p><p>
+   * The returned publisher behaves similarly to: <pre>{@code
+   * Flux.concatDelayError(
+   *   publisher,
+   *   onTerminationPublisher)
+   *   .doOnCancel(onTerminationPublisher::subscribe)
+   * }</pre>
+   * However, the code above can result in:
+   * <pre>
+   *   reactor.core.Exceptions$StaticThrowable: Operator has been terminated
+   * </pre>
+   * This seems to happen when the concatDelayError publisher receives a cancel
+   * from a downstream subscriber after it has already received onComplete from
+   * a upstream publisher.
+   * </p>
+   * @param publisher First publisher which is subscribed to.
+   * @param onTerminationPublisher Publisher which is subscribed to when the
+   * first publisher terminates, or a subcription is cancelled.
+   * @return The concatenated publisher.
+   * @param <T> Type of objects emitted to onNext
+   */
+  static <T> Publisher<T> concatTerminal(
+    Publisher<T> publisher, Publisher<Void> onTerminationPublisher) {
+    return Flux.usingWhen(
+      COMPLETED_PUBLISHER,
+      ignored -> publisher,
+      ignored -> onTerminationPublisher);
+  }
+}

src/test/java/oracle/r2dbc/impl/OracleStatementImplTest.java

@@ -3222,7 +3222,7 @@ private void verifyConcurrentFetch(Connection connection) {
       // Create many statements and execute them in parallel.
       @SuppressWarnings({"unchecked","rawtypes"})
       Publisher<Long>[] publishers =
-        new Publisher[Runtime.getRuntime().availableProcessors() * 4];
+        new Publisher[Runtime.getRuntime().availableProcessors() * 2];
 
       for (int i = 0; i < publishers.length; i++) {