#2 - Add DatabaseClient.

Provide DatabaseClient and add ExternalDatabase class to provide database connection details/encapsulate a dockerized testcontainer.

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.
+ * <p>
+ * 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.
+ * <p>
+ * <b>Note:</b> 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<Row, RowMetadata, Map<String, Object>> {
+
+	public final static ColumnMapRowMapper INSTANCE = new ColumnMapRowMapper();
+
+	@Override
+	public Map<String, Object> apply(Row row, RowMetadata rowMetadata) {
+
+		Collection<? extends ColumnMetadata> columns = IterableUtils.toCollection(rowMetadata.getColumnMetadatas());
+		int columnCount = columns.size();
+		Map<String, Object> 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.
+	 * <p>
+	 * 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<String, Object> 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.
+	 * <p>
+	 * 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);
+	}
+}

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.
+ * <p>
+ * 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<Builder> 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<String> sqlSupplier);
+	}
+
+	/**
+	 * Contract for specifying a SQL call along with options leading to the exchange.
+	 */
+	interface GenericExecuteSpec extends BindSpec<GenericExecuteSpec> {
+
+		/**
+		 * Define the target type the result should be mapped to. <br />
+		 * Skip this step if you are anyway fine with the default conversion.
+		 *
+		 * @param resultType must not be {@literal null}.
+		 * @param <R> result type.
+		 */
+		<R> TypedExecuteSpec<R> as(Class<R> resultType);
+
+		/**
+		 * Perform the SQL call and retrieve the result.
+		 */
+		FetchSpec<Map<String, Object>> fetch();
+
+		/**
+		 * Perform the SQL request and return a {@link SqlResult}.
+		 *
+		 * @return a {@code Mono} for the result
+		 */
+		Mono<SqlResult<Map<String, Object>>> exchange();
+	}
+
+	/**
+	 * Contract for specifying a SQL call along with options leading to the exchange.
+	 */
+	interface TypedExecuteSpec<T> extends BindSpec<TypedExecuteSpec<T>> {
+
+		/**
+		 * Define the target type the result should be mapped to. <br />
+		 * Skip this step if you are anyway fine with the default conversion.
+		 *
+		 * @param resultType must not be {@literal null}.
+		 * @param <R> result type.
+		 */
+		<R> TypedExecuteSpec<R> as(Class<R> resultType);
+
+		/**
+		 * Perform the SQL call and retrieve the result.
+		 */
+		FetchSpec<T> fetch();
+
+		/**
+		 * Perform the SQL request and return a {@link SqlResult}.
+		 *
+		 * @return a {@code Mono} for the result
+		 */
+		Mono<SqlResult<T>> exchange();
+	}
+
+	/**
+	 * Contract for specifying {@code SELECT} options leading to the exchange.
+	 */
+	interface SelectFromSpec {
+
+		GenericSelectSpec from(String table);
+
+		<T> TypedSelectSpec<T> from(Class<T> 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);
+
+		<T> TypedInsertSpec<T> into(Class<T> table);
+	}
+
+	/**
+	 * Contract for specifying {@code SELECT} options leading to the exchange.
+	 */
+	interface GenericSelectSpec extends SelectSpec<GenericSelectSpec> {
+
+		/**
+		 * Define the target type the result should be mapped to. <br />
+		 * Skip this step if you are anyway fine with the default conversion.
+		 *
+		 * @param resultType must not be {@literal null}.
+		 * @param <R> result type.
+		 */
+		<R> TypedSelectSpec<R> as(Class<R> resultType);
+
+		/**
+		 * Perform the SQL call and retrieve the result.
+		 */
+		FetchSpec<Map<String, Object>> fetch();
+
+		/**
+		 * Perform the SQL request and return a {@link SqlResult}.
+		 *
+		 * @return a {@code Mono} for the result
+		 */
+		Mono<SqlResult<Map<String, Object>>> exchange();
+	}
+
+	/**
+	 * Contract for specifying {@code SELECT} options leading to the exchange.
+	 */
+	interface TypedSelectSpec<T> extends SelectSpec<TypedSelectSpec<T>> {
+
+		/**
+		 * Define the target type the result should be mapped to. <br />
+		 * Skip this step if you are anyway fine with the default conversion.
+		 *
+		 * @param resultType must not be {@literal null}.
+		 * @param <R> result type.
+		 */
+		<R> TypedSelectSpec<R> as(Class<R> resultType);
+
+		/**
+		 * Configure a result mapping {@link java.util.function.Function}.
+		 *
+		 * @param mappingFunction must not be {@literal null}.
+		 * @param <R> result type.
+		 * @return
+		 */
+		<R> TypedSelectSpec<R> extract(BiFunction<Row, RowMetadata, R> mappingFunction);
+
+		/**
+		 * Perform the SQL call and retrieve the result.
+		 */
+		FetchSpec<T> fetch();
+
+		/**
+		 * Perform the SQL request and return a {@link SqlResult}.
+		 *
+		 * @return a {@code Mono} for the result
+		 */
+		Mono<SqlResult<T>> exchange();
+	}
+
+	/**
+	 * Contract for specifying {@code SELECT} options leading to the exchange.
+	 */
+	interface SelectSpec<S extends SelectSpec<S>> {
+
+		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<T> {
+
+		/**
+		 * 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<T> table(String tableName);
+
+		/**
+		 * Insert the given {@link Publisher} to insert one or more objects.
+		 *
+		 * @param objectToInsert
+		 * @return
+		 */
+		InsertSpec using(Publisher<T> objectToInsert);
+	}
+
+	/**
+	 * Contract for specifying {@code INSERT} options leading to the exchange.
+	 */
+	interface InsertSpec {
+
+		/**
+		 * Perform the SQL call.
+		 */
+		Mono<Void> then();
+
+		/**
+		 * Perform the SQL request and return a {@link SqlResult}.
+		 *
+		 * @return a {@code Mono} for the result
+		 */
+		Mono<SqlResult<Map<String, Object>>> exchange();
+	}
+
+	/**
+	 * Contract for specifying parameter bindings.
+	 */
+	interface BindSpec<S extends BindSpec<S>> {
+
+		/**
+		 * 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<T> {
+
+		/**
+		 * 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<T> one();
+
+		/**
+		 * Get the first or no result.
+		 *
+		 * @return {@link Mono#empty()} if no match found. Never {@literal null}.
+		 */
+		Mono<T> first();
+
+		/**
+		 * Get all matching elements.
+		 *
+		 * @return never {@literal null}.
+		 */
+		Flux<T> all();
+
+		/**
+		 * Get the number of updated rows.
+		 *
+		 * @return {@link Mono} emitting the number of updated rows. Never {@literal null}.
+		 */
+		Mono<Integer> rowsUpdated();
+	}
+
+}

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 <T> Flux<T> execute(Function<Connection, Flux<T>> action) throws DataAccessException {
+
+		Assert.notNull(action, "Callback object must not be null");
+
+		Mono<Connection> 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<String, Optional<Object>> byName,
+			Map<Integer, Optional<Object>> 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<String> 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<Integer, Optional<Object>> byIndex;
+		final Map<String, Optional<Object>> byName;
+		final Supplier<String> sqlSupplier;
+
+		GenericExecuteSpecSupport(Supplier<String> 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;
+		}
+
+		<T> Mono<SqlResult<T>> exchange(String sql, BiFunction<Row, RowMetadata, T> mappingFunction) {
+
+			return execute(it -> {
+
+				Statement statement = it.createStatement(sql);
+				doBind(statement, byName, byIndex);
+
+				return Flux
+						.just((SqlResult<T>) new DefaultSqlResult<>(sql, Flux.from(statement.add().execute()), mappingFunction));
+			}).next();
+		}
+
+		public GenericExecuteSpecSupport bind(int index, Object value) {
+
+			Map<Integer, Optional<Object>> byIndex = new LinkedHashMap<>(this.byIndex);
+			byIndex.put(index, Optional.of(value));
+
+			return createInstance(byIndex, this.byName, this.sqlSupplier);
+		}
+
+		public GenericExecuteSpecSupport bindNull(int index) {
+
+			Map<Integer, Optional<Object>> 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<String, Optional<Object>> 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<String, Optional<Object>> byName = new LinkedHashMap<>(this.byName);
+			byName.put(name, Optional.empty());
+
+			return createInstance(this.byIndex, byName, this.sqlSupplier);
+		}
+
+		protected GenericExecuteSpecSupport createInstance(Map<Integer, Optional<Object>> byIndex,
+				Map<String, Optional<Object>> byName, Supplier<String> 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<Integer, Optional<Object>> byIndex, Map<String, Optional<Object>> byName,
+				Supplier<String> sqlSupplier) {
+			super(byIndex, byName, sqlSupplier);
+		}
+
+		DefaultGenericExecuteSpec(Supplier<String> sqlSupplier) {
+			super(sqlSupplier);
+		}
+
+		@Override
+		public <R> TypedExecuteSpec<R> as(Class<R> resultType) {
+
+			Assert.notNull(resultType, "Result type must not be null!");
+
+			return new DefaultTypedGenericExecuteSpec<>(this.byIndex, this.byName, this.sqlSupplier, resultType);
+		}
+
+		@Override
+		public FetchSpec<Map<String, Object>> 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<SqlResult<Map<String, Object>>> 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<Integer, Optional<Object>> byIndex,
+				Map<String, Optional<Object>> byName, Supplier<String> sqlSupplier) {
+			return new DefaultGenericExecuteSpec(byIndex, byName, sqlSupplier);
+		}
+	}
+
+	/**
+	 * Default {@link org.springframework.data.jdbc.core.function.DatabaseClient.GenericExecuteSpec} implementation.
+	 */
+	@SuppressWarnings("unchecked")
+	private class DefaultTypedGenericExecuteSpec<T> extends GenericExecuteSpecSupport implements TypedExecuteSpec<T> {
+
+		private final Class<T> typeToRead;
+		private final BiFunction<Row, RowMetadata, T> mappingFunction;
+
+		DefaultTypedGenericExecuteSpec(Map<Integer, Optional<Object>> byIndex, Map<String, Optional<Object>> byName,
+				Supplier<String> sqlSupplier, Class<T> typeToRead) {
+
+			super(byIndex, byName, sqlSupplier);
+
+			this.typeToRead = typeToRead;
+			this.mappingFunction = dataAccessStrategy.getRowMapper(typeToRead);
+		}
+
+		@Override
+		public <R> TypedExecuteSpec<R> as(Class<R> resultType) {
+
+			Assert.notNull(resultType, "Result type must not be null!");
+
+			return new DefaultTypedGenericExecuteSpec<>(this.byIndex, this.byName, this.sqlSupplier, resultType);
+		}
+
+		@Override
+		public FetchSpec<T> fetch() {
+			String sql = getSql();
+			return new DefaultFetchSpec<>(sql, exchange(sql, mappingFunction).flatMapMany(SqlResult::all),
+					exchange(sql, mappingFunction).flatMap(FetchSpec::rowsUpdated));
+		}
+
+		@Override
+		public Mono<SqlResult<T>> exchange() {
+			return exchange(getSql(), mappingFunction);
+		}
+
+		@Override
+		public DefaultTypedGenericExecuteSpec<T> bind(int index, Object value) {
+			return (DefaultTypedGenericExecuteSpec<T>) super.bind(index, value);
+		}
+
+		@Override
+		public DefaultTypedGenericExecuteSpec<T> bindNull(int index) {
+			return (DefaultTypedGenericExecuteSpec<T>) super.bindNull(index);
+		}
+
+		@Override
+		public DefaultTypedGenericExecuteSpec<T> bind(String name, Object value) {
+			return (DefaultTypedGenericExecuteSpec) super.bind(name, value);
+		}
+
+		@Override
+		public DefaultTypedGenericExecuteSpec<T> bindNull(String name) {
+			return (DefaultTypedGenericExecuteSpec<T>) super.bindNull(name);
+		}
+
+		@Override
+		public DefaultTypedGenericExecuteSpec<T> bind(Object bean) {
+			return (DefaultTypedGenericExecuteSpec<T>) super.bind(bean);
+		}
+
+		@Override
+		protected DefaultTypedGenericExecuteSpec<T> createInstance(Map<Integer, Optional<Object>> byIndex,
+				Map<String, Optional<Object>> byName, Supplier<String> 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 <T> TypedInsertSpec<T> into(Class<T> 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<String, Optional<Object>> byName;
+
+		@Override
+		public GenericInsertSpec value(String field, Object value) {
+
+			Assert.notNull(field, "Field must not be null!");
+
+			Map<String, Optional<Object>> 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<String, Optional<Object>> byName = new LinkedHashMap<>(this.byName);
+			byName.put(field, Optional.empty());
+
+			return new DefaultGenericInsertSpec(this.table, byName);
+		}
+
+		@Override
+		public Mono<Void> then() {
+			return exchange().flatMapMany(FetchSpec::all).then();
+		}
+
+		@Override
+		public Mono<SqlResult<Map<String, Object>>> 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<Object> o : byName.values()) {
+
+					if (o.isPresent()) {
+						o.ifPresent(v -> statement.bind(index.getAndIncrement(), v));
+					} else {
+						statement.bindNull("$" + (index.getAndIncrement() + 1), 0); // TODO: What is type?
+					}
+				}
+
+				SqlResult<Map<String, Object>> result = new DefaultSqlResult<>(sql,
+						Flux.from(statement.executeReturningGeneratedKeys()), ColumnMapRowMapper.INSTANCE);
+				return Flux.just(result);
+
+			}).next();
+		}
+	}
+
+	/**
+	 * Default implementation of {@link org.springframework.data.jdbc.core.function.DatabaseClient.TypedInsertSpec}.
+	 */
+	@RequiredArgsConstructor
+	class DefaultTypedInsertSpec<T> implements TypedInsertSpec<T>, InsertSpec {
+
+		private final Class<?> typeToInsert;
+		private final String table;
+		private final Publisher<T> objectToInsert;
+
+		public DefaultTypedInsertSpec(Class<?> typeToInsert) {
+
+			this.typeToInsert = typeToInsert;
+			this.table = dataAccessStrategy.getTableName(typeToInsert);
+			this.objectToInsert = Mono.empty();
+		}
+
+		@Override
+		public TypedInsertSpec<T> table(String tableName) {
+
+			Assert.hasText(tableName, "Table name must not be null or empty!");
+
+			return new DefaultTypedInsertSpec<>(typeToInsert, tableName, objectToInsert);
+		}
+
+		@Override
+		public InsertSpec using(T objectToInsert) {
+
+			Assert.notNull(objectToInsert, "Object to insert must not be null!");
+
+			return new DefaultTypedInsertSpec<>(typeToInsert, table, Mono.just(objectToInsert));
+		}
+
+		@Override
+		public InsertSpec using(Publisher<T> objectToInsert) {
+
+			Assert.notNull(objectToInsert, "Publisher to insert must not be null!");
+
+			return new DefaultTypedInsertSpec<>(typeToInsert, table, objectToInsert);
+		}
+
+		@Override
+		public Mono<Void> then() {
+			return exchange().flatMapMany(FetchSpec::all).then();
+		}
+
+		@Override
+		public Mono<SqlResult<Map<String, Object>>> exchange() {
+
+			return Mono.from(objectToInsert).flatMap(toInsert -> {
+
+				StringBuilder builder = new StringBuilder();
+
+				List<Pair<String, Object>> insertValues = dataAccessStrategy.getInsert(toInsert);
+				String fieldNames = insertValues.stream().map(Pair::getFirst).collect(Collectors.joining(","));
+				String placeholders = IntStream.range(0, insertValues.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 (Pair<String, Object> pair : insertValues) {
+
+						if (pair.getSecond() != null) { // TODO: Better type to transport null values.
+							statement.bind(index.getAndIncrement(), pair.getSecond());
+						} else {
+							statement.bindNull("$" + (index.getAndIncrement() + 1), 0); // TODO: What is type?
+						}
+					}
+
+					SqlResult<Map<String, Object>> result = new DefaultSqlResult<>(sql,
+							Flux.from(statement.executeReturningGeneratedKeys()), ColumnMapRowMapper.INSTANCE);
+					return Flux.just(result);
+
+				}).next();
+			});
+		}
+	}
+
+	/**
+	 * Default {@link org.springframework.data.jdbc.core.function.DatabaseClient.SqlResult} implementation.
+	 */
+	static class DefaultSqlResult<T> implements SqlResult<T> {
+
+		private final String sql;
+		private final Flux<Result> result;
+		private final FetchSpec<T> fetchSpec;
+
+		DefaultSqlResult(String sql, Flux<Result> result, BiFunction<Row, RowMetadata, T> mappingFunction) {
+
+			this.sql = sql;
+			this.result = result;
+			this.fetchSpec = new DefaultFetchSpec<>(sql, result.flatMap(it -> it.map(mappingFunction)),
+					result.flatMap(Result::getRowsUpdated).next());
+		}
+
+		@Override
+		public <R> SqlResult<R> extract(BiFunction<Row, RowMetadata, R> mappingFunction) {
+			return new DefaultSqlResult<>(sql, result, mappingFunction);
+		}
+
+		@Override
+		public Mono<T> one() {
+			return fetchSpec.one();
+		}
+
+		@Override
+		public Mono<T> first() {
+			return fetchSpec.first();
+		}
+
+		@Override
+		public Flux<T> all() {
+			return fetchSpec.all();
+		}
+
+		@Override
+		public Mono<Integer> rowsUpdated() {
+			return fetchSpec.rowsUpdated();
+		}
+	}
+
+	@RequiredArgsConstructor
+	static class DefaultFetchSpec<T> implements FetchSpec<T> {
+
+		private final String sql;
+		private final Flux<T> result;
+		private final Mono<Integer> updatedRows;
+
+		@Override
+		public Mono<T> one() {
+
+			return all().buffer(2) //
+					.flatMap(it -> {
+
+						if (it.isEmpty()) {
+							return Mono.empty();
+						}
+
+						if (it.size() > 1) {
+							return Mono.error(new IncorrectResultSizeDataAccessException(
+									String.format("Query [%s] returned non unique result.", this.sql), 1));
+						}
+
+						return Mono.just(it.get(0));
+					}).next();
+		}
+
+		@Override
+		public Mono<T> first() {
+			return all().next();
+		}
+
+		@Override
+		public Flux<T> all() {
+			return result;
+		}
+
+		@Override
+		public Mono<Integer> rowsUpdated() {
+			return updatedRows;
+		}
+	}
+
+	private static <T> Flux<T> doInConnection(Function<Connection, Flux<T>> action, Connection it) {
+
+		try {
+			return action.apply(it);
+		} catch (RuntimeException e) {
+
+			String sql = getSql(action);
+			return Flux.error(new DefaultDatabaseClient.UncategorizedSQLException("ConnectionCallback", sql, e) {});
+		}
+	}
+
+	/**
+	 * Determine SQL from potential provider object.
+	 *
+	 * @param sqlProvider object that's potentially a SqlProvider
+	 * @return the SQL string, or {@code null}
+	 * @see SqlProvider
+	 */
+	@Nullable
+	private static String getSql(Object sqlProvider) {
+
+		if (sqlProvider instanceof SqlProvider) {
+			return ((SqlProvider) sqlProvider).getSql();
+		} else {
+			return null;
+		}
+	}
+
+	/**
+	 * Invocation handler that suppresses close calls on R2DBC Connections. Also prepares returned Statement
+	 * (Prepared/CallbackStatement) objects.
+	 *
+	 * @see Connection#close()
+	 */
+	private class CloseSuppressingInvocationHandler implements InvocationHandler {
+
+		private final Connection target;
+
+		CloseSuppressingInvocationHandler(Connection target) {
+			this.target = target;
+		}
+
+		@Override
+		@Nullable
+		public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
+			// Invocation on ConnectionProxy interface coming in...
+
+			if (method.getName().equals("equals")) {
+				// Only consider equal when proxies are identical.
+				return (proxy == args[0]);
+			} else if (method.getName().equals("hashCode")) {
+				// Use hashCode of PersistenceManager proxy.
+				return System.identityHashCode(proxy);
+			} else if (method.getName().equals("unwrap")) {
+				if (((Class<?>) args[0]).isInstance(proxy)) {
+					return proxy;
+				}
+			} else if (method.getName().equals("isWrapperFor")) {
+				if (((Class<?>) args[0]).isInstance(proxy)) {
+					return true;
+				}
+			} else if (method.getName().equals("close")) {
+				// Handle close method: suppress, not valid.
+				return Mono.error(new UnsupportedOperationException("Close is not supported!"));
+			} else if (method.getName().equals("getTargetConnection")) {
+				// Handle getTargetConnection method: return underlying Connection.
+				return this.target;
+			}
+
+			// Invoke method on target Connection.
+			try {
+				Object retVal = method.invoke(this.target, args);
+
+				return retVal;
+			} catch (InvocationTargetException ex) {
+				throw ex.getTargetException();
+			}
+		}
+	}
+
+	private static class UncategorizedSQLException extends UncategorizedDataAccessException {
+
+		/** SQL that led to the problem */
+		@Nullable private final String sql;
+
+		/**
+		 * Constructor for UncategorizedSQLException.
+		 *
+		 * @param task name of current task
+		 * @param sql the offending SQL statement
+		 * @param ex the root cause
+		 */
+		public UncategorizedSQLException(String task, @Nullable String sql, Exception ex) {
+			super(String.format("%s; uncategorized SQLException%s; %s", task, sql != null ? " for SQL [" + sql + "]" : "",
+					ex.getMessage()), ex);
+			this.sql = sql;
+		}
+
+		/**
+		 * Return the SQL that led to the problem (if known).
+		 */
+		@Nullable
+		public String getSql() {
+			return this.sql;
+		}
+	}
+}

