Add aws-secretsmanager-r2dbc

[eddumelendez]

Similar to aws-secretsmanager-jdbc, a new project can be created to support secretsmanager for r2dbc.

[IRuzhnikov]

Hi! what are you think about this?

I have two alternative ways for this:

1. this way works for biggest web services. just add dependencies and db connection health check for restarting service by vm host:

<dependency>
    <!-- docs of config https://docs.awspring.io/spring-cloud-aws/docs/current/reference/html/index.html#integrating-your-spring-cloud-application-with-the-aws-secrets-manager -->
    <!-- list of config https://docs.awspring.io/spring-cloud-aws/docs/current/reference/html/appendix.html -->
    <groupId>io.awspring.cloud</groupId>
    <artifactId>spring-cloud-starter-aws-secrets-manager-config</artifactId>
</dependency>
<dependency>
    <!-- docs of config https://docs.awspring.io/spring-cloud-aws/docs/current/reference/html/index.html#integrating-your-spring-cloud-application-with-the-aws-parameter-store -->
    <!-- list of config https://docs.awspring.io/spring-cloud-aws/docs/current/reference/html/appendix.html -->
    <groupId>io.awspring.cloud</groupId>
    <artifactId>spring-cloud-starter-aws-parameter-store-config</artifactId>
</dependency>

2. Disclaimer: this way seems biggest, just because i can't access to initialization code in spring library
   I made draft version for remake connection when connection broken:
   add same deps that in first way

import io.r2dbc.pool.ConnectionPool;
import io.r2dbc.pool.ConnectionPoolConfiguration;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.ConnectionFactory;
import io.r2dbc.spi.ConnectionFactoryMetadata;
import lombok.Getter;
import lombok.SneakyThrows;
import org.reactivestreams.Publisher;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.boot.autoconfigure.r2dbc.ConnectionFactoryOptionsBuilderCustomizer;
import org.springframework.boot.autoconfigure.r2dbc.R2dbcAutoConfiguration;
import org.springframework.boot.autoconfigure.r2dbc.R2dbcProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.cglib.core.ReflectUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Import;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.lang.reflect.Method;

import static lombok.AccessLevel.PROTECTED;

@Component
@Getter(PROTECTED)
@Import(R2dbcAutoConfiguration.class)
@SuppressWarnings("JavadocReference")
public class AwsR2dbcConnectionFactoryBeanPostProcessor implements BeanPostProcessor, ApplicationContextAware {
    private ApplicationContext ctx;

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        if (bean instanceof ConnectionPool) {
            return wrapPool((ConnectionPool) bean);
        } else if (bean instanceof ConnectionFactory) {
            return wrapFactory((ConnectionFactory) bean);
        }
        return bean;
    }

    /**
     * Logic based on {@link org.springframework.boot.autoconfigure.r2dbc.ConnectionFactoryConfigurations.PoolConfiguration}
     */
    protected ConnectionPool wrapPool(ConnectionPool bean) {
        var connectionFactory = wrapFactory(bean.unwrap());
        var pool = getCtx().getBean(R2dbcProperties.class).getPool();
        PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
        ConnectionPoolConfiguration.Builder builder = ConnectionPoolConfiguration.builder(connectionFactory);
        map.from(pool.getMaxIdleTime()).to(builder::maxIdleTime);
        map.from(pool.getMaxLifeTime()).to(builder::maxLifeTime);
        map.from(pool.getMaxAcquireTime()).to(builder::maxAcquireTime);
        map.from(pool.getMaxCreateConnectionTime()).to(builder::maxCreateConnectionTime);
        map.from(pool.getInitialSize()).to(builder::initialSize);
        map.from(pool.getMaxSize()).to(builder::maxSize);
        map.from(pool.getValidationQuery()).whenHasText().to(builder::validationQuery);
        map.from(pool.getValidationDepth()).to(builder::validationDepth);
        return new ConnectionPool(builder.build());
    }

    protected ConnectionFactory wrapFactory(ConnectionFactory bean) {
        return new AwsSecretManagerConnectionFactory(bean, getCtx());
    }

    @Override
    public void setApplicationContext(ApplicationContext ctx) throws BeansException {
        this.ctx = ctx;
    }

    @Getter(PROTECTED)
    public static class AwsSecretManagerConnectionFactory implements ConnectionFactory {
        //TODO make this class accessible????
        protected static final String BASE_CONNECTION_FACTORY =
                "org.springframework.boot.autoconfigure.r2dbc.ConnectionFactoryConfigurations" +
                        "$GenericConfiguration.connectionFactory(" +
                        "org.springframework.boot.autoconfigure.r2dbc.R2dbcProperties," +
                        "org.springframework.core.io.ResourceLoader," +
                        "org.springframework.beans.factory.ObjectProvider" +
                        ")";
        //TODO make sure threadsafe
        private ConnectionFactory current;
        private final ApplicationContext ctx;
        private final Method factoryMethod;
        private final Object factoryOwnerInstance;

        public AwsSecretManagerConnectionFactory(ConnectionFactory current, ApplicationContext ctx) {
            this.current = current;
            this.ctx = ctx;

            // ATTENTION! for library experience extract this creator to self customizer bean!
            factoryMethod = ReflectUtils.findMethod(BASE_CONNECTION_FACTORY, ctx.getClassLoader());
            factoryMethod.trySetAccessible();
            factoryOwnerInstance = ReflectUtils.newInstance(factoryMethod.getDeclaringClass());
        }

        @Override
        public Publisher<? extends Connection> create() {
            var publisher = getCurrent().create();
            if (publisher instanceof Mono) {
                return ((Mono<? extends Connection>) publisher).onErrorResume(this::createFallback);
            } else if (publisher instanceof Flux) {
                return ((Flux<? extends Connection>) publisher).onErrorResume(this::createFallback);
            }
            return publisher;
        }

        private <P extends Publisher<? extends Connection>> P createFallback(Throwable throwable) {
            //TODO do this only for Auth exceptions
            current = recreate();
            //TODO make it retryable
            return (P) create();
        }

        @Override
        public ConnectionFactoryMetadata getMetadata() {
            return getCurrent().getMetadata();
        }

        public ConnectionFactory unwrap() {
            return getCurrent();
        }

        @SneakyThrows
        protected ConnectionFactory recreate() {
            var properties = getCtx().getBean(R2dbcProperties.class);
            var customizers = getCtx().getBeanProvider(ConnectionFactoryOptionsBuilderCustomizer.class);
            return (ConnectionFactory) getFactoryMethod().invoke(getFactoryOwnerInstance(), properties, getCtx(), customizers);
        }

        @Override
        public String toString() {
            return getClass().getSimpleName() + "[" + getCurrent().toString() + "]";
        }
    }
}