Add R2DBC support

modules/postgresql/build.gradle

@@ -2,8 +2,12 @@ description = "Testcontainers :: JDBC :: PostgreSQL"
 
 dependencies {
     compile project(':jdbc')
+    compileOnly project(':r2dbc')
 
     testCompile 'org.postgresql:postgresql:42.2.10'
     testCompile 'commons-dbutils:commons-dbutils:1.7'
     testCompile 'com.zaxxer:HikariCP-java6:2.3.13'
+
+    testCompile project(':r2dbc')
+    testCompile 'io.r2dbc:r2dbc-postgresql:0.8.1.RELEASE'
 }

modules/postgresql/src/main/java/org/testcontainers/containers/PostgreSQLR2DBCDatabaseContainer.java

@@ -0,0 +1,33 @@
+package org.testcontainers.containers;
+
+import io.r2dbc.spi.ConnectionFactoryOptions;
+import lombok.RequiredArgsConstructor;
+import lombok.experimental.Delegate;
+import org.testcontainers.lifecycle.Startable;
+import org.testcontainers.r2dbc.R2DBCDatabaseContainer;
+
+@RequiredArgsConstructor
+public final class PostgreSQLR2DBCDatabaseContainer implements R2DBCDatabaseContainer {
+
+    @Delegate(types = Startable.class)
+    private final PostgreSQLContainer<?> container;
+
+    public static ConnectionFactoryOptions getOptions(PostgreSQLContainer<?> container) {
+        ConnectionFactoryOptions options = ConnectionFactoryOptions.builder()
+            .option(ConnectionFactoryOptions.DRIVER, PostgreSQLR2DBCDatabaseContainerProvider.DRIVER)
+            .build();
+
+        return new PostgreSQLR2DBCDatabaseContainer(container).configure(options);
+    }
+
+    @Override
+    public ConnectionFactoryOptions configure(ConnectionFactoryOptions options) {
+        return options.mutate()
+            .option(ConnectionFactoryOptions.HOST, container.getContainerIpAddress())
+            .option(ConnectionFactoryOptions.PORT, container.getMappedPort(PostgreSQLContainer.POSTGRESQL_PORT))
+            .option(ConnectionFactoryOptions.DATABASE, container.getDatabaseName())
+            .option(ConnectionFactoryOptions.USER, container.getUsername())
+            .option(ConnectionFactoryOptions.PASSWORD, container.getPassword())
+            .build();
+    }
+}

modules/postgresql/src/main/java/org/testcontainers/containers/PostgreSQLR2DBCDatabaseContainerProvider.java

@@ -0,0 +1,26 @@
+package org.testcontainers.containers;
+
+import io.r2dbc.spi.ConnectionFactoryOptions;
+import org.testcontainers.r2dbc.R2DBCDatabaseContainer;
+import org.testcontainers.r2dbc.R2DBCDatabaseContainerProvider;
+
+public final class PostgreSQLR2DBCDatabaseContainerProvider implements R2DBCDatabaseContainerProvider {
+
+    static final String DRIVER = "postgresql";
+
+    @Override
+    public boolean supports(ConnectionFactoryOptions options) {
+        return DRIVER.equals(options.getRequiredValue(ConnectionFactoryOptions.DRIVER));
+    }
+
+    @Override
+    public R2DBCDatabaseContainer createContainer(ConnectionFactoryOptions options) {
+        PostgreSQLContainer<?> container = new PostgreSQLContainer<>(options.getRequiredValue(IMAGE_OPTION))
+            .withDatabaseName(options.getRequiredValue(ConnectionFactoryOptions.DATABASE));
+
+        if (Boolean.TRUE.equals(options.getValue(REUSABLE_OPTION))) {
+            container.withReuse(true);
+        }
+        return new PostgreSQLR2DBCDatabaseContainer(container);
+    }
+}

modules/postgresql/src/main/resources/META-INF/services/org.testcontainers.r2dbc.R2DBCDatabaseContainerProvider

@@ -0,0 +1 @@
+org.testcontainers.containers.PostgreSQLR2DBCDatabaseContainerProvider

modules/postgresql/src/test/java/org/testcontainers/containers/PostgreSQLR2DBCDatabaseContainerTest.java

@@ -0,0 +1,42 @@
+package org.testcontainers.containers;
+
+import io.r2dbc.spi.Closeable;
+import io.r2dbc.spi.Connection;
+import io.r2dbc.spi.ConnectionFactories;
+import org.junit.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import static org.junit.Assert.*;
+
+public class PostgreSQLR2DBCDatabaseContainerTest {
+
+    @Test
+    public void testGetOptions() {
+        try (PostgreSQLContainer<?> container = new PostgreSQLContainer<>()) {
+            container.start();
+
+            int result = Flux
+                .usingWhen(
+                    Mono.just(
+                        ConnectionFactories.get(
+                            PostgreSQLR2DBCDatabaseContainer.getOptions(container)
+                        )
+                    ),
+                    connectionFactory -> {
+                        return Flux
+                            .usingWhen(
+                                connectionFactory.create(),
+                                connection -> connection.createStatement("SELECT 42").execute(),
+                                Connection::close
+                            )
+                            .flatMap(it -> it.map((row, meta) -> (Integer) row.get(0)));
+                    },
+                    it -> ((Closeable) it).close()
+                )
+                .blockFirst();
+
+            assertEquals(42, result);
+        }
+    }
+}

modules/postgresql/src/test/resources/logback-test.xml

@@ -0,0 +1,16 @@
+<configuration>
+
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <!-- encoders are assigned the type
+             ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} %-5level %logger - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <root level="INFO">
+        <appender-ref ref="STDOUT"/>
+    </root>
+
+    <logger name="org.testcontainers" level="DEBUG"/>
+</configuration>

