Added Support for providing configuration option for supplying password function (#157)

Motivation:
Currently r2dbc-mysql does not support IAM based Authentication for
authenticating with AWS Aurora RDS database. The way IAM based
authentication works is requesting token from AWS RDS for that hostname
and username (which is same as AWS IAM Role name). These tokens are
valid for 15 minutes, cannot reuse same token after 15 minutes.
By adding configuration option for supplying password function, whenever
a new connection is made then password is retrieved using supplier
function each time.

Modification:
Modified `MySqlConnectionFactoryProvider` - Added new configurable
option.
`Option<Publisher<String>> PASSWORD_SUPPLIER =
Option.valueOf("passwordSupplier");`
Modified `MySqlConnectionConfiguration` - Added the new configuration
for Password Supplier function.
`Publisher<String> passwordSupplier;`
Modified `MySqlConnectionFactory` - Retrieves Password Supplier function
from configuration, and then retrieves password each time connection
factory is created.

Result:
Users can provide a supplier function using `PASSWORD_SUPPLIER` option.
This function will be used for retrieving password/token each time.

```
public ConnectionFactory writeConnectionFactory(final RdsTokenGenerator rdsTokenGenerator) {
        return ConnectionFactories.get(ConnectionFactoryOptions.builder()
            .option(ConnectionFactoryOptions.DRIVER, "mysql")
            .option(ConnectionFactoryOptions.HOST, "Hostname of AWS Aurora DB instance")
            .option(ConnectionFactoryOptions.PORT, 3306)
            .option(ConnectionFactoryOptions.USER, "IAM ROLE Having access to RDS")
            .option(MySqlConnectionFactoryProvider.PASSWORD_SUPPLIER, rdsTokenGenerator. generateAuthenticationToken())
            .build());
    }
```

Example of `RdsTokenGenerator`
```
public class RdsTokenGenerator {
    public Mono<String> generateAuthenticationToken() {
        return Mono.fromCallable(() -> RdsUtilities.builder()
            .credentialsProvider(DefaultCredentialsProvider.create())
            .region(Region.US_EAST_1)
            .build();
            .generateAuthenticationToken((builder) ->
                builder
                    .hostname(hostname)
                    .port(port)
                    .username(user)
            ))
            .flatMap(token -> LOGGER.info("Retrieved token from RdsUtilities")
                .then(Mono.just(token)))
            .subscribeOn(Schedulers.boundedElastic());
    }

}
```

Author: bhosale

src/main/java/io/asyncer/r2dbc/mysql/MySqlConnectionConfiguration.java

@@ -21,6 +21,7 @@
 import io.asyncer.r2dbc.mysql.extension.Extension;
 import io.netty.handler.ssl.SslContextBuilder;
 import org.jetbrains.annotations.Nullable;
+import org.reactivestreams.Publisher;
 
 import javax.net.ssl.HostnameVerifier;
 import java.net.Socket;
@@ -92,12 +93,15 @@ public final class MySqlConnectionConfiguration {
 
     private final Extensions extensions;
 
+    @Nullable
+    private final Publisher<String> passwordSupplier;
+
     private MySqlConnectionConfiguration(boolean isHost, String domain, int port, MySqlSslConfiguration ssl,
         boolean tcpKeepAlive, boolean tcpNoDelay, @Nullable Duration connectTimeout,
         @Nullable Duration socketTimeout, ZeroDateOption zeroDateOption, @Nullable ZoneId serverZoneId,
         String user, @Nullable CharSequence password, @Nullable String database,
         @Nullable Predicate<String> preferPrepareStatement, int queryCacheSize, int prepareCacheSize,
-        Extensions extensions) {
+        Extensions extensions, @Nullable Publisher<String> passwordSupplier) {
         this.isHost = isHost;
         this.domain = domain;
         this.port = port;
@@ -115,6 +119,7 @@ private MySqlConnectionConfiguration(boolean isHost, String domain, int port, My
         this.queryCacheSize = queryCacheSize;
         this.prepareCacheSize = prepareCacheSize;
         this.extensions = extensions;
+        this.passwordSupplier = passwordSupplier;
     }
 
     /**
@@ -204,6 +209,11 @@ Extensions getExtensions() {
         return extensions;
     }
 
+    @Nullable
+    Publisher<String> getPasswordSupplier() {
+        return passwordSupplier;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -229,14 +239,15 @@ public boolean equals(Object o) {
             Objects.equals(preferPrepareStatement, that.preferPrepareStatement) &&
             queryCacheSize == that.queryCacheSize &&
             prepareCacheSize == that.prepareCacheSize &&
-            extensions.equals(that.extensions);
+            extensions.equals(that.extensions) &&
+            Objects.equals(passwordSupplier, that.passwordSupplier);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(isHost, domain, port, ssl, tcpKeepAlive, tcpNoDelay,
             connectTimeout, socketTimeout, serverZoneId, zeroDateOption, user, password, database,
-            preferPrepareStatement, queryCacheSize, prepareCacheSize, extensions);
+            preferPrepareStatement, queryCacheSize, prepareCacheSize, extensions, passwordSupplier);
     }
 
     @Override
@@ -248,15 +259,15 @@ public String toString() {
                 ", zeroDateOption=" + zeroDateOption + ", user='" + user + '\'' + ", password=" + password +
                 ", database='" + database + "', preferPrepareStatement=" + preferPrepareStatement +
                 ", queryCacheSize=" + queryCacheSize + ", prepareCacheSize=" + prepareCacheSize +
-                ", extensions=" + extensions + '}';
+                ", extensions=" + extensions + ", passwordSupplier="+ passwordSupplier + '}';
         }
 
         return "MySqlConnectionConfiguration{, unixSocket='" + domain + "', connectTimeout=" +
             connectTimeout + ", socketTimeout=" + socketTimeout + ", serverZoneId=" + serverZoneId +
             ", zeroDateOption=" + zeroDateOption + ", user='" + user + "', password=" + password +
             ", database='" + database + "', preferPrepareStatement=" + preferPrepareStatement +
             ", queryCacheSize=" + queryCacheSize + ", prepareCacheSize=" + prepareCacheSize +
-            ", extensions=" + extensions + '}';
+            ", extensions=" + extensions +  ", passwordSupplier="+ passwordSupplier + '}';
     }
 
     /**
@@ -327,6 +338,9 @@ public static final class Builder {
 
         private final List<Extension> extensions = new ArrayList<>();
 
+        @Nullable
+        private Publisher<String> passwordSupplier;
+
         /**
          * Builds an immutable {@link MySqlConnectionConfiguration} with current options.
          *
@@ -351,7 +365,7 @@ public MySqlConnectionConfiguration build() {
             return new MySqlConnectionConfiguration(isHost, domain, port, ssl, tcpKeepAlive, tcpNoDelay,
                 connectTimeout, socketTimeout, zeroDateOption, serverZoneId, user, password, database,
                 preferPrepareStatement, queryCacheSize, prepareCacheSize,
-                Extensions.from(extensions, autodetectExtensions));
+                Extensions.from(extensions, autodetectExtensions), passwordSupplier);
         }
 
         /**
@@ -779,6 +793,16 @@ public Builder extendWith(Extension extension) {
             return this;
         }
 
+        /**
+         * Registers a password supplier function.
+         * @param passwordSupplier function to retrieve password before making connection.
+         * @return this {@link Builder}.
+         */
+        public Builder passwordSupplier(Publisher<String> passwordSupplier) {
+            this.passwordSupplier = passwordSupplier;
+            return this;
+        }
+
         private SslMode requireSslMode() {
             SslMode sslMode = this.sslMode;
 

src/main/java/io/asyncer/r2dbc/mysql/MySqlConnectionFactory.java

@@ -28,11 +28,14 @@
 import io.netty.channel.unix.DomainSocketAddress;
 import io.r2dbc.spi.ConnectionFactory;
 import io.r2dbc.spi.ConnectionFactoryMetadata;
+import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
+import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
 
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.util.Objects;
 import java.util.function.Predicate;
 
 import static io.asyncer.r2dbc.mysql.internal.util.AssertUtils.requireNonNull;
@@ -91,22 +94,54 @@ public static MySqlConnectionFactory from(MySqlConnectionConfiguration configura
             Extensions extensions = configuration.getExtensions();
             Predicate<String> prepare = configuration.getPreferPrepareStatement();
             int prepareCacheSize = configuration.getPrepareCacheSize();
+            Publisher<String> passwordSupplier = configuration.getPasswordSupplier();
+            if (Objects.nonNull(passwordSupplier)) {
+                return Mono.from(passwordSupplier)
+                        .flatMap(token -> getMySqlConnection(
+                            configuration, queryCache,
+                            ssl, address,
+                            database, user,
+                            sslMode, context,
+                            extensions, prepare,
+                            prepareCacheSize, token));
+            }
+            return getMySqlConnection(configuration, queryCache,
+                ssl, address,
+                database, user,
+                sslMode, context,
+                extensions, prepare,
+                prepareCacheSize, password);
+        }));
+    }
 
-            return Client.connect(ssl, address, configuration.isTcpKeepAlive(), configuration.isTcpNoDelay(),
+    @NotNull
+    private static Mono<MySqlConnection> getMySqlConnection(
+            final MySqlConnectionConfiguration configuration,
+            final LazyQueryCache queryCache,
+            final MySqlSslConfiguration ssl,
+            final SocketAddress address,
+            final String database,
+            final String user,
+            final SslMode sslMode,
+            final ConnectionContext context,
+            final Extensions extensions,
+            @Nullable final Predicate<String> prepare,
+            final int prepareCacheSize,
+            @Nullable final CharSequence password) {
+        return Client.connect(ssl, address, configuration.isTcpKeepAlive(), configuration.isTcpNoDelay(),
                 context, configuration.getConnectTimeout(), configuration.getSocketTimeout())
-                .flatMap(client -> QueryFlow.login(client, sslMode, database, user, password, context))
-                .flatMap(client -> {
-                    ByteBufAllocator allocator = client.getByteBufAllocator();
-                    CodecsBuilder builder = Codecs.builder(allocator);
-                    PrepareCache prepareCache = Caches.createPrepareCache(prepareCacheSize);
-
-                    extensions.forEach(CodecRegistrar.class, registrar ->
-                        registrar.register(allocator, builder));
-
-                    return MySqlConnection.init(client, builder.build(), context, queryCache.get(),
-                        prepareCache, prepare);
-                });
-        }));
+            .flatMap(client -> QueryFlow.login(client, sslMode, database, user, password, context))
+            .flatMap(client -> {
+                ByteBufAllocator allocator = client.getByteBufAllocator();
+                CodecsBuilder builder = Codecs.builder(allocator);
+                PrepareCache prepareCache = Caches.createPrepareCache(prepareCacheSize);
+
+                extensions.forEach(CodecRegistrar.class, registrar ->
+                    registrar.register(allocator, builder));
+
+                return MySqlConnection.init(client, builder.build(), context, queryCache.get(),
+                    prepareCache, prepare);
+            });
     }
 
     private static final class LazyQueryCache {

src/main/java/io/asyncer/r2dbc/mysql/MySqlConnectionFactoryProvider.java

@@ -23,6 +23,7 @@
 import io.r2dbc.spi.ConnectionFactoryOptions;
 import io.r2dbc.spi.ConnectionFactoryProvider;
 import io.r2dbc.spi.Option;
+import org.reactivestreams.Publisher;
 
 import javax.net.ssl.HostnameVerifier;
 import java.time.Duration;
@@ -203,6 +204,14 @@ public final class MySqlConnectionFactoryProvider implements ConnectionFactoryPr
      */
     public static final Option<Boolean> AUTODETECT_EXTENSIONS = Option.valueOf("autodetectExtensions");
 
+    /**
+     * Password Supplier function can be used to retrieve password before creating a connection.
+     * This can be used with Amazon RDS Aurora IAM authentication, wherein it requires token to be generated.
+     * The token is valid for 15 minutes, and this token will be used as password.
+     *
+     */
+    public static final Option<Publisher<String>> PASSWORD_SUPPLIER = Option.valueOf("passwordSupplier");
+
     @Override
     public ConnectionFactory create(ConnectionFactoryOptions options) {
         requireNonNull(options, "connectionFactoryOptions must not be null");
@@ -261,6 +270,8 @@ static MySqlConnectionConfiguration setup(ConnectionFactoryOptions options) {
             .to(builder::socketTimeout);
         mapper.optional(DATABASE).asString()
             .to(builder::database);
+        mapper.optional(PASSWORD_SUPPLIER).as(Publisher.class)
+            .to(builder::passwordSupplier);
 
         return builder.build();
     }

src/test/java/io/asyncer/r2dbc/mysql/MySqlConnectionConfigurationTest.java

@@ -25,13 +25,16 @@
 import org.assertj.core.api.ThrowableTypeAssert;
 import org.jetbrains.annotations.Nullable;
 import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 import java.time.Duration;
 import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
@@ -189,6 +192,21 @@ void nonAutodetectExtensions() {
         assertThat(list).isEmpty();
     }
 
+    @Test
+    void validPasswordSupplier() {
+        final Mono<String> passwordSupplier = Mono.just("123456");
+        Mono.from(MySqlConnectionConfiguration.builder()
+                .host(HOST)
+                .user(USER)
+                .passwordSupplier(passwordSupplier)
+                .autodetectExtensions(false)
+                .build()
+                .getPasswordSupplier())
+            .as(StepVerifier::create)
+            .expectNext("123456")
+            .verifyComplete();
+    }
+
     private static MySqlConnectionConfiguration unixSocketSslMode(SslMode sslMode) {
         return MySqlConnectionConfiguration.builder()
             .unixSocket(UNIX_SOCKET)

src/test/java/io/asyncer/r2dbc/mysql/MySqlConnectionFactoryProviderTest.java

@@ -24,6 +24,8 @@
 import io.r2dbc.spi.Option;
 import org.assertj.core.api.Assert;
 import org.junit.jupiter.api.Test;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
 
 import javax.net.ssl.HostnameVerifier;
 import javax.net.ssl.SSLSession;
@@ -34,7 +36,9 @@
 import java.util.Collections;
 import java.util.function.Function;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 
+import static io.asyncer.r2dbc.mysql.MySqlConnectionFactoryProvider.PASSWORD_SUPPLIER;
 import static io.asyncer.r2dbc.mysql.MySqlConnectionFactoryProvider.USE_SERVER_PREPARE_STATEMENT;
 import static io.r2dbc.spi.ConnectionFactoryOptions.CONNECT_TIMEOUT;
 import static io.r2dbc.spi.ConnectionFactoryOptions.DATABASE;
@@ -390,6 +394,20 @@ void invalidServerPreparing() {
                 .option(USE_SERVER_PREPARE_STATEMENT, NotPredicate.class.getPackage() + "NonePredicate")
                 .build()));
     }
+
+    @Test
+    void validPasswordSupplier() {
+        final Publisher<String> passwordSupplier = Mono.just("123456");
+        ConnectionFactoryOptions options = ConnectionFactoryOptions.builder()
+            .option(DRIVER, "mysql")
+            .option(HOST, "127.0.0.1")
+            .option(USER, "root")
+            .option(PASSWORD_SUPPLIER, passwordSupplier)
+            .build();
+
+        assertThat(ConnectionFactories.get(options)).isExactlyInstanceOf(MySqlConnectionFactory.class);
+    }
+
 }
 
 final class MockException extends RuntimeException {