src/main/java/org/springframework/data/jdbc/core/function/DefaultDatabaseClientBuilder.java

@@ -0,0 +1,96 @@
+/*
+ * 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 java.util.function.Consumer;
+
+import org.springframework.data.jdbc.core.function.DatabaseClient.Builder;
+import org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator;
+import org.springframework.jdbc.support.SQLExceptionTranslator;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+import io.r2dbc.spi.ConnectionFactory;
+
+/**
+ * Default implementation of {@link DatabaseClient.Builder}.
+ *
+ * @author Mark Paluch
+ */
+class DefaultDatabaseClientBuilder implements DatabaseClient.Builder {
+
+	private @Nullable ConnectionFactory connector;
+	private SQLExceptionTranslator exceptionTranslator = new SQLErrorCodeSQLExceptionTranslator();
+	private ReactiveDataAccessStrategy accessStrategy = new DefaultReactiveDataAccessStrategy();
+
+	DefaultDatabaseClientBuilder() {}
+
+	DefaultDatabaseClientBuilder(DefaultDatabaseClientBuilder other) {
+
+		Assert.notNull(other, "DefaultDatabaseClientBuilder must not be null!");
+
+		this.connector = other.connector;
+		this.exceptionTranslator = exceptionTranslator;
+	}
+
+	@Override
+	public Builder connectionFactory(ConnectionFactory factory) {
+
+		Assert.notNull(factory, "ConnectionFactory must not be null!");
+
+		this.connector = factory;
+		return this;
+	}
+
+	@Override
+	public Builder exceptionTranslator(SQLExceptionTranslator exceptionTranslator) {
+
+		Assert.notNull(exceptionTranslator, "SQLExceptionTranslator must not be null!");
+
+		this.exceptionTranslator = exceptionTranslator;
+		return this;
+	}
+
+	@Override
+	public Builder dataAccessStrategy(ReactiveDataAccessStrategy accessStrategy) {
+
+		Assert.notNull(accessStrategy, "ReactiveDataAccessStrategy must not be null!");
+
+		this.accessStrategy = accessStrategy;
+		return this;
+	}
+
+	@Override
+	public DatabaseClient build() {
+
+		return new DefaultDatabaseClient(this.connector, exceptionTranslator, accessStrategy,
+				new DefaultDatabaseClientBuilder(this));
+	}
+
+	@Override
+	public DatabaseClient.Builder clone() {
+		return new DefaultDatabaseClientBuilder(this);
+	}
+
+	@Override
+	public DatabaseClient.Builder apply(Consumer<DatabaseClient.Builder> builderConsumer) {
+		Assert.notNull(builderConsumer, "BuilderConsumer must not be null");
+
+		builderConsumer.accept(this);
+		return this;
+	}
+}