modules/r2dbc/build.gradle

@@ -0,0 +1,10 @@
+description = "Testcontainers :: R2DBC"
+
+dependencies {
+    compile project(':testcontainers')
+    compile 'io.r2dbc:r2dbc-spi:0.8.1.RELEASE'
+
+    testCompile 'org.assertj:assertj-core:3.14.0'
+    testCompile 'io.r2dbc:r2dbc-postgresql:0.8.1.RELEASE'
+    testCompile project(':postgresql')
+}

modules/r2dbc/src/main/java/org/testcontainers/r2dbc/CancellableSubscription.java

@@ -0,0 +1,23 @@
+package org.testcontainers.r2dbc;
+
+import org.reactivestreams.Subscription;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+class CancellableSubscription implements Subscription {
+
+    private final AtomicBoolean cancelled = new AtomicBoolean();
+
+    @Override
+    public void request(long n) {
+    }
+
+    @Override
+    public void cancel() {
+        cancelled.set(true);
+    }
+
+    public boolean isCancelled() {
+        return cancelled.get();
+    }
+}

modules/r2dbc/src/main/java/org/testcontainers/r2dbc/ConnectionPublisher.java

@@ -0,0 +1,161 @@
+package org.testcontainers.r2dbc;
+
+import io.r2dbc.spi.Connection;
+import io.r2dbc.spi.ConnectionFactory;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
+
+/**
+ * Design notes:
+ * - ConnectionPublisher is Mono-like (0..1), the request amount is ignored
+ * - given the testing nature, the performance requirements are less strict
+ * - "synchronized" is used to avoid races
+ * - Reactive Streams spec violations are not checked (e.g. non-positive request)
+ */
+class ConnectionPublisher implements Publisher<Connection> {
+
+    private final Supplier<CompletableFuture<ConnectionFactory>> futureSupplier;
+
+    ConnectionPublisher(Supplier<CompletableFuture<ConnectionFactory>> futureSupplier) {
+        this.futureSupplier = futureSupplier;
+    }
+
+    @Override
+    public void subscribe(Subscriber<? super Connection> actual) {
+        actual.onSubscribe(new StateMachineSubscription(actual));
+    }
+
+    private class StateMachineSubscription implements Subscription {
+
+        private final Subscriber<? super Connection> actual;
+
+        Subscription subscriptionState;
+
+        StateMachineSubscription(Subscriber<? super Connection> actual) {
+            this.actual = actual;
+            subscriptionState = new WaitRequestSubscriptionState();
+        }
+
+        @Override
+        public synchronized void request(long n) {
+            subscriptionState.request(n);
+        }
+
+        @Override
+        public synchronized void cancel() {
+            subscriptionState.cancel();
+        }
+
+        synchronized void transitionTo(SubscriptionState newState) {
+            subscriptionState = newState;
+            newState.enter();
+        }
+
+        abstract class SubscriptionState implements Subscription {
+            void enter() {
+            }
+        }
+
+        class WaitRequestSubscriptionState extends SubscriptionState {
+
+            @Override
+            public void request(long n) {
+                transitionTo(new WaitFutureCompletionSubscriptionState());
+            }
+
+            @Override
+            public void cancel() {
+            }
+        }
+
+        class WaitFutureCompletionSubscriptionState extends SubscriptionState {
+
+            private CompletableFuture<ConnectionFactory> future;
+
+            @Override
+            void enter() {
+                this.future = futureSupplier.get();
+
+                future.whenComplete((connectionFactory, e) -> {
+                    if (e != null) {
+                        actual.onSubscribe(EmptySubscription.INSTANCE);
+                        actual.onError(e);
+                        return;
+                    }
+
+                    Publisher<? extends Connection> publisher = connectionFactory.create();
+                    transitionTo(new ProxySubscriptionState(publisher));
+                });
+            }
+
+            @Override
+            public void request(long n) {
+            }
+
+            @Override
+            public void cancel() {
+                future.cancel(true);
+            }
+        }
+
+        class ProxySubscriptionState extends SubscriptionState implements Subscriber<Connection> {
+
+            private final Publisher<? extends Connection> publisher;
+
+            private Subscription s;
+
+            private boolean cancelled = false;
+
+            ProxySubscriptionState(Publisher<? extends Connection> publisher) {
+                this.publisher = publisher;
+            }
+
+            @Override
+            void enter() {
+                publisher.subscribe(this);
+            }
+
+            @Override
+            public void request(long n) {
+                // Ignore
+            }
+
+            @Override
+            public synchronized void cancel() {
+                cancelled = true;
+                if (s != null) {
+                    s.cancel();
+                }
+            }
+
+            @Override
+            public synchronized void onSubscribe(Subscription s) {
+                this.s = s;
+                if (!cancelled) {
+                    s.request(1);
+                } else {
+                    s.cancel();
+                }
+            }
+
+            @Override
+            public void onNext(Connection connection) {
+                actual.onNext(connection);
+            }
+
+            @Override
+            public void onError(Throwable t) {
+                actual.onError(t);
+            }
+
+            @Override
+            public void onComplete() {
+                actual.onComplete();
+            }
+        }
+    }
+}

modules/r2dbc/src/main/java/org/testcontainers/r2dbc/EmptySubscription.java

@@ -0,0 +1,17 @@
+package org.testcontainers.r2dbc;
+
+import org.reactivestreams.Subscription;
+
+enum EmptySubscription implements Subscription {
+    INSTANCE;
+
+    @Override
+    public void request(long n) {
+
+    }
+
+    @Override
+    public void cancel() {
+
+    }
+}