ConnectionFactory.create() supports multiple subscribers

Author: Michael-A-McMahon

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

@@ -53,13 +53,21 @@
  *   <li>{@link ConnectionFactoryOptions#HOST}</li>
  * </ul>
  * <h3 id="supported_options">Supported Options</h3><p>
- * This implementation supports the following options for connection creation:
+ * This implementation supports the following well known options for connection
+ * creation:
  * </p><ul>
  *   <li>{@link ConnectionFactoryOptions#PORT}</li>
  *   <li>{@link ConnectionFactoryOptions#DATABASE}</li>
  *   <li>{@link ConnectionFactoryOptions#USER}</li>
  *   <li>{@link ConnectionFactoryOptions#PASSWORD}</li>
+ *   <li>{@link ConnectionFactoryOptions#CONNECT_TIMEOUT}</li>
+ *   <li>{@link ConnectionFactoryOptions#SSL}</li>
  * </ul>
+ * <h3 id="extended_options">Supported Options</h3><p>
+ * This implementation supports extended options having the name of a
+ * subset of Oracle JDBC connection properties. The list of supported
+ * connection properties is specified by {@link OracleReactiveJdbcAdapter}.
+ * </p>
  *
  * @author  harayuanwang, michael-a-mcmahon
  * @since   0.1.0
@@ -116,7 +124,6 @@ final class OracleConnectionFactoryImpl implements ConnectionFactory {
    */
   OracleConnectionFactoryImpl(ConnectionFactoryOptions options) {
     OracleR2dbcExceptions.requireNonNull(options, "options is null.");
-
     adapter = ReactiveJdbcAdapter.getOracleAdapter();
     dataSource = adapter.createDataSource(options);
   }
@@ -138,14 +145,15 @@ final class OracleConnectionFactoryImpl implements ConnectionFactory {
    * the returned publisher, so that the database can reclaim the resources
    * allocated for that connection.
    * </p><p>
-   * The returned publisher does not support multiple subscribers. After a
-   * subscriber has subscribed, the returned publisher emits {@code onError}
-   * with an {@link IllegalStateException} to all subsequent subscribers.
+   * The returned publisher supports multiple subscribers. One {@code
+   * Connection} is emitted to each subscriber that subscribes and signals
+   * demand.
    * </p>
    */
   @Override
   public Publisher<Connection> create() {
-    return Mono.fromDirect(adapter.publishConnection(dataSource))
+    return Mono.defer(() ->
+        Mono.fromDirect(adapter.publishConnection(dataSource)))
       .map(conn -> new OracleConnectionImpl(adapter, conn));
   }
 

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

@@ -33,6 +33,8 @@
 
 import java.sql.SQLException;
 import java.time.Duration;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -115,21 +117,24 @@ public void testCreate() {
             .build())
         .create();
 
-    // Expect publisher to emit 1 connection
-    AtomicInteger counter = new AtomicInteger(0);
+    // Expect publisher to emit one connection to each subscriber
+    Set<Connection> connections = new HashSet<>();
     Flux.from(connectionPublisher)
-      .doOnNext(connection -> counter.incrementAndGet())
+      .doOnNext(connections::add)
       .doOnNext(connection -> Mono.from(connection.close()).subscribe())
       .blockLast(DatabaseConfig.connectTimeout());
-    assertEquals(1, counter.get());
+    assertEquals(1, connections.size());
+    Flux.from(connectionPublisher)
+      .doOnNext(connections::add)
+      .doOnNext(connection -> Mono.from(connection.close()).subscribe())
+      .blockLast(DatabaseConfig.connectTimeout());
+    assertEquals(2, connections.size());
+    Flux.from(connectionPublisher)
+      .doOnNext(connections::add)
+      .doOnNext(connection -> Mono.from(connection.close()).subscribe())
+      .blockLast(DatabaseConfig.connectTimeout());
+    assertEquals(3, connections.size());
 
-    // Expect publisher to reject multiple subscribers
-    try {
-      Mono.from(connectionPublisher)
-        .block(Duration.ofSeconds(1));
-      fail("Connection publisher did not reject multiple subscribers");
-    }
-    catch (IllegalStateException expected) { }
   }
 
   /**