src/main/java/org/springframework/data/jdbc/core/function/DefaultReactiveDataAccessStrategy.java

@@ -0,0 +1,75 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+import java.util.function.BiFunction;
+
+import org.springframework.data.convert.EntityInstantiators;
+import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentEntity;
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentProperty;
+import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.util.Pair;
+import org.springframework.util.ClassUtils;
+
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+
+/**
+ * @author Mark Paluch
+ */
+public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStrategy {
+
+	private final EntityInstantiators instantiators = new EntityInstantiators();
+	private final JdbcMappingContext mappingContext = new JdbcMappingContext();
+
+	@Override
+	public List<Pair<String, Object>> getInsert(Object object) {
+
+		Class<?> userClass = ClassUtils.getUserClass(object);
+
+		JdbcPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(userClass);
+		PersistentPropertyAccessor propertyAccessor = entity.getPropertyAccessor(object);
+
+		List<Pair<String, Object>> values = new ArrayList<>();
+
+		for (JdbcPersistentProperty property : entity) {
+
+			Object value = propertyAccessor.getProperty(property);
+
+			if (value == null) {
+				continue;
+			}
+
+			values.add(Pair.of(property.getColumnName(), value));
+		}
+
+		return values;
+	}
+
+	@Override
+	public <T> BiFunction<Row, RowMetadata, T> getRowMapper(Class<T> typeToRead) {
+		return new EntityRowMapper<T>((JdbcPersistentEntity) mappingContext.getRequiredPersistentEntity(typeToRead),
+				instantiators, mappingContext);
+	}
+
+	@Override
+	public String getTableName(Class<?> type) {
+		return mappingContext.getRequiredPersistentEntity(type).getTableName();
+	}
+}

