Option Values from Supplier and Publisher

README.md

@@ -214,6 +214,93 @@ that is specific to Oracle Database and the Oracle JDBC Driver. Extended options
 are declared in the
 [OracleR2dbcOptions](src/main/java/oracle/r2dbc/OracleR2dbcOptions.java) class.
 
+#### Support for Supplier and Publisher as Option Values
+Most options can have a value provided by a `Supplier` or `Publisher`.
+
+Oracle R2DBC requests the value of an `Option` from a `Supplier` or `Publisher`
+each time the `Publisher` returned by `ConnectionFactory.create()` creates a new
+`Connection`. Each `Connection` can then be configured with values that change 
+over time, such as a password which is periodically rotated.
+
+If a `Supplier` provides the value of an `Option`, then Oracle R2DBC requests 
+the value by invoking `Supplier.get()`. If `get()` returns `null`,
+then no value is configured for the `Option`. If `get()` throws a
+`RuntimeException`, then it is set as the initial cause of an
+`R2dbcException` emitted by the `Publisher` returned by
+`ConnectionFactory.create()`. The `Supplier` must have a thread safe `get()` 
+method, as multiple subscribers may request connections concurrently.
+
+If a `Publisher` provides the value of an `Option`, then Oracle R2DBC requests
+the value by subscribing to the `Publisher` and signalling demand.
+The first value emitted to `onNext` will be used as the value of the `Option`.
+If the `Publisher` emits `onComplete` before `onNext`, then no value is
+configured for the `Option`. If the `Publisher` emits `onError` before `onNext`, 
+then the `Throwable` is set as the initial cause of an
+`R2dbcException` emitted by the `Publisher` returned by
+`ConnectionFactory.create()`.
+
+The following example configures the `PASSWORD` option with a `Supplier`:
+```java
+  void configurePassword(ConnectionFactoryOptions.Builder optionsBuilder) {
+
+    // Cast the PASSWORD option
+    Option<Supplier<CharSequence>> suppliedOption = OracleR2dbcOptions.supplied(PASSWORD);
+
+    // Supply a password
+    Supplier<CharSequence> supplier = () -> getPassword();
+
+    // Configure the builder
+    optionsBuilder.option(suppliedOption, supplier); 
+  }
+```
+A more concise example configures `TLS_WALLET_PASSWORD` as a `Publisher`
+```java
+  void configurePassword(ConnectionFactoryOptions.Builder optionsBuilder) {
+    optionsBuilder.option(
+      OracleR2dbcOptions.published(TLS_WALLET_PASSWORD),
+      Mono.fromSupplier(() -> getWalletPassword()));
+  }
+```
+These examples use the `supplied(Option)` and `published(Option)` methods 
+declared by `oracle.r2dbc.OracleR2dbcOptions`. These methods cast an `Option<T>`
+to `Option<Supplier<T>>` and `Option<Publisher<T>>`, respectively. It is 
+necessary to cast the generic type of the `Option` when calling
+`ConnectionFactoryOptions.Builder.option(Option<T>, T)` in order for the call to
+compile and not throw a `ClassCastException` at runtime. It is not strictly 
+required that `supplied(Option)` or `published(Option)` be used to cast the 
+`Option`. These methods are only meant to offer code readability and
+convenience.
+
+Note that the following code would compile, but fails at runtime with a
+`ClassCastException`:
+```java
+  void configurePassword(ConnectionFactoryOptions.Builder optionsBuilder) {
+    Publisher<CharSequence> publisher = Mono.fromSupplier(() -> getPassword());
+    // Doesn't work. Throws ClassCastException at runtime:
+    optionsBuilder.option(PASSWORD, PASSWORD.cast(publisher));
+  }
+```
+To avoid a `ClassCastException`, the generic type of an `Option` must match the
+actual type of the value passed to 
+`ConnectionFactoryOptions.Builder.option(Option<T>, T)`.
+
+For a small set of options, providing values with a `Supplier` or `Publisher` 
+is not supported:
+- `DRIVER`
+- `PROTOCOL`
+
+Providing values for these options would not be interoperable with
+`io.r2dbc.spi.ConnectionFactories` and `r2dbc-pool`.
+
+Normally, Oracle R2DBC will not retain references to `Option` values after
+`ConnectionFactories.create(ConnectionFactoryOptions)` returns. However, if
+the value of at least one `Option` is provided by a `Supplier` or `Publisher`, 
+then Oracle R2DBC will retain a reference to all `Option` values until the 
+`ConnectionFactory.create()` `Publisher` emits a `Connection` or error. This is
+important to keep in mind when `Option` values may be mutated. In particular,
+a password may only be cleared from memory after the `create()` `Publisher` 
+emits a `Connection` or error.
+
 #### Configuring an Oracle Net Descriptor
 The `oracle.r2dbc.OracleR2dbcOptions.DESCRIPTOR` option may be used to configure
 an Oracle Net Descriptor of the form ```(DESCRIPTION=...)```. If this option is

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

@@ -22,10 +22,12 @@
 
 import io.r2dbc.spi.Option;
 import oracle.jdbc.OracleConnection;
+import org.reactivestreams.Publisher;
 
 import java.util.Set;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ForkJoinPool;
+import java.util.function.Supplier;
 
 /**
  * Extended {@link Option}s supported by the Oracle R2DBC Driver.
@@ -522,4 +524,66 @@ public static Set<Option<?>> options() {
     return OPTIONS;
   }
 
+  /**
+   * <p>
+   * Casts an <code>Option&lt;T&gt;</code> to
+   * <code>Option&lt;Supplier&lt;T&gt;&gt;</code>. For instance, if an
+   * <code>Option&lt;CharSequence&gt;</code> is passed to this method, it is
+   * returned as an
+   * <code>Option&lt;Supplier&lt;CharSequence&gt;&gt;</code>.
+   * </p><p>
+   * This method can used when configuring an <code>Option</code> with values
+   * from a <code>Supplier</code>:
+   * <pre>{@code
+   * void configurePassword(ConnectionFactoryOptions.Builder optionsBuilder) {
+   *   optionsBuilder.option(supplied(PASSWORD), () -> getPassword());
+   * }
+   *
+   * CharSequence getPassword() {
+   *   // ... return a database password ...
+   * }
+   * }</pre>
+   * </p><p>
+   * It is not strictly necessary to use this method when configuring an
+   * <code>Option</code> with a value from a <code>Supplier</code>. This method
+   * is offered for code readability and convenience.
+   * </p>
+   */
+  public static <T> Option<Supplier<T>> supplied(Option<T> option) {
+    @SuppressWarnings("unchecked")
+    Option<Supplier<T>> supplierOption = (Option<Supplier<T>>)option;
+    return supplierOption;
+  }
+
+  /**
+   * <p>
+   * Casts an <code>Option&lt;T&gt;</code> to
+   * <code>Option&lt;Publisher&lt;T&gt;&gt;</code>. For instance, if an
+   * <code>Option&lt;CharSequence&gt;</code> is passed to this method, it
+   * is returned as an
+   * <code>Option&lt;Publisher&lt;CharSequence&gt;&gt;</code>.
+   * </p><p>
+   * This method can used when configuring an <code>Option</code> with values
+   * from a <code>Publisher</code>:
+   * <pre>{@code
+   * void configurePassword(ConnectionFactoryOptions.Builder optionsBuilder) {
+   *   optionsBuilder.option(published(PASSWORD), getPasswordPublisher());
+   * }
+   *
+   * Publisher<CharSequence> getPasswordPublisher() {
+   *   // ... publish a database password ...
+   * }
+   * }</pre>
+   * </p><p>
+   * It is not strictly necessary to use this method when configuring an
+   * <code>Option</code> with a value from a <code>Publisher</code>. This method
+   * is offered for code readability and convenience.
+   * </p>
+   */
+  public static <T> Option<Publisher<T>> published(Option<T> option) {
+    @SuppressWarnings("unchecked")
+    Option<Publisher<T>> publisherOption = (Option<Publisher<T>>)option;
+    return publisherOption;
+  }
+
 }

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

@@ -279,7 +279,7 @@ public Publisher<Connection> create() {
    */
   @Override
   public ConnectionFactoryMetadata getMetadata() {
-    return () -> "Oracle Database";
+    return OracleConnectionFactoryMetadataImpl.INSTANCE;
   }
 
 }

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

@@ -0,0 +1,43 @@
+/*
+  Copyright (c) 2020, 2021, 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 io.r2dbc.spi.ConnectionFactoryMetadata;
+
+/**
+ * Implementation of {@code ConnectionFactoryMetaData} which names
+ * "Oracle Database" as the database product that a
+ * {@link io.r2dbc.spi.ConnectionFactory} connects to.
+ */
+final class OracleConnectionFactoryMetadataImpl
+  implements ConnectionFactoryMetadata {
+
+  static final OracleConnectionFactoryMetadataImpl INSTANCE =
+    new OracleConnectionFactoryMetadataImpl();
+
+  private OracleConnectionFactoryMetadataImpl() {}
+
+  @Override
+  public String getName() {
+    return "Oracle Database";
+  }
+}

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

@@ -96,7 +96,11 @@ public OracleConnectionFactoryProviderImpl() { }
   public ConnectionFactory create(ConnectionFactoryOptions options) {
     assert supports(options) : "Options are not supported: " + options;
     requireNonNull(options, "options must not be null.");
-    return new OracleConnectionFactoryImpl(options);
+
+    if (SuppliedOptionConnectionFactory.containsSuppliedValue(options))
+      return new SuppliedOptionConnectionFactory(options);
+    else
+      return new OracleConnectionFactoryImpl(options);
   }
 
   /**