diff --git a/pom.xml b/pom.xml
new file mode 100644
index 0000000..90ffd65
--- /dev/null
+++ b/pom.xml
@@ -0,0 +1,411 @@
+
+
+
+ 4.0.0
+
+ org.springframework.data
+ spring-data-jdbc
+ 1.1.0.r2dbc-SNAPSHOT
+
+ Spring Data JDBC
+ Spring Data module for JDBC repositories.
+ http://projects.spring.io/spring-data-jdbc
+
+
+ org.springframework.data.build
+ spring-data-parent
+ 2.2.0.BUILD-SNAPSHOT
+
+
+
+
+ DATAJDBC
+
+ 2.2.0.BUILD-SNAPSHOT
+ spring.data.jdbc
+ reuseReports
+
+ 3.6.2
+ 0.1.4
+ 2.2.8
+ 3.4.6
+ 1.3.2
+ 5.1.41
+ 42.0.0
+ 2.2.3
+ 1.0.0.BUILD-SNAPSHOT
+ 1.0.0.BUILD-SNAPSHOT
+ 1.7.3
+
+
+
+ 2017
+
+
+
+ schauder
+ Jens Schauder
+ jschauder(at)pivotal.io
+ Pivotal Software, Inc.
+ https://pivotal.io
+
+ Project Lead
+
+ +1
+
+
+ gregturn
+ Greg L. Turnquist
+ gturnquist(at)pivotal.io
+ Pivotal Software, Inc.
+ https://pivotal.io
+
+ Project Contributor
+
+ -6
+
+
+
+
+
+ release
+
+
+
+ org.jfrog.buildinfo
+ artifactory-maven-plugin
+ false
+
+
+
+
+
+
+ no-jacoco
+
+
+
+ org.jacoco
+ jacoco-maven-plugin
+
+
+ jacoco-initialize
+ none
+
+
+
+
+
+
+
+
+
+ all-dbs
+
+
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+
+
+ mysql-test
+ test
+
+ test
+
+
+
+ **/*IntegrationTests.java
+
+
+ **/*HsqlIntegrationTests.java
+
+
+ mysql
+
+
+
+
+ postgres-test
+ test
+
+ test
+
+
+
+ **/*IntegrationTests.java
+
+
+ **/*HsqlIntegrationTests.java
+
+
+ postgres
+
+
+
+
+ mariadb-test
+ test
+
+ test
+
+
+
+ **/*IntegrationTests.java
+
+
+ **/*HsqlIntegrationTests.java
+
+
+ mariadb
+
+
+
+
+
+
+
+
+
+
+
+
+
+ ${project.groupId}
+ spring-data-commons
+ ${springdata.commons}
+
+
+
+ org.springframework
+ spring-tx
+
+
+
+ org.springframework
+ spring-context
+
+
+
+ org.springframework
+ spring-beans
+
+
+
+ org.springframework
+ spring-jdbc
+
+
+
+ org.springframework
+ spring-core
+
+
+
+ io.r2dbc
+ r2dbc-spi
+ ${r2dbc-spi.version}
+ true
+
+
+
+ io.projectreactor
+ reactor-core
+ true
+
+
+
+ org.mybatis
+ mybatis-spring
+ ${mybatis-spring.version}
+ true
+
+
+
+ org.mybatis
+ mybatis
+ ${mybatis.version}
+ true
+
+
+
+ org.hsqldb
+ hsqldb
+ ${hsqldb.version}
+ test
+
+
+
+ org.assertj
+ assertj-core
+ ${assertj-core.version}
+ test
+
+
+
+ io.projectreactor
+ reactor-test
+ test
+
+
+
+ mysql
+ mysql-connector-java
+ ${mysql-connector-java.version}
+ test
+
+
+
+ org.postgresql
+ postgresql
+ ${postgresql.version}
+ test
+
+
+
+ org.mariadb.jdbc
+ mariadb-java-client
+ ${mariadb-java-client.version}
+ test
+
+
+
+ io.r2dbc
+ r2dbc-postgresql
+ ${r2dbc-postgresql.version}
+ test
+
+
+
+ de.schauderhaft.degraph
+ degraph-check
+ ${degraph-check.version}
+ test
+
+
+
+ org.testcontainers
+ mysql
+ ${testcontainers.version}
+ test
+
+
+ org.slf4j
+ jcl-over-slf4j
+
+
+
+
+
+ org.testcontainers
+ postgresql
+ ${testcontainers.version}
+ test
+
+
+
+ org.testcontainers
+ mariadb
+ ${testcontainers.version}
+ test
+
+
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+ 2.12
+
+
+ org.apache.maven.plugins
+ maven-dependency-plugin
+ 3.1.0
+
+
+
+
+
+
+
+ org.jacoco
+ jacoco-maven-plugin
+ ${jacoco}
+
+ ${jacoco.destfile}
+
+
+
+ jacoco-initialize
+
+ prepare-agent
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+
+
+ default-test
+
+
+ **/*Tests.java
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-assembly-plugin
+
+
+ org.codehaus.mojo
+ wagon-maven-plugin
+
+
+ org.asciidoctor
+ asciidoctor-maven-plugin
+
+
+
+
+
+
+ spring-libs-snapshot
+ https://repo.spring.io/libs-snapshot
+
+
+ nebhale-snapshots
+ https://raw.githubusercontent.com/nebhale/r2dbc/maven/snapshot
+
+ true
+
+
+
+ nebhale-milestones
+ https://raw.githubusercontent.com/nebhale/r2dbc/maven/milestone
+
+ false
+
+
+
+
+
+
+ spring-plugins-snapshot
+ https://repo.spring.io/plugins-snapshot
+
+
+
+
diff --git a/src/main/java/org/springframework/data/jdbc/core/function/ColumnMapRowMapper.java b/src/main/java/org/springframework/data/jdbc/core/function/ColumnMapRowMapper.java
new file mode 100644
index 0000000..663d88f
--- /dev/null
+++ b/src/main/java/org/springframework/data/jdbc/core/function/ColumnMapRowMapper.java
@@ -0,0 +1,102 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.core.function;
+
+import io.r2dbc.spi.ColumnMetadata;
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.function.BiFunction;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.LinkedCaseInsensitiveMap;
+
+/**
+ * {@link RowMapper} implementation that creates a {@link Map} for each row, representing all columns as key-value
+ * pairs: one entry for each column, with the column name as key.
+ *
+ * The {@link Map} implementation to use and the key to use for each column in the column Map can be customized through
+ * overriding {@link #createColumnMap} and {@link #getColumnKey}, respectively.
+ *
+ * Note: By default, {@link ColumnMapRowMapper} will try to build a linked {@link Map} with case-insensitive
+ * keys, to preserve column order as well as allow any casing to be used for column names. This requires Commons
+ * Collections on the classpath (which will be autodetected). Else, the fallback is a standard linked
+ * {@link java.util.HashMap}, which will still preserve column order but requires the application to specify the column
+ * names in the same casing as exposed by the driver.
+ *
+ * @author Mark Paluch
+ */
+public class ColumnMapRowMapper implements BiFunction> {
+
+ public final static ColumnMapRowMapper INSTANCE = new ColumnMapRowMapper();
+
+ @Override
+ public Map apply(Row row, RowMetadata rowMetadata) {
+
+ Collection columns = IterableUtils.toCollection(rowMetadata.getColumnMetadatas());
+ int columnCount = columns.size();
+ Map mapOfColValues = createColumnMap(columnCount);
+
+ int index = 0;
+ for (ColumnMetadata column : columns) {
+
+ String key = getColumnKey(column.getName());
+ Object obj = getColumnValue(row, index++);
+ mapOfColValues.put(key, obj);
+ }
+ return mapOfColValues;
+ }
+
+ /**
+ * Create a {@link Map} instance to be used as column map.
+ *
+ * By default, a linked case-insensitive Map will be created.
+ *
+ * @param columnCount the column count, to be used as initial capacity for the Map.
+ * @return the new {@link Map} instance.
+ * @see LinkedCaseInsensitiveMap
+ */
+ protected Map createColumnMap(int columnCount) {
+ return new LinkedCaseInsensitiveMap<>(columnCount);
+ }
+
+ /**
+ * Determine the key to use for the given column in the column {@link Map}.
+ *
+ * @param columnName the column name as returned by the {@link Row}.
+ * @return the column key to use.
+ * @see ColumnMetadata#getName()
+ */
+ protected String getColumnKey(String columnName) {
+ return columnName;
+ }
+
+ /**
+ * Retrieve a R2DBC object value for the specified column.
+ *
+ * The default implementation uses the {@link Row#get(Object, Class)} method.
+ *
+ * @param row is the {@link Row} holding the data.
+ * @param index is the column index.
+ * @return the Object returned.
+ */
+ @Nullable
+ protected Object getColumnValue(Row row, int index) {
+ return row.get(index, Object.class);
+ }
+}
diff --git a/src/main/java/org/springframework/data/jdbc/core/function/DatabaseClient.java b/src/main/java/org/springframework/data/jdbc/core/function/DatabaseClient.java
new file mode 100644
index 0000000..5926592
--- /dev/null
+++ b/src/main/java/org/springframework/data/jdbc/core/function/DatabaseClient.java
@@ -0,0 +1,451 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.core.function;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.Map;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import org.reactivestreams.Publisher;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
+import org.springframework.jdbc.support.SQLExceptionTranslator;
+
+import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+
+/**
+ * A non-blocking, reactive client for performing database calls requests with Reactive Streams back pressure. Provides
+ * a higher level, common API over R2DBC client libraries.
+ *
+ * Use one of the static factory methods {@link #create(ConnectionFactory)} or obtain a {@link DatabaseClient#builder()}
+ * to create an instance.
+ *
+ * @author Mark Paluch
+ */
+public interface DatabaseClient {
+
+ /**
+ * Prepare an SQL call returning a result.
+ */
+ SqlSpec execute();
+
+ /**
+ * Prepare an SQL SELECT call.
+ */
+ SelectFromSpec select();
+
+ /**
+ * Prepare an SQL INSERT call.
+ */
+ InsertIntoSpec insert();
+
+ /**
+ * Return a builder to mutate properties of this database client.
+ */
+ DatabaseClient.Builder mutate();
+
+ // Static, factory methods
+
+ /**
+ * A variant of {@link #create()} that accepts a {@link io.r2dbc.spi.ConnectionFactory}
+ */
+ static DatabaseClient create(ConnectionFactory factory) {
+ return new DefaultDatabaseClientBuilder().connectionFactory(factory).build();
+ }
+
+ /**
+ * Obtain a {@code WebClient} builder.
+ */
+ static DatabaseClient.Builder builder() {
+ return new DefaultDatabaseClientBuilder();
+ }
+
+ /**
+ * A mutable builder for creating a {@link DatabaseClient}.
+ */
+ interface Builder {
+
+ /**
+ * Configures the {@link ConnectionFactory R2DBC connector}.
+ *
+ * @param factory must not be {@literal null}.
+ * @return {@code this} {@link Builder}.
+ */
+ Builder connectionFactory(ConnectionFactory factory);
+
+ /**
+ * Configures a {@link SQLExceptionTranslator}.
+ *
+ * @param exceptionTranslator must not be {@literal null}.
+ * @return {@code this} {@link Builder}.
+ */
+ Builder exceptionTranslator(SQLExceptionTranslator exceptionTranslator);
+
+ /**
+ * Configures a {@link ReactiveDataAccessStrategy}.
+ *
+ * @param accessStrategy must not be {@literal null}.
+ * @return {@code this} {@link Builder}.
+ */
+ Builder dataAccessStrategy(ReactiveDataAccessStrategy accessStrategy);
+
+ /**
+ * Configures a {@link Consumer} to configure this builder.
+ *
+ * @param builderConsumer must not be {@literal null}.
+ * @return {@code this} {@link Builder}.
+ */
+ Builder apply(Consumer builderConsumer);
+
+ /**
+ * Builder the {@link DatabaseClient} instance.
+ */
+ DatabaseClient build();
+ }
+
+ /**
+ * Contract for specifying a SQL call along with options leading to the exchange.
+ */
+ interface SqlSpec {
+
+ /**
+ * Specify a static {@code sql} string to execute.
+ *
+ * @param sql must not be {@literal null} or empty.
+ * @return a new {@link GenericExecuteSpec}.
+ */
+ GenericExecuteSpec sql(String sql);
+
+ /**
+ * Specify a static {@link Supplier SQL supplier} that provides SQL to execute.
+ *
+ * @param sqlSupplier must not be {@literal null}.
+ * @return a new {@link GenericExecuteSpec}.
+ */
+ GenericExecuteSpec sql(Supplier sqlSupplier);
+ }
+
+ /**
+ * Contract for specifying a SQL call along with options leading to the exchange.
+ */
+ interface GenericExecuteSpec extends BindSpec {
+
+ /**
+ * Define the target type the result should be mapped to.
+ * Skip this step if you are anyway fine with the default conversion.
+ *
+ * @param resultType must not be {@literal null}.
+ * @param result type.
+ */
+ TypedExecuteSpec as(Class resultType);
+
+ /**
+ * Perform the SQL call and retrieve the result.
+ */
+ FetchSpec> fetch();
+
+ /**
+ * Perform the SQL request and return a {@link SqlResult}.
+ *
+ * @return a {@code Mono} for the result
+ */
+ Mono>> exchange();
+ }
+
+ /**
+ * Contract for specifying a SQL call along with options leading to the exchange.
+ */
+ interface TypedExecuteSpec extends BindSpec> {
+
+ /**
+ * Define the target type the result should be mapped to.
+ * Skip this step if you are anyway fine with the default conversion.
+ *
+ * @param resultType must not be {@literal null}.
+ * @param result type.
+ */
+ TypedExecuteSpec as(Class resultType);
+
+ /**
+ * Perform the SQL call and retrieve the result.
+ */
+ FetchSpec fetch();
+
+ /**
+ * Perform the SQL request and return a {@link SqlResult}.
+ *
+ * @return a {@code Mono} for the result
+ */
+ Mono> exchange();
+ }
+
+ /**
+ * Contract for specifying {@code SELECT} options leading to the exchange.
+ */
+ interface SelectFromSpec {
+
+ GenericSelectSpec from(String table);
+
+ TypedSelectSpec from(Class table);
+ }
+
+ /**
+ * Contract for specifying {@code SELECT} options leading to the exchange.
+ */
+ interface InsertIntoSpec {
+
+ /**
+ * Specify the target {@literal table} to insert into.
+ *
+ * @param table must not be {@literal null} or empty.
+ * @return
+ */
+ GenericInsertSpec into(String table);
+
+ TypedInsertSpec into(Class table);
+ }
+
+ /**
+ * Contract for specifying {@code SELECT} options leading to the exchange.
+ */
+ interface GenericSelectSpec extends SelectSpec {
+
+ /**
+ * Define the target type the result should be mapped to.
+ * Skip this step if you are anyway fine with the default conversion.
+ *
+ * @param resultType must not be {@literal null}.
+ * @param result type.
+ */
+ TypedSelectSpec as(Class resultType);
+
+ /**
+ * Perform the SQL call and retrieve the result.
+ */
+ FetchSpec> fetch();
+
+ /**
+ * Perform the SQL request and return a {@link SqlResult}.
+ *
+ * @return a {@code Mono} for the result
+ */
+ Mono>> exchange();
+ }
+
+ /**
+ * Contract for specifying {@code SELECT} options leading to the exchange.
+ */
+ interface TypedSelectSpec extends SelectSpec> {
+
+ /**
+ * Define the target type the result should be mapped to.
+ * Skip this step if you are anyway fine with the default conversion.
+ *
+ * @param resultType must not be {@literal null}.
+ * @param result type.
+ */
+ TypedSelectSpec as(Class resultType);
+
+ /**
+ * Configure a result mapping {@link java.util.function.Function}.
+ *
+ * @param mappingFunction must not be {@literal null}.
+ * @param result type.
+ * @return
+ */
+ TypedSelectSpec extract(BiFunction mappingFunction);
+
+ /**
+ * Perform the SQL call and retrieve the result.
+ */
+ FetchSpec fetch();
+
+ /**
+ * Perform the SQL request and return a {@link SqlResult}.
+ *
+ * @return a {@code Mono} for the result
+ */
+ Mono> exchange();
+ }
+
+ /**
+ * Contract for specifying {@code SELECT} options leading to the exchange.
+ */
+ interface SelectSpec> {
+
+ S project(String... selectedFields);
+
+ S where(Object criteriaDefinition);
+
+ S orderBy(Sort sort);
+
+ S page(Pageable page);
+ }
+
+ /**
+ * Contract for specifying {@code INSERT} options leading to the exchange.
+ */
+ interface GenericInsertSpec extends InsertSpec {
+
+ /**
+ * Specify a field and non-{@literal null} value to insert.
+ *
+ * @param field must not be {@literal null} or empty.
+ * @param value must not be {@literal null}
+ */
+ GenericInsertSpec value(String field, Object value);
+
+ /**
+ * Specify a {@literal null} value to insert.
+ *
+ * @param field must not be {@literal null} or empty.
+ */
+ GenericInsertSpec nullValue(String field);
+ }
+
+ /**
+ * Contract for specifying {@code SELECT} options leading the exchange.
+ */
+ interface TypedInsertSpec {
+
+ /**
+ * Insert the given {@code objectToInsert}.
+ *
+ * @param objectToInsert
+ * @return
+ */
+ InsertSpec using(T objectToInsert);
+
+ /**
+ * Use the given {@code tableName} as insert target.
+ *
+ * @param tableName must not be {@literal null} or empty.
+ * @return
+ */
+ TypedInsertSpec table(String tableName);
+
+ /**
+ * Insert the given {@link Publisher} to insert one or more objects.
+ *
+ * @param objectToInsert
+ * @return
+ */
+ InsertSpec using(Publisher objectToInsert);
+ }
+
+ /**
+ * Contract for specifying {@code INSERT} options leading to the exchange.
+ */
+ interface InsertSpec {
+
+ /**
+ * Perform the SQL call.
+ */
+ Mono then();
+
+ /**
+ * Perform the SQL request and return a {@link SqlResult}.
+ *
+ * @return a {@code Mono} for the result
+ */
+ Mono>> exchange();
+ }
+
+ /**
+ * Contract for specifying parameter bindings.
+ */
+ interface BindSpec> {
+
+ /**
+ * Bind a non-{@literal null} value to a parameter identified by its {@code index}.
+ *
+ * @param index
+ * @param value must not be {@literal null}.
+ */
+ S bind(int index, Object value);
+
+ /**
+ * Bind a {@literal null} value to a parameter identified by its {@code index}.
+ *
+ * @param index
+ */
+ S bindNull(int index);
+
+ /**
+ * Bind a non-{@literal null} value to a parameter identified by its {@code name}.
+ *
+ * @param name must not be {@literal null} or empty.
+ * @param value must not be {@literal null}.
+ */
+ S bind(String name, Object value);
+
+ /**
+ * Bind a {@literal null} value to a parameter identified by its {@code name}.
+ *
+ * @param name must not be {@literal null} or empty.
+ */
+ S bindNull(String name);
+
+ /**
+ * Bind a bean according to Java {@link java.beans.BeanInfo Beans} using property names.
+ *
+ * @param bean must not be {@literal null}.
+ */
+ S bind(Object bean);
+ }
+
+ /**
+ * Contract for fetching results.
+ */
+ interface FetchSpec {
+
+ /**
+ * Get exactly zero or one result.
+ *
+ * @return {@link Mono#empty()} if no match found. Never {@literal null}.
+ * @throws org.springframework.dao.IncorrectResultSizeDataAccessException if more than one match found.
+ */
+ Mono one();
+
+ /**
+ * Get the first or no result.
+ *
+ * @return {@link Mono#empty()} if no match found. Never {@literal null}.
+ */
+ Mono first();
+
+ /**
+ * Get all matching elements.
+ *
+ * @return never {@literal null}.
+ */
+ Flux all();
+
+ /**
+ * Get the number of updated rows.
+ *
+ * @return {@link Mono} emitting the number of updated rows. Never {@literal null}.
+ */
+ Mono rowsUpdated();
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/jdbc/core/function/DefaultDatabaseClient.java b/src/main/java/org/springframework/data/jdbc/core/function/DefaultDatabaseClient.java
new file mode 100644
index 0000000..de640ca
--- /dev/null
+++ b/src/main/java/org/springframework/data/jdbc/core/function/DefaultDatabaseClient.java
@@ -0,0 +1,793 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.core.function;
+
+import io.r2dbc.spi.Connection;
+import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.Result;
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+import io.r2dbc.spi.Statement;
+import lombok.RequiredArgsConstructor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
+import org.springframework.dao.DataAccessException;
+import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.dao.UncategorizedDataAccessException;
+import org.springframework.data.jdbc.core.function.connectionfactory.ConnectionProxy;
+import org.springframework.data.util.Pair;
+import org.springframework.jdbc.core.SqlProvider;
+import org.springframework.jdbc.support.SQLExceptionTranslator;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Default implementation of {@link DatabaseClient}.
+ *
+ * @author Mark Paluch
+ */
+class DefaultDatabaseClient implements DatabaseClient {
+
+ /** Logger available to subclasses */
+ protected final Log logger = LogFactory.getLog(getClass());
+
+ private final ConnectionFactory connector;
+
+ private final SQLExceptionTranslator exceptionTranslator;
+
+ private final ReactiveDataAccessStrategy dataAccessStrategy;
+
+ private final DefaultDatabaseClientBuilder builder;
+
+ public DefaultDatabaseClient(ConnectionFactory connector, SQLExceptionTranslator exceptionTranslator,
+ ReactiveDataAccessStrategy dataAccessStrategy, DefaultDatabaseClientBuilder builder) {
+
+ this.connector = connector;
+ this.exceptionTranslator = exceptionTranslator;
+ this.dataAccessStrategy = dataAccessStrategy;
+ this.builder = builder;
+ }
+
+ @Override
+ public Builder mutate() {
+ return builder;
+ }
+
+ @Override
+ public SqlSpec execute() {
+ return new DefaultSqlSpec();
+ }
+
+ @Override
+ public SelectFromSpec select() {
+ throw new UnsupportedOperationException("Implement me");
+ }
+
+ @Override
+ public InsertIntoSpec insert() {
+ return new DefaultInsertIntoSpec();
+ }
+
+ public Flux execute(Function> action) throws DataAccessException {
+
+ Assert.notNull(action, "Callback object must not be null");
+
+ Mono connectionMono = Mono.from(obtainConnectionFactory().create());
+ // Create close-suppressing Connection proxy, also preparing returned Statements.
+
+ return connectionMono.flatMapMany(connection -> {
+
+ Connection connectionToUse = createConnectionProxy(connection);
+
+ // TODO: Release connection
+ return doInConnection(action, connectionToUse);
+
+ }).onErrorMap(SQLException.class, ex -> {
+
+ String sql = getSql(action);
+ return translateException("ConnectionCallback", sql, ex);
+ });
+ }
+
+ /**
+ * Obtain the {@link ConnectionFactory} for actual use.
+ *
+ * @return the ConnectionFactory (never {@code null})
+ * @throws IllegalStateException in case of no DataSource set
+ */
+ protected ConnectionFactory obtainConnectionFactory() {
+ return connector;
+ }
+
+ /**
+ * Create a close-suppressing proxy for the given R2DBC Connection. Called by the {@code execute} method.
+ *
+ * @param con the R2DBC Connection to create a proxy for
+ * @return the Connection proxy
+ */
+ protected Connection createConnectionProxy(Connection con) {
+ return (Connection) Proxy.newProxyInstance(ConnectionProxy.class.getClassLoader(),
+ new Class[] { ConnectionProxy.class }, new CloseSuppressingInvocationHandler(con));
+ }
+
+ /**
+ * Translate the given {@link SQLException} into a generic {@link DataAccessException}.
+ *
+ * @param task readable text describing the task being attempted
+ * @param sql SQL query or update that caused the problem (may be {@code null})
+ * @param ex the offending {@code SQLException}
+ * @return a DataAccessException wrapping the {@code SQLException} (never {@code null})
+ */
+ protected DataAccessException translateException(String task, @Nullable String sql, SQLException ex) {
+ DataAccessException dae = exceptionTranslator.translate(task, sql, ex);
+ return (dae != null ? dae : new UncategorizedSQLException(task, sql, ex));
+ }
+
+ static void doBind(Statement statement, Map> byName,
+ Map> byIndex) {
+
+ byIndex.forEach((i, o) -> {
+
+ if (o.isPresent()) {
+ o.ifPresent(v -> statement.bind(i, v));
+ } else {
+ statement.bindNull(i, 0); // TODO: What is type?
+ }
+ });
+
+ byName.forEach((name, o) -> {
+
+ if (o.isPresent()) {
+ o.ifPresent(v -> statement.bind(name, v));
+ } else {
+ statement.bindNull(name, 0); // TODO: What is type?
+ }
+ });
+
+ }
+
+ /**
+ * Default {@link org.springframework.data.jdbc.core.function.DatabaseClient.SqlSpec} implementation.
+ */
+ private class DefaultSqlSpec implements SqlSpec {
+
+ @Override
+ public GenericExecuteSpec sql(String sql) {
+
+ Assert.hasText(sql, "SQL must not be null or empty!");
+ return sql(() -> sql);
+ }
+
+ @Override
+ public GenericExecuteSpec sql(Supplier sqlSupplier) {
+
+ Assert.notNull(sqlSupplier, "SQL Supplier must not be null!");
+
+ return new DefaultGenericExecuteSpec(sqlSupplier);
+ }
+ }
+
+ /**
+ * Default {@link org.springframework.data.jdbc.core.function.DatabaseClient.GenericExecuteSpec} implementation.
+ */
+ @RequiredArgsConstructor
+ private class GenericExecuteSpecSupport {
+
+ final Map> byIndex;
+ final Map> byName;
+ final Supplier sqlSupplier;
+
+ GenericExecuteSpecSupport(Supplier sqlSupplier) {
+
+ this.byIndex = Collections.emptyMap();
+ this.byName = Collections.emptyMap();
+ this.sqlSupplier = sqlSupplier;
+ }
+
+ protected String getSql() {
+
+ String sql = sqlSupplier.get();
+ Assert.state(sql != null, "SQL supplier returned null!");
+ return sql;
+ }
+
+ Mono> exchange(String sql, BiFunction mappingFunction) {
+
+ return execute(it -> {
+
+ Statement statement = it.createStatement(sql);
+ doBind(statement, byName, byIndex);
+
+ return Flux
+ .just((SqlResult) new DefaultSqlResult<>(sql, Flux.from(statement.add().execute()), mappingFunction));
+ }).next();
+ }
+
+ public GenericExecuteSpecSupport bind(int index, Object value) {
+
+ Map> byIndex = new LinkedHashMap<>(this.byIndex);
+ byIndex.put(index, Optional.of(value));
+
+ return createInstance(byIndex, this.byName, this.sqlSupplier);
+ }
+
+ public GenericExecuteSpecSupport bindNull(int index) {
+
+ Map> byIndex = new LinkedHashMap<>(this.byIndex);
+ byIndex.put(index, Optional.empty());
+
+ return createInstance(byIndex, this.byName, this.sqlSupplier);
+ }
+
+ public GenericExecuteSpecSupport bind(String name, Object value) {
+
+ Assert.hasText(name, "Parameter name must not be null or empty!");
+
+ Map> byName = new LinkedHashMap<>(this.byName);
+ byName.put(name, Optional.of(value));
+
+ return createInstance(this.byIndex, byName, this.sqlSupplier);
+ }
+
+ public GenericExecuteSpecSupport bindNull(String name) {
+
+ Assert.hasText(name, "Parameter name must not be null or empty!");
+
+ Map> byName = new LinkedHashMap<>(this.byName);
+ byName.put(name, Optional.empty());
+
+ return createInstance(this.byIndex, byName, this.sqlSupplier);
+ }
+
+ protected GenericExecuteSpecSupport createInstance(Map> byIndex,
+ Map> byName, Supplier sqlSupplier) {
+ return new GenericExecuteSpecSupport(byIndex, byName, sqlSupplier);
+ }
+
+ public GenericExecuteSpecSupport bind(Object bean) {
+
+ Assert.notNull(bean, "Bean must not be null!");
+
+ throw new UnsupportedOperationException("Implement me!");
+ }
+ }
+
+ /**
+ * Default {@link org.springframework.data.jdbc.core.function.DatabaseClient.GenericExecuteSpec} implementation.
+ */
+ private class DefaultGenericExecuteSpec extends GenericExecuteSpecSupport implements GenericExecuteSpec {
+
+ DefaultGenericExecuteSpec(Map> byIndex, Map> byName,
+ Supplier sqlSupplier) {
+ super(byIndex, byName, sqlSupplier);
+ }
+
+ DefaultGenericExecuteSpec(Supplier sqlSupplier) {
+ super(sqlSupplier);
+ }
+
+ @Override
+ public TypedExecuteSpec as(Class resultType) {
+
+ Assert.notNull(resultType, "Result type must not be null!");
+
+ return new DefaultTypedGenericExecuteSpec<>(this.byIndex, this.byName, this.sqlSupplier, resultType);
+ }
+
+ @Override
+ public FetchSpec> fetch() {
+
+ String sql = getSql();
+ return new DefaultFetchSpec<>(sql, exchange(sql, ColumnMapRowMapper.INSTANCE).flatMapMany(SqlResult::all),
+ exchange(sql, ColumnMapRowMapper.INSTANCE).flatMap(FetchSpec::rowsUpdated));
+ }
+
+ @Override
+ public Mono>> exchange() {
+ return exchange(getSql(), ColumnMapRowMapper.INSTANCE);
+ }
+
+ @Override
+ public DefaultGenericExecuteSpec bind(int index, Object value) {
+ return (DefaultGenericExecuteSpec) super.bind(index, value);
+ }
+
+ @Override
+ public DefaultGenericExecuteSpec bindNull(int index) {
+ return (DefaultGenericExecuteSpec) super.bindNull(index);
+ }
+
+ @Override
+ public DefaultGenericExecuteSpec bind(String name, Object value) {
+ return (DefaultGenericExecuteSpec) super.bind(name, value);
+ }
+
+ @Override
+ public DefaultGenericExecuteSpec bindNull(String name) {
+ return (DefaultGenericExecuteSpec) super.bindNull(name);
+ }
+
+ @Override
+ public DefaultGenericExecuteSpec bind(Object bean) {
+ return (DefaultGenericExecuteSpec) super.bind(bean);
+ }
+
+ @Override
+ protected GenericExecuteSpecSupport createInstance(Map> byIndex,
+ Map> byName, Supplier sqlSupplier) {
+ return new DefaultGenericExecuteSpec(byIndex, byName, sqlSupplier);
+ }
+ }
+
+ /**
+ * Default {@link org.springframework.data.jdbc.core.function.DatabaseClient.GenericExecuteSpec} implementation.
+ */
+ @SuppressWarnings("unchecked")
+ private class DefaultTypedGenericExecuteSpec extends GenericExecuteSpecSupport implements TypedExecuteSpec {
+
+ private final Class typeToRead;
+ private final BiFunction mappingFunction;
+
+ DefaultTypedGenericExecuteSpec(Map> byIndex, Map> byName,
+ Supplier sqlSupplier, Class typeToRead) {
+
+ super(byIndex, byName, sqlSupplier);
+
+ this.typeToRead = typeToRead;
+ this.mappingFunction = dataAccessStrategy.getRowMapper(typeToRead);
+ }
+
+ @Override
+ public TypedExecuteSpec as(Class resultType) {
+
+ Assert.notNull(resultType, "Result type must not be null!");
+
+ return new DefaultTypedGenericExecuteSpec<>(this.byIndex, this.byName, this.sqlSupplier, resultType);
+ }
+
+ @Override
+ public FetchSpec fetch() {
+ String sql = getSql();
+ return new DefaultFetchSpec<>(sql, exchange(sql, mappingFunction).flatMapMany(SqlResult::all),
+ exchange(sql, mappingFunction).flatMap(FetchSpec::rowsUpdated));
+ }
+
+ @Override
+ public Mono> exchange() {
+ return exchange(getSql(), mappingFunction);
+ }
+
+ @Override
+ public DefaultTypedGenericExecuteSpec bind(int index, Object value) {
+ return (DefaultTypedGenericExecuteSpec) super.bind(index, value);
+ }
+
+ @Override
+ public DefaultTypedGenericExecuteSpec bindNull(int index) {
+ return (DefaultTypedGenericExecuteSpec) super.bindNull(index);
+ }
+
+ @Override
+ public DefaultTypedGenericExecuteSpec bind(String name, Object value) {
+ return (DefaultTypedGenericExecuteSpec) super.bind(name, value);
+ }
+
+ @Override
+ public DefaultTypedGenericExecuteSpec bindNull(String name) {
+ return (DefaultTypedGenericExecuteSpec) super.bindNull(name);
+ }
+
+ @Override
+ public DefaultTypedGenericExecuteSpec bind(Object bean) {
+ return (DefaultTypedGenericExecuteSpec) super.bind(bean);
+ }
+
+ @Override
+ protected DefaultTypedGenericExecuteSpec createInstance(Map> byIndex,
+ Map> byName, Supplier sqlSupplier) {
+ return new DefaultTypedGenericExecuteSpec<>(byIndex, byName, sqlSupplier, typeToRead);
+ }
+ }
+
+ /**
+ * Default {@link org.springframework.data.jdbc.core.function.DatabaseClient.InsertIntoSpec} implementation.
+ */
+ class DefaultInsertIntoSpec implements InsertIntoSpec {
+
+ @Override
+ public GenericInsertSpec into(String table) {
+ return new DefaultGenericInsertSpec(table, Collections.emptyMap());
+ }
+
+ @Override
+ public TypedInsertSpec into(Class table) {
+ return new DefaultTypedInsertSpec<>(table);
+ }
+ }
+
+ /**
+ * Default implementation of {@link org.springframework.data.jdbc.core.function.DatabaseClient.GenericInsertSpec}.
+ */
+ @RequiredArgsConstructor
+ class DefaultGenericInsertSpec implements GenericInsertSpec {
+
+ private final String table;
+ private final Map> byName;
+
+ @Override
+ public GenericInsertSpec value(String field, Object value) {
+
+ Assert.notNull(field, "Field must not be null!");
+
+ Map> byName = new LinkedHashMap<>(this.byName);
+ byName.put(field, Optional.of(value));
+
+ return new DefaultGenericInsertSpec(this.table, byName);
+ }
+
+ @Override
+ public GenericInsertSpec nullValue(String field) {
+
+ Assert.notNull(field, "Field must not be null!");
+
+ Map> byName = new LinkedHashMap<>(this.byName);
+ byName.put(field, Optional.empty());
+
+ return new DefaultGenericInsertSpec(this.table, byName);
+ }
+
+ @Override
+ public Mono then() {
+ return exchange().flatMapMany(FetchSpec::all).then();
+ }
+
+ @Override
+ public Mono>> exchange() {
+
+ if (byName.isEmpty()) {
+ throw new IllegalStateException("Insert fields is empty!");
+ }
+
+ StringBuilder builder = new StringBuilder();
+ String fieldNames = byName.keySet().stream().collect(Collectors.joining(","));
+ String placeholders = IntStream.range(0, byName.size()).mapToObj(i -> "$" + (i + 1))
+ .collect(Collectors.joining(","));
+
+ builder.append("INSERT INTO ").append(table).append(" (").append(fieldNames).append(") ").append(" VALUES(")
+ .append(placeholders).append(")");
+
+ return execute(it -> {
+
+ String sql = builder.toString();
+ Statement statement = it.createStatement(sql);
+
+ AtomicInteger index = new AtomicInteger();
+ for (Optional