Open connection validation needs to be inside of the subscribed stream

[elefeint]

Hello,

First, and completely out of scope for a github issue, I've enjoyed both parts of your Reactive Summit presentation.

This is a little bit of a drive-by comment because I am not an Oracle R2DBC driver user, but we've ran into similar scenarios when implementing the Cloud Spanner R2DBC driver. In OracleConnectionImpl (for example when beginning a transaction), the validation of whether the connection is currently open happens in the beginning of the method, outside of the reactive stream creation. That can lead to the validation passing at the time the method is called (when the reactive stream operation chain is defined), but the connection becoming closed by the time the resulting publisher is subscribed to.

[Michael-A-McMahon]

Yup! The synchronous validation isn't going to ensure that the connection is still open when the database call is actually initiated by a subscriber subscribing.

For Oracle R2DBC, this works itself out because we call Oracle JDBC's API once the subscriber subscribes. Oracle JDBC is going to check if the connection is open, and throw a SQLException if it is not. Oracle R2DBC just translates that SQLException into an R2dbcException that gets emitted to Subscriber.onError(...).

So I'm not sure if there's an issue here. This test will emit an R2dbcException to the subscriber:

  static void test() {
    ConnectionFactory connectionFactory = ConnectionFactories.get(
      "r2dbc:oracle://scott:tiger@localhost:1521/pdb21");

    // Connection is opened
    Connection connection =
      Flux.from(connectionFactory.create()).blockLast();

    // Passes the check for an open connection
    Publisher<Void> beginTransactionPublisher = connection.beginTransaction();

    // Closes the connection
    Flux.from(connection.close()).blockLast();

    try {
      Flux.from(beginTransactionPublisher)
        .blockLast();
    }
    catch (R2dbcException r2dbcException) {
      // Expect an error for a closed connection
      r2dbcException.printStackTrace();
    }
  }

It might be good to standardize this behavior though. Perhaps the R2DBC SPI should be explicit about how a Publisher emits R2dbcException if the connection is closed when the subscriber subscribes? Or, maybe it would make sense to emit IllegalStateException to be consistent with the synchronous check?

[elefeint]

This makes sense, thank you for the explanation! The type of exception is likely up to implementation -- IllegalStateException namechecks operating on a closed object specifically, but the general R2DBC exception explanation also fits -- "an error occurs during an interaction with a data source".