src/main/java/org/springframework/data/jdbc/core/function/EntityRowMapper.java

@@ -0,0 +1,168 @@
+/*
+ * 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 lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+
+import java.sql.ResultSet;
+import java.util.function.BiFunction;
+
+import org.springframework.core.convert.ConversionService;
+import org.springframework.data.convert.EntityInstantiators;
+import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentEntity;
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentProperty;
+import org.springframework.data.mapping.MappingException;
+import org.springframework.data.mapping.PersistentProperty;
+import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.mapping.PreferredConstructor.Parameter;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
+import org.springframework.data.mapping.model.ParameterValueProvider;
+import org.springframework.util.ClassUtils;
+
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+
+/**
+ * Maps a {@link io.r2dbc.spi.Row} to an entity of type {@code T}, including entities referenced.
+ *
+ * @author Mark Paluch
+ * @since 1.0
+ */
+public class EntityRowMapper<T> implements BiFunction<Row, RowMetadata, T> {
+
+	private final JdbcPersistentEntity<T> entity;
+	private final EntityInstantiators entityInstantiators;
+	private final ConversionService conversions;
+	private final MappingContext<JdbcPersistentEntity<?>, JdbcPersistentProperty> context;
+
+	public EntityRowMapper(JdbcPersistentEntity<T> entity, EntityInstantiators entityInstantiators,
+			JdbcMappingContext context) {
+
+		this.entity = entity;
+		this.entityInstantiators = entityInstantiators;
+		this.conversions = context.getConversions();
+		this.context = context;
+	}
+
+	@Override
+	public T apply(Row row, RowMetadata metadata) {
+
+		T result = createInstance(row, "", entity);
+
+		ConvertingPropertyAccessor propertyAccessor = new ConvertingPropertyAccessor(entity.getPropertyAccessor(result),
+				conversions);
+
+		for (JdbcPersistentProperty property : entity) {
+
+			if (property.isCollectionLike()) {
+				throw new UnsupportedOperationException();
+			} else if (property.isMap()) {
+
+				throw new UnsupportedOperationException();
+			} else {
+				propertyAccessor.setProperty(property, readFrom(row, property, ""));
+			}
+		}
+
+		return result;
+	}
+
+	/**
+	 * Read a single value or a complete Entity from the {@link ResultSet} passed as an argument.
+	 *
+	 * @param row the {@link Row} to extract the value from. Must not be {@code null}.
+	 * @param property the {@link JdbcPersistentProperty} for which the value is intended. Must not be {@code null}.
+	 * @param prefix to be used for all column names accessed by this method. Must not be {@code null}.
+	 * @return the value read from the {@link ResultSet}. May be {@code null}.
+	 */
+	private Object readFrom(Row row, JdbcPersistentProperty property, String prefix) {
+
+		try {
+
+			if (property.isEntity()) {
+				return readEntityFrom(row, property);
+			}
+
+			return row.get(prefix + property.getColumnName(), getType(property));
+
+		} catch (Exception o_O) {
+			throw new MappingException(String.format("Could not read property %s from result set!", property), o_O);
+		}
+	}
+
+	private Class<?> getType(JdbcPersistentProperty property) {
+		return ClassUtils.resolvePrimitiveIfNecessary(property.getActualType());
+	}
+
+	private <S> S readEntityFrom(Row row, PersistentProperty<?> property) {
+
+		String prefix = property.getName() + "_";
+
+		@SuppressWarnings("unchecked")
+		JdbcPersistentEntity<S> entity = (JdbcPersistentEntity<S>) context
+				.getRequiredPersistentEntity(property.getActualType());
+
+		if (readFrom(row, entity.getRequiredIdProperty(), prefix) == null) {
+			return null;
+		}
+
+		S instance = createInstance(row, prefix, entity);
+
+		PersistentPropertyAccessor accessor = entity.getPropertyAccessor(instance);
+		ConvertingPropertyAccessor propertyAccessor = new ConvertingPropertyAccessor(accessor, conversions);
+
+		for (JdbcPersistentProperty p : entity) {
+			propertyAccessor.setProperty(p, readFrom(row, p, prefix));
+		}
+
+		return instance;
+	}
+
+	private <S> S createInstance(Row row, String prefix, JdbcPersistentEntity<S> entity) {
+
+		return entityInstantiators.getInstantiatorFor(entity).createInstance(entity,
+				new RowParameterValueProvider(row, entity, conversions, prefix));
+	}
+
+	@RequiredArgsConstructor
+	private static class RowParameterValueProvider implements ParameterValueProvider<JdbcPersistentProperty> {
+
+		@NonNull private final Row resultSet;
+		@NonNull private final JdbcPersistentEntity<?> entity;
+		@NonNull private final ConversionService conversionService;
+		@NonNull private final String prefix;
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mapping.model.ParameterValueProvider#getParameterValue(org.springframework.data.mapping.PreferredConstructor.Parameter)
+		 */
+		@Override
+		public <T> T getParameterValue(Parameter<T, JdbcPersistentProperty> parameter) {
+
+			String column = prefix + entity.getRequiredPersistentProperty(parameter.getName()).getColumnName();
+
+			try {
+				return conversionService.convert(resultSet.get(column, parameter.getType().getType()),
+						parameter.getType().getType());
+			} catch (Exception o_O) {
+				throw new MappingException(String.format("Couldn't read column %s from Row.", column), o_O);
+			}
+		}
+	}
+}

src/main/java/org/springframework/data/jdbc/core/function/IterableUtils.java

@@ -0,0 +1,62 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.springframework.util.Assert;
+
+/**
+ * @author Mark Paluch
+ */
+class IterableUtils {
+
+	static <T> Collection<T> toCollection(Iterable<T> iterable) {
+
+		Assert.notNull(iterable, "Iterable must not be null!");
+
+		if (iterable instanceof Collection) {
+			return (Collection<T>) iterable;
+		}
+
+		List<T> result = new ArrayList<>();
+
+		for (T element : iterable) {
+			result.add(element);
+		}
+
+		return result;
+	}
+
+	static <T> List<T> toList(Iterable<T> iterable) {
+
+		Assert.notNull(iterable, "Iterable must not be null!");
+
+		if (iterable instanceof List) {
+			return (List<T>) iterable;
+		}
+
+		List<T> result = new ArrayList<>();
+
+		for (T element : iterable) {
+			result.add(element);
+		}
+
+		return result;
+	}
+}

src/main/java/org/springframework/data/jdbc/core/function/ReactiveDataAccessStrategy.java

@@ -0,0 +1,37 @@
+/*
+ * 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 java.util.List;
+import java.util.function.BiFunction;
+
+import org.springframework.data.util.Pair;
+
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+
+/**
+ * @author Mark Paluch
+ */
+public interface ReactiveDataAccessStrategy {
+
+	List<Pair<String, Object>> getInsert(Object object);
+
+	// TODO: Broaden T to Mono<T>/Flux<T> for reactive relational data access?
+	<T> BiFunction<Row, RowMetadata, T> getRowMapper(Class<T> typeToRead);
+
+	String getTableName(Class<?> type);
+}

src/main/java/org/springframework/data/jdbc/core/function/SqlResult.java

@@ -0,0 +1,31 @@
+/*
+ * 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 java.util.function.BiFunction;
+
+import org.springframework.data.jdbc.core.function.DatabaseClient.FetchSpec;
+
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+
+/**
+ * @author Mark Paluch
+ */
+public interface SqlResult<T> extends FetchSpec<T> {
+
+	<R> SqlResult<R> extract(BiFunction<Row, RowMetadata, R> mappingFunction);
+}

src/main/java/org/springframework/data/jdbc/core/function/connectionfactory/ConnectionProxy.java

@@ -0,0 +1,41 @@
+/*
+ * 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.connectionfactory;
+
+import io.r2dbc.spi.Connection;
+
+import java.sql.Wrapper;
+
+/**
+ * Subinterface of {@link Connection} to be implemented by Connection proxies. Allows access to the underlying target
+ * Connection.
+ * <p>
+ * This interface can be checked when there is a need to cast to a native R2DBC {@link Connection}. Alternatively, all
+ * such connections also support JDBC 4.0's {@link Connection#unwrap}.
+ *
+ * @author Mark Paluch
+ */
+public interface ConnectionProxy extends Connection, Wrapper {
+
+	/**
+	 * Return the target Connection of this proxy.
+	 * <p>
+	 * This will typically be the native driver Connection or a wrapper from a connection pool.
+	 *
+	 * @return the underlying Connection (never {@code null})
+	 */
+	Connection getTargetConnection();
+}