#2 - Create relational and r2dbc packages.

Move types into relational and r2dbc packages in preparation for a later module separation.
2018-06-21 08:44:03 +02:00
parent cee7479cd0
commit 8b006abbe5
43 changed files with 212 additions and 158 deletions
--- a/src/main/java/org/springframework/data/r2dbc/function/ConnectionAccessor.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/ConnectionAccessor.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.r2dbc.function;
+
+import io.r2dbc.spi.Connection;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Function;
+
+import org.springframework.dao.DataAccessException;
+
+/**
+ * Interface declaring methods that accept callback {@link Function} to operate within the scope of a
+ * {@link Connection}. Callback functions operate on a provided connection and must not close the connection as the
+ * connections may be pooled or be subject to other kinds of resource management.
+ * <p/>
+ * Callback functions are responsible for creating a {@link org.reactivestreams.Publisher} that defines the scope of how
+ * long the allocated {@link Connection} is valid. Connections are released after the publisher terminates.
+ *
+ * @author Mark Paluch
+ */
+public interface ConnectionAccessor {
+
+	/**
+	 * Execute a callback {@link Function} within a {@link Connection} scope. The function is responsible for creating a
+	 * {@link Mono}. The connection is released after the {@link Mono} terminates (or the subscription is cancelled).
+	 * Connection resources must not be passed outside of the {@link Function} closure, otherwise resources may get
+	 * defunct.
+	 *
+	 * @param action must not be {@literal null}.
+	 * @return the resulting {@link Mono}.
+	 * @throws DataAccessException
+	 */
+	<T> Mono<T> inConnection(Function<Connection, Mono<T>> action) throws DataAccessException;
+
+	/**
+	 * Execute a callback {@link Function} within a {@link Connection} scope. The function is responsible for creating a
+	 * {@link Flux}. The connection is released after the {@link Flux} terminates (or the subscription is cancelled).
+	 * Connection resources must not be passed outside of the {@link Function} closure, otherwise resources may get
+	 * defunct.
+	 *
+	 * @param action must not be {@literal null}.
+	 * @return the resulting {@link Flux}.
+	 * @throws DataAccessException
+	 */
+	<T> Flux<T> inConnectionMany(Function<Connection, Flux<T>> action) throws DataAccessException;
+
+}
--- a/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java
@@ -0,0 +1,445 @@
+/*
+ * 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.r2dbc.function;
+
+import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+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;
+
+/**
+ * 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 DatabaseClient} 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 {
+
+		/**
+		 * Specify the source {@literal table} to select from.
+		 *
+		 * @param table must not be {@literal null} or empty.
+		 * @return
+		 */
+		GenericSelectSpec from(String table);
+
+		/**
+		 * Specify the source table to select from to using the {@link Class entity class}.
+		 *
+		 * @param table must not be {@literal null}.
+		 * @return
+		 */
+		<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);
+
+		/**
+		 * Specify the target table to insert to using the {@link Class entity class}.
+		 *
+		 * @param table must not be {@literal null}.
+		 * @return
+		 */
+		<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>> {
+
+		/**
+		 * Configure projected fields.
+		 *
+		 * @param selectedFields must not be {@literal null}.
+		 */
+		S project(String... selectedFields);
+
+
+		/**
+		 * Configure {@link Sort}.
+		 *
+		 * @param sort must not be {@literal null}.
+		 */
+		S orderBy(Sort sort);
+
+		/**
+		 * Configure pagination. Overrides {@link Sort} if the {@link Pageable} contains a {@link Sort} object.
+		 *
+		 * @param page must not be {@literal null}.
+		 */
+		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);
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClientBuilder.java
+++ b/src/main/java/org/springframework/data/r2dbc/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.r2dbc.function;
+
+import io.r2dbc.spi.ConnectionFactory;
+
+import java.util.function.Consumer;
+
+import org.springframework.data.r2dbc.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;
+
+/**
+ * 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;
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultFetchSpec.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/DefaultFetchSpec.java
@@ -0,0 +1,85 @@
+/*
+ * 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.r2dbc.function;
+
+import io.r2dbc.spi.Connection;
+import lombok.RequiredArgsConstructor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Function;
+
+import org.springframework.dao.IncorrectResultSizeDataAccessException;
+
+/**
+ * Default implementation of {@link FetchSpec}.
+ *
+ * @author Mark Paluch
+ */
+@RequiredArgsConstructor
+class DefaultFetchSpec<T> implements FetchSpec<T> {
+
+	private final ConnectionAccessor connectionAccessor;
+	private final String sql;
+	private final Function<Connection, Flux<T>> resultFunction;
+	private final Function<Connection, Mono<Integer>> updatedRowsFunction;
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#one()
+	 */
+	@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();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#first()
+	 */
+	@Override
+	public Mono<T> first() {
+		return all().next();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#all()
+	 */
+	@Override
+	public Flux<T> all() {
+		return connectionAccessor.inConnectionMany(resultFunction);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#rowsUpdated()
+	 */
+	@Override
+	public Mono<Integer> rowsUpdated() {
+		return connectionAccessor.inConnection(updatedRowsFunction);
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java
@@ -0,0 +1,128 @@
+/*
+ * 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.r2dbc.function;
+
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.BiFunction;
+import java.util.stream.Collectors;
+
+import org.springframework.data.convert.EntityInstantiators;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Order;
+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.r2dbc.function.convert.EntityRowMapper;
+import org.springframework.data.util.Pair;
+import org.springframework.data.util.StreamUtils;
+import org.springframework.util.ClassUtils;
+
+/**
+ * @author Mark Paluch
+ */
+public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStrategy {
+
+	private final JdbcMappingContext mappingContext;
+	private final EntityInstantiators instantiators;
+
+	public DefaultReactiveDataAccessStrategy() {
+		this(new JdbcMappingContext(), new EntityInstantiators());
+	}
+
+	public DefaultReactiveDataAccessStrategy(JdbcMappingContext mappingContext, EntityInstantiators instantiators) {
+		this.mappingContext = mappingContext;
+		this.instantiators = instantiators;
+	}
+
+	@Override
+	public List<String> getAllFields(Class<?> typeToRead) {
+
+		JdbcPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(typeToRead);
+
+		if (persistentEntity == null) {
+			return Collections.singletonList("*");
+		}
+
+		return StreamUtils.createStreamFromIterator(persistentEntity.iterator()) //
+				.map(JdbcPersistentProperty::getColumnName) //
+				.collect(Collectors.toList());
+	}
+
+	@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 Sort getMappedSort(Class<?> typeToRead, Sort sort) {
+
+		JdbcPersistentEntity<?> entity = mappingContext.getPersistentEntity(typeToRead);
+		if (entity == null) {
+			return sort;
+		}
+
+		List<Order> mappedOrder = new ArrayList<>();
+
+		for (Order order : sort) {
+
+			JdbcPersistentProperty persistentProperty = entity.getPersistentProperty(order.getProperty());
+			if (persistentProperty == null) {
+				mappedOrder.add(order);
+			} else {
+				mappedOrder
+						.add(Order.by(persistentProperty.getColumnName()).with(order.getNullHandling()).with(order.getDirection()));
+			}
+		}
+
+		return Sort.by(mappedOrder);
+	}
+
+	@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();
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultSqlResult.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/DefaultSqlResult.java
@@ -0,0 +1,92 @@
+/*
+ * 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.r2dbc.function;
+
+import io.r2dbc.spi.Connection;
+import io.r2dbc.spi.Result;
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.function.BiFunction;
+import java.util.function.Function;
+
+/**
+ * Default {@link SqlResult} implementation.
+ *
+ * @author Mark Paluch
+ */
+class DefaultSqlResult<T> implements SqlResult<T> {
+
+	private final ConnectionAccessor connectionAccessor;
+	private final String sql;
+	private final Function<Connection, Flux<Result>> resultFunction;
+	private final Function<Connection, Mono<Integer>> updatedRowsFunction;
+	private final FetchSpec<T> fetchSpec;
+
+	DefaultSqlResult(ConnectionAccessor connectionAccessor, String sql, Function<Connection, Flux<Result>> resultFunction,
+			Function<Connection, Mono<Integer>> updatedRowsFunction, BiFunction<Row, RowMetadata, T> mappingFunction) {
+
+		this.sql = sql;
+		this.connectionAccessor = connectionAccessor;
+		this.resultFunction = resultFunction;
+		this.updatedRowsFunction = updatedRowsFunction;
+
+		this.fetchSpec = new DefaultFetchSpec<>(connectionAccessor, sql,
+				it -> resultFunction.apply(it).flatMap(result -> result.map(mappingFunction)), updatedRowsFunction);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.SqlResult#extract(java.util.function.BiFunction)
+	 */
+	@Override
+	public <R> SqlResult<R> extract(BiFunction<Row, RowMetadata, R> mappingFunction) {
+		return new DefaultSqlResult<>(connectionAccessor, sql, resultFunction, updatedRowsFunction, mappingFunction);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#one()
+	 */
+	@Override
+	public Mono<T> one() {
+		return fetchSpec.one();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#first()
+	 */
+	@Override
+	public Mono<T> first() {
+		return fetchSpec.first();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#all()
+	 */
+	@Override
+	public Flux<T> all() {
+		return fetchSpec.all();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.function.FetchSpec#rowsUpdated()
+	 */
+	@Override
+	public Mono<Integer> rowsUpdated() {
+		return fetchSpec.rowsUpdated();
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/function/FetchSpec.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/FetchSpec.java
@@ -0,0 +1,56 @@
+/*
+ * 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.r2dbc.function;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * Contract for fetching results.
+ *
+ * @author Mark Paluch
+ */
+public 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();
+}
--- a/src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java
@@ -0,0 +1,42 @@
+/*
+ * 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.r2dbc.function;
+
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+
+import java.util.List;
+import java.util.function.BiFunction;
+
+import org.springframework.data.domain.Sort;
+import org.springframework.data.util.Pair;
+
+/**
+ * @author Mark Paluch
+ */
+public interface ReactiveDataAccessStrategy {
+
+	List<String> getAllFields(Class<?> typeToRead);
+
+	List<Pair<String, Object>> getInsert(Object object);
+
+	Sort getMappedSort(Class<?> typeToRead, Sort sort);
+
+	// 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);
+}
--- a/src/main/java/org/springframework/data/r2dbc/function/SqlResult.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/SqlResult.java
@@ -0,0 +1,38 @@
+/*
+ * 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.r2dbc.function;
+
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+
+import java.util.function.BiFunction;
+
+/**
+ * Mappable {@link FetchSpec} that accepts a {@link BiFunction mapping function} to map SQL {@link Row}s.
+ *
+ * @author Mark Paluch
+ */
+public interface SqlResult<T> extends FetchSpec<T> {
+
+	/**
+	 * Apply a {@link BiFunction mapping function} to the result that emits {@link Row}s.
+	 *
+	 * @param mappingFunction must not be {@literal null}.
+	 * @param <R>
+	 * @return a new {@link SqlResult} with {@link BiFunction mapping function} applied.
+	 */
+	<R> SqlResult<R> extract(BiFunction<Row, RowMetadata, R> mappingFunction);
+}
--- a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionProxy.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionProxy.java
@@ -0,0 +1,40 @@
+/*
+ * 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.r2dbc.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}.
+ *
+ * @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();
+}
--- a/src/main/java/org/springframework/data/r2dbc/function/convert/ColumnMapRowMapper.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/convert/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.r2dbc.function.convert;
+
+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);
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/function/convert/EntityRowMapper.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/convert/EntityRowMapper.java
@@ -0,0 +1,167 @@
+/*
+ * 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.r2dbc.function.convert;
+
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+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;
+
+/**
+ * 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);
+			}
+		}
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/function/convert/IterableUtils.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/convert/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.r2dbc.function.convert;
+
+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;
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/function/convert/MappingR2dbcConverter.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/convert/MappingR2dbcConverter.java
@@ -0,0 +1,105 @@
+/*
+ * 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.r2dbc.function.convert;
+
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.BiFunction;
+
+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.mapping.context.MappingContext;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+/**
+ * Converter for R2DBC.
+ *
+ * @author Mark Paluch
+ */
+public class MappingR2dbcConverter {
+
+	private final MappingContext<? extends JdbcPersistentEntity<?>, JdbcPersistentProperty> mappingContext;
+
+	public MappingR2dbcConverter(
+			MappingContext<? extends JdbcPersistentEntity<?>, JdbcPersistentProperty> mappingContext) {
+		this.mappingContext = mappingContext;
+	}
+
+	/**
+	 * Returns a {@link Map} that maps column names to an {@link Optional} value. Used {@link Optional#empty()} if the
+	 * underlying property is {@literal null}.
+	 *
+	 * @param object must not be {@literal null}.
+	 * @return
+	 */
+	public Map<String, Optional<Object>> getFieldsToUpdate(Object object) {
+
+		Assert.notNull(object, "Entity object must not be null!");
+
+		Class<?> userClass = ClassUtils.getUserClass(object);
+		JdbcPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(userClass);
+
+		Map<String, Optional<Object>> update = new LinkedHashMap<>();
+
+		PersistentPropertyAccessor propertyAccessor = entity.getPropertyAccessor(object);
+
+		for (JdbcPersistentProperty property : entity) {
+			update.put(property.getColumnName(), Optional.ofNullable(propertyAccessor.getProperty(property)));
+		}
+
+		return update;
+	}
+
+	/**
+	 * Returns a {@link java.util.function.Function} that populates the id property of the {@code object} from a
+	 * {@link Row}.
+	 *
+	 * @param object must not be {@literal null}.
+	 * @return
+	 */
+	@SuppressWarnings("unchecked")
+	public <T> BiFunction<Row, RowMetadata, T> populateIdIfNecessary(T object) {
+
+		Assert.notNull(object, "Entity object must not be null!");
+
+		Class<?> userClass = ClassUtils.getUserClass(object);
+		JdbcPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(userClass);
+
+		return (row, metadata) -> {
+
+			PersistentPropertyAccessor propertyAccessor = entity.getPropertyAccessor(object);
+			JdbcPersistentProperty idProperty = entity.getRequiredIdProperty();
+
+			if (propertyAccessor.getProperty(idProperty) == null) {
+
+				propertyAccessor.setProperty(idProperty, row.get(idProperty.getColumnName(), idProperty.getColumnType()));
+				return (T) propertyAccessor.getBean();
+			}
+
+			return object;
+		};
+	}
+
+	public MappingContext<? extends JdbcPersistentEntity<?>, JdbcPersistentProperty> getMappingContext() {
+		return mappingContext;
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/repository/R2dbcRepository.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/R2dbcRepository.java
@@ -0,0 +1,27 @@
+/*
+ * 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.r2dbc.repository;
+
+import org.springframework.data.repository.NoRepositoryBean;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+
+/**
+ * R2DBC specific {@link org.springframework.data.repository.Repository} interface with reactive support.
+ *
+ * @author Mark Paluch
+ */
+@NoRepositoryBean
+public interface R2dbcRepository<T, ID> extends ReactiveCrudRepository<T, ID> {}
--- a/src/main/java/org/springframework/data/r2dbc/repository/package-info.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/package-info.java
@@ -0,0 +1,7 @@
+/**
+ * R2DBC-specific repository implementation.
+ */
+@NonNullApi
+package org.springframework.data.r2dbc.repository;
+
+import org.springframework.lang.NonNullApi;
--- a/src/main/java/org/springframework/data/r2dbc/repository/query/AbstractR2dbcQuery.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/query/AbstractR2dbcQuery.java
@@ -0,0 +1,149 @@
+/*
+ * 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.r2dbc.repository.query;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.reactivestreams.Publisher;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.convert.EntityInstantiators;
+import org.springframework.data.r2dbc.function.DatabaseClient;
+import org.springframework.data.r2dbc.function.DatabaseClient.GenericExecuteSpec;
+import org.springframework.data.r2dbc.function.FetchSpec;
+import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter;
+import org.springframework.data.r2dbc.repository.query.R2dbcQueryExecution.ResultProcessingConverter;
+import org.springframework.data.r2dbc.repository.query.R2dbcQueryExecution.ResultProcessingExecution;
+import org.springframework.data.relational.repository.query.RelationalParameterAccessor;
+import org.springframework.data.relational.repository.query.RelationalParametersParameterAccessor;
+import org.springframework.data.repository.query.ParameterAccessor;
+import org.springframework.data.repository.query.RepositoryQuery;
+import org.springframework.data.repository.query.ResultProcessor;
+import org.springframework.data.repository.query.ReturnedType;
+import org.springframework.util.Assert;
+
+/**
+ * Base class for reactive {@link RepositoryQuery} implementations for R2DBC.
+ *
+ * @author Mark Paluch
+ */
+public abstract class AbstractR2dbcQuery implements RepositoryQuery {
+
+	private final R2dbcQueryMethod method;
+	private final DatabaseClient databaseClient;
+	private final MappingR2dbcConverter converter;
+	private final EntityInstantiators instantiators;
+
+	/**
+	 * Creates a new {@link AbstractR2dbcQuery} from the given {@link R2dbcQueryMethod} and {@link DatabaseClient}.
+	 *
+	 * @param method must not be {@literal null}.
+	 * @param databaseClient must not be {@literal null}.
+	 * @param converter must not be {@literal null}.
+	 */
+	public AbstractR2dbcQuery(R2dbcQueryMethod method, DatabaseClient databaseClient, MappingR2dbcConverter converter) {
+
+		Assert.notNull(method, "R2dbcQueryMethod must not be null!");
+		Assert.notNull(databaseClient, "DatabaseClient must not be null!");
+		Assert.notNull(converter, "MappingR2dbcConverter must not be null!");
+
+		this.method = method;
+		this.databaseClient = databaseClient;
+		this.converter = converter;
+		this.instantiators = new EntityInstantiators();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.RepositoryQuery#getQueryMethod()
+	 */
+	public R2dbcQueryMethod getQueryMethod() {
+		return method;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.RepositoryQuery#execute(java.lang.Object[])
+	 */
+	public Object execute(Object[] parameters) {
+
+		return method.hasReactiveWrapperParameter() ? executeDeferred(parameters)
+				: execute(new RelationalParametersParameterAccessor(method, parameters));
+	}
+
+	@SuppressWarnings("unchecked")
+	private Object executeDeferred(Object[] parameters) {
+
+		R2dbcParameterAccessor parameterAccessor = new R2dbcParameterAccessor(method, parameters);
+
+		if (getQueryMethod().isCollectionQuery()) {
+			return Flux.defer(() -> (Publisher<Object>) execute(parameterAccessor));
+		}
+
+		return Mono.defer(() -> (Mono<Object>) execute(parameterAccessor));
+	}
+
+	private Object execute(RelationalParameterAccessor parameterAccessor) {
+
+		// TODO: ConvertingParameterAccessor
+		BindableQuery query = createQuery(parameterAccessor);
+
+		ResultProcessor processor = method.getResultProcessor().withDynamicProjection(parameterAccessor);
+		GenericExecuteSpec boundQuery = query.bind(databaseClient.execute().sql(query));
+		FetchSpec<?> fetchSpec = boundQuery.as(resolveResultType(processor)).fetch();
+
+		String tableName = method.getEntityInformation().getTableName();
+
+		R2dbcQueryExecution execution = getExecution(
+				new ResultProcessingConverter(processor, converter.getMappingContext(), instantiators));
+
+		return execution.execute(fetchSpec, processor.getReturnedType().getDomainType(), tableName);
+	}
+
+	private Class<?> resolveResultType(ResultProcessor resultProcessor) {
+
+		ReturnedType returnedType = resultProcessor.getReturnedType();
+
+		return returnedType.isProjecting() ? returnedType.getDomainType() : returnedType.getReturnedType();
+	}
+
+	/**
+	 * Returns the execution instance to use.
+	 *
+	 * @param resultProcessing must not be {@literal null}.
+	 * @return
+	 */
+	private R2dbcQueryExecution getExecution(Converter<Object, Object> resultProcessing) {
+		return new ResultProcessingExecution(getExecutionToWrap(), resultProcessing);
+	}
+
+	private R2dbcQueryExecution getExecutionToWrap() {
+
+		if (method.isCollectionQuery()) {
+			return (q, t, c) -> q.all();
+		}
+
+		return (q, t, c) -> q.one();
+	}
+
+	/**
+	 * Creates a {@link BindableQuery} instance using the given {@link ParameterAccessor}
+	 *
+	 * @param accessor must not be {@literal null}.
+	 * @return
+	 */
+	protected abstract BindableQuery createQuery(RelationalParameterAccessor accessor);
+}
--- a/src/main/java/org/springframework/data/r2dbc/repository/query/BindableQuery.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/query/BindableQuery.java
@@ -0,0 +1,36 @@
+/*
+ * 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.r2dbc.repository.query;
+
+import java.util.function.Supplier;
+
+import org.springframework.data.r2dbc.function.DatabaseClient.BindSpec;
+
+/**
+ * Interface declaring a query that supplies SQL and can bind parameters to a {@link BindSpec}.
+ *
+ * @author Mark Paluch
+ */
+public interface BindableQuery extends Supplier<String> {
+
+	/**
+	 * Bind parameters to the {@link BindSpec query}.
+	 *
+	 * @param bindSpec must not be {@literal null}.
+	 * @return the bound query object.
+	 */
+	<T extends BindSpec<T>> T bind(T bindSpec);
+}
--- a/src/main/java/org/springframework/data/r2dbc/repository/query/R2dbcParameterAccessor.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/query/R2dbcParameterAccessor.java
@@ -0,0 +1,100 @@
+/*
+ * 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.r2dbc.repository.query;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.data.relational.repository.query.RelationalParametersParameterAccessor;
+import org.springframework.data.repository.util.ReactiveWrapperConverters;
+import org.springframework.data.repository.util.ReactiveWrappers;
+
+/**
+ * Reactive {@link org.springframework.data.repository.query.ParametersParameterAccessor} implementation that subscribes
+ * to reactive parameter wrapper types upon creation. This class performs synchronization when accessing parameters.
+ *
+ * @author Mark Paluch
+ */
+class R2dbcParameterAccessor extends RelationalParametersParameterAccessor {
+
+	private final Object[] values;
+	private final List<MonoProcessor<?>> subscriptions;
+
+	/**
+	 * Creates a new {@link R2dbcParameterAccessor}.
+	 */
+	public R2dbcParameterAccessor(R2dbcQueryMethod method, Object... values) {
+
+		super(method, values);
+
+		this.values = values;
+		this.subscriptions = new ArrayList<>(values.length);
+
+		for (int i = 0; i < values.length; i++) {
+
+			Object value = values[i];
+
+			if (value == null || !ReactiveWrappers.supports(value.getClass())) {
+				subscriptions.add(null);
+				continue;
+			}
+
+			if (ReactiveWrappers.isSingleValueType(value.getClass())) {
+				subscriptions.add(ReactiveWrapperConverters.toWrapper(value, Mono.class).toProcessor());
+			} else {
+				subscriptions.add(ReactiveWrapperConverters.toWrapper(value, Flux.class).collectList().toProcessor());
+			}
+		}
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.query.ParametersParameterAccessor#getValue(int)
+	 */
+	@SuppressWarnings("unchecked")
+	@Override
+	protected <T> T getValue(int index) {
+
+		if (subscriptions.get(index) != null) {
+			return (T) subscriptions.get(index).block();
+		}
+
+		return super.getValue(index);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.repository.query.JdbcParametersParameterAccessor#getValues()
+	 */
+	@Override
+	public Object[] getValues() {
+
+		Object[] result = new Object[values.length];
+		for (int i = 0; i < result.length; i++) {
+			result[i] = getValue(i);
+		}
+		return result;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.query.ParametersParameterAccessor#getBindableValue(int)
+	 */
+	public Object getBindableValue(int index) {
+		return getValue(getParameters().getBindableParameter(index).getIndex());
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/repository/query/R2dbcQueryExecution.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/query/R2dbcQueryExecution.java
@@ -0,0 +1,88 @@
+/*
+ * 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.r2dbc.repository.query;
+
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.convert.EntityInstantiators;
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentEntity;
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentProperty;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.r2dbc.function.FetchSpec;
+import org.springframework.data.relational.repository.query.DtoInstantiatingConverter;
+import org.springframework.data.repository.query.ResultProcessor;
+import org.springframework.data.repository.query.ReturnedType;
+import org.springframework.util.ClassUtils;
+
+/**
+ * Set of classes to contain query execution strategies. Depending (mostly) on the return type of a
+ * {@link org.springframework.data.repository.query.QueryMethod}.
+ *
+ * @author Mark Paluch
+ */
+interface R2dbcQueryExecution {
+
+	Object execute(FetchSpec<?> query, Class<?> type, String tableName);
+
+	/**
+	 * An {@link R2dbcQueryExecution} that wraps the results of the given delegate with the given result processing.
+	 */
+	@RequiredArgsConstructor
+	final class ResultProcessingExecution implements R2dbcQueryExecution {
+
+		private final @NonNull R2dbcQueryExecution delegate;
+		private final @NonNull Converter<Object, Object> converter;
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.jdbc.repository.query.R2dbcQueryExecution#execute(org.springframework.data.jdbc.core.function.FetchSpec, java.lang.Class, java.lang.String)
+		 */
+		@Override
+		public Object execute(FetchSpec<?> query, Class<?> type, String tableName) {
+			return converter.convert(delegate.execute(query, type, tableName));
+		}
+	}
+
+	/**
+	 * A {@link Converter} to post-process all source objects using the given {@link ResultProcessor}.
+	 */
+	@RequiredArgsConstructor
+	final class ResultProcessingConverter implements Converter<Object, Object> {
+
+		private final @NonNull ResultProcessor processor;
+		private final @NonNull MappingContext<? extends JdbcPersistentEntity<?>, JdbcPersistentProperty> mappingContext;
+		private final @NonNull EntityInstantiators instantiators;
+
+		/* (non-Javadoc)
+		 * @see org.springframework.core.convert.converter.Converter#convert(java.lang.Object)
+		 */
+		@Override
+		public Object convert(Object source) {
+
+			ReturnedType returnedType = processor.getReturnedType();
+
+			if (ClassUtils.isPrimitiveOrWrapper(returnedType.getReturnedType())) {
+				return source;
+			}
+
+			Converter<Object, Object> converter = new DtoInstantiatingConverter(returnedType.getReturnedType(),
+					mappingContext, instantiators);
+
+			return processor.processResult(source, converter);
+		}
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/repository/query/R2dbcQueryMethod.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/query/R2dbcQueryMethod.java
@@ -0,0 +1,231 @@
+/*
+ * 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.r2dbc.repository.query;
+
+import static org.springframework.data.repository.util.ClassUtils.*;
+
+import java.lang.reflect.Method;
+import java.util.Optional;
+
+import org.springframework.core.annotation.AnnotatedElementUtils;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Slice;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentEntity;
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentProperty;
+import org.springframework.data.jdbc.repository.query.Query;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.projection.ProjectionFactory;
+import org.springframework.data.relational.repository.query.RelationalEntityMetadata;
+import org.springframework.data.relational.repository.query.RelationalParameters;
+import org.springframework.data.relational.repository.query.SimpleRelationalEntityMetadata;
+import org.springframework.data.repository.core.RepositoryMetadata;
+import org.springframework.data.repository.query.Parameter;
+import org.springframework.data.repository.query.QueryMethod;
+import org.springframework.data.repository.util.ReactiveWrapperConverters;
+import org.springframework.data.repository.util.ReactiveWrappers;
+import org.springframework.data.util.ClassTypeInformation;
+import org.springframework.data.util.TypeInformation;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+/**
+ * Reactive specific implementation of {@link QueryMethod}.
+ *
+ * @author Mark Paluch
+ */
+public class R2dbcQueryMethod extends QueryMethod {
+
+	private static final ClassTypeInformation<Page> PAGE_TYPE = ClassTypeInformation.from(Page.class);
+	private static final ClassTypeInformation<Slice> SLICE_TYPE = ClassTypeInformation.from(Slice.class);
+
+	private final Method method;
+	private final MappingContext<? extends JdbcPersistentEntity<?>, JdbcPersistentProperty> mappingContext;
+	private final Optional<Query> query;
+
+	private @Nullable RelationalEntityMetadata<?> metadata;
+
+	/**
+	 * Creates a new {@link R2dbcQueryMethod} from the given {@link Method}.
+	 *
+	 * @param method must not be {@literal null}.
+	 * @param metadata must not be {@literal null}.
+	 * @param projectionFactory must not be {@literal null}.
+	 * @param mappingContext must not be {@literal null}.
+	 */
+	public R2dbcQueryMethod(Method method, RepositoryMetadata metadata, ProjectionFactory projectionFactory,
+			MappingContext<? extends JdbcPersistentEntity<?>, JdbcPersistentProperty> mappingContext) {
+
+		super(method, metadata, projectionFactory);
+
+		Assert.notNull(mappingContext, "MappingContext must not be null!");
+
+		this.mappingContext = mappingContext;
+
+		if (hasParameterOfType(method, Pageable.class)) {
+
+			TypeInformation<?> returnType = ClassTypeInformation.fromReturnTypeOf(method);
+
+			boolean multiWrapper = ReactiveWrappers.isMultiValueType(returnType.getType());
+			boolean singleWrapperWithWrappedPageableResult = ReactiveWrappers.isSingleValueType(returnType.getType())
+					&& (PAGE_TYPE.isAssignableFrom(returnType.getRequiredComponentType())
+							|| SLICE_TYPE.isAssignableFrom(returnType.getRequiredComponentType()));
+
+			if (singleWrapperWithWrappedPageableResult) {
+				throw new InvalidDataAccessApiUsageException(
+						String.format("'%s.%s' must not use sliced or paged execution. Please use Flux.buffer(size, skip).",
+								ClassUtils.getShortName(method.getDeclaringClass()), method.getName()));
+			}
+
+			if (!multiWrapper) {
+				throw new IllegalStateException(String.format(
+						"Method has to use a either multi-item reactive wrapper return type or a wrapped Page/Slice type. Offending method: %s",
+						method.toString()));
+			}
+
+			if (hasParameterOfType(method, Sort.class)) {
+				throw new IllegalStateException(String.format("Method must not have Pageable *and* Sort parameter. "
+						+ "Use sorting capabilities on Pageble instead! Offending method: %s", method.toString()));
+			}
+		}
+
+		this.method = method;
+		this.query = Optional.ofNullable(AnnotatedElementUtils.findMergedAnnotation(method, Query.class));
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.query.QueryMethod#createParameters(java.lang.reflect.Method)
+	 */
+	@Override
+	protected RelationalParameters createParameters(Method method) {
+		return new RelationalParameters(method);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.query.QueryMethod#isCollectionQuery()
+	 */
+	@Override
+	public boolean isCollectionQuery() {
+		return !(isPageQuery() || isSliceQuery()) && ReactiveWrappers.isMultiValueType(method.getReturnType());
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.query.QueryMethod#isModifyingQuery()
+	 */
+	@Override
+	public boolean isModifyingQuery() {
+		return super.isModifyingQuery();
+	}
+
+	/*
+	 * All reactive query methods are streaming queries.
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.QueryMethod#isStreamQuery()
+	 */
+	@Override
+	public boolean isStreamQuery() {
+		return true;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.query.QueryMethod#getEntityInformation()
+	 */
+	@Override
+	@SuppressWarnings("unchecked")
+	public RelationalEntityMetadata<?> getEntityInformation() {
+
+		if (metadata == null) {
+
+			Class<?> returnedObjectType = getReturnedObjectType();
+			Class<?> domainClass = getDomainClass();
+
+			if (ClassUtils.isPrimitiveOrWrapper(returnedObjectType)) {
+
+				this.metadata = new SimpleRelationalEntityMetadata<>((Class<Object>) domainClass,
+						mappingContext.getRequiredPersistentEntity(domainClass));
+
+			} else {
+
+				JdbcPersistentEntity<?> returnedEntity = mappingContext.getPersistentEntity(returnedObjectType);
+				JdbcPersistentEntity<?> managedEntity = mappingContext.getRequiredPersistentEntity(domainClass);
+				returnedEntity = returnedEntity == null || returnedEntity.getType().isInterface() ? managedEntity
+						: returnedEntity;
+				JdbcPersistentEntity<?> tableEntity = domainClass.isAssignableFrom(returnedObjectType) ? returnedEntity
+						: managedEntity;
+
+				this.metadata = new SimpleRelationalEntityMetadata<>((Class<Object>) returnedEntity.getType(), tableEntity);
+			}
+		}
+
+		return this.metadata;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.query.QueryMethod#getParameters()
+	 */
+	@Override
+	public RelationalParameters getParameters() {
+		return (RelationalParameters) super.getParameters();
+	}
+
+	/**
+	 * Check if the given {@link org.springframework.data.repository.query.QueryMethod} receives a reactive parameter
+	 * wrapper as one of its parameters.
+	 *
+	 * @return {@literal true} if the given {@link org.springframework.data.repository.query.QueryMethod} receives a
+	 *         reactive parameter wrapper as one of its parameters.
+	 */
+	public boolean hasReactiveWrapperParameter() {
+
+		for (Parameter parameter : getParameters()) {
+			if (ReactiveWrapperConverters.supports(parameter.getType())) {
+				return true;
+			}
+		}
+		return false;
+	}
+
+	/**
+	 * Returns the required query string declared in a {@link Query} annotation or throws {@link IllegalStateException} if
+	 * neither the annotation found nor the attribute was specified.
+	 *
+	 * @return the query string.
+	 * @throws IllegalStateException in case query method has no annotated query.
+	 */
+	public String getRequiredAnnotatedQuery() {
+		return this.query.map(Query::value)
+				.orElseThrow(() -> new IllegalStateException("Query method " + this + " has no annotated query"));
+	}
+
+	/**
+	 * Returns the {@link Query} annotation that is applied to the method or {@literal null} if none available.
+	 *
+	 * @return the optional query annotation.
+	 */
+	Optional<Query> getQueryAnnotation() {
+		return this.query;
+	}
+
+	/**
+	 * @return {@literal true} if the {@link Method} is annotated with {@link Query}.
+	 */
+	public boolean hasAnnotatedQuery() {
+		return getQueryAnnotation().isPresent();
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/repository/query/StringBasedR2dbcQuery.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/query/StringBasedR2dbcQuery.java
@@ -0,0 +1,112 @@
+/*
+ * 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.r2dbc.repository.query;
+
+import org.springframework.data.jdbc.repository.query.Query;
+import org.springframework.data.r2dbc.function.DatabaseClient;
+import org.springframework.data.r2dbc.function.DatabaseClient.BindSpec;
+import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter;
+import org.springframework.data.relational.repository.query.RelationalParameterAccessor;
+import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.util.Assert;
+
+/**
+ * String-based {@link StringBasedR2dbcQuery} implementation.
+ * <p>
+ * A {@link StringBasedR2dbcQuery} expects a query method to be annotated with {@link Query} with a SQL query.
+ *
+ * @author Mark Paluch
+ */
+public class StringBasedR2dbcQuery extends AbstractR2dbcQuery {
+
+	private final String sql;
+
+	/**
+	 * Creates a new {@link StringBasedR2dbcQuery} for the given {@link StringBasedR2dbcQuery}, {@link DatabaseClient},
+	 * {@link SpelExpressionParser}, and {@link QueryMethodEvaluationContextProvider}.
+	 *
+	 * @param queryMethod must not be {@literal null}.
+	 * @param databaseClient must not be {@literal null}.
+	 * @param converter must not be {@literal null}.
+	 * @param expressionParser must not be {@literal null}.
+	 * @param evaluationContextProvider must not be {@literal null}.
+	 */
+	public StringBasedR2dbcQuery(R2dbcQueryMethod queryMethod, DatabaseClient databaseClient,
+			MappingR2dbcConverter converter, SpelExpressionParser expressionParser,
+			QueryMethodEvaluationContextProvider evaluationContextProvider) {
+
+		this(queryMethod.getRequiredAnnotatedQuery(), queryMethod, databaseClient, converter, expressionParser,
+				evaluationContextProvider);
+	}
+
+	/**
+	 * Create a new {@link StringBasedR2dbcQuery} for the given {@code query}, {@link R2dbcQueryMethod},
+	 * {@link DatabaseClient}, {@link SpelExpressionParser}, and {@link QueryMethodEvaluationContextProvider}.
+	 *
+	 * @param method must not be {@literal null}.
+	 * @param databaseClient must not be {@literal null}.
+	 * @param converter must not be {@literal null}.
+	 * @param expressionParser must not be {@literal null}.
+	 * @param evaluationContextProvider must not be {@literal null}.
+	 */
+	public StringBasedR2dbcQuery(String query, R2dbcQueryMethod method, DatabaseClient databaseClient,
+			MappingR2dbcConverter converter, SpelExpressionParser expressionParser,
+			QueryMethodEvaluationContextProvider evaluationContextProvider) {
+
+		super(method, databaseClient, converter);
+
+		Assert.hasText(query, "Query must not be empty");
+
+		this.sql = query;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.repository.query.AbstractR2dbcQuery#createQuery(org.springframework.data.jdbc.repository.query.JdbcParameterAccessor)
+	 */
+	@Override
+	protected BindableQuery createQuery(RelationalParameterAccessor accessor) {
+
+		return new BindableQuery() {
+
+			@Override
+			public <T extends BindSpec<T>> T bind(T bindSpec) {
+
+				T bindSpecToUse = bindSpec;
+
+				// TODO: Encapsulate PostgreSQL-specific bindings
+				int index = 1;
+				for (Object value : accessor.getValues()) {
+
+					if (value == null) {
+						if (accessor.hasBindableNullValue()) {
+							bindSpecToUse = bindSpecToUse.bindNull("$" + (index++));
+						}
+					} else {
+						bindSpecToUse = bindSpecToUse.bind("$" + (index++), value);
+					}
+				}
+
+				return bindSpecToUse;
+			}
+
+			@Override
+			public String get() {
+				return sql;
+			}
+		};
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/repository/query/package-info.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/query/package-info.java
@@ -0,0 +1,9 @@
+/**
+ * Query support for R2DBC repositories.
+ */
+@NonNullApi
+@NonNullFields
+package org.springframework.data.r2dbc.repository.query;
+
+import org.springframework.lang.NonNullApi;
+import org.springframework.lang.NonNullFields;
--- a/src/main/java/org/springframework/data/r2dbc/repository/support/R2dbcRepositoryFactory.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/support/R2dbcRepositoryFactory.java
@@ -0,0 +1,162 @@
+/*
+ * 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.r2dbc.repository.support;
+
+import lombok.AccessLevel;
+import lombok.RequiredArgsConstructor;
+
+import java.lang.reflect.Method;
+import java.util.Optional;
+
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentEntity;
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentProperty;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.projection.ProjectionFactory;
+import org.springframework.data.r2dbc.function.DatabaseClient;
+import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter;
+import org.springframework.data.r2dbc.repository.R2dbcRepository;
+import org.springframework.data.r2dbc.repository.query.R2dbcQueryMethod;
+import org.springframework.data.r2dbc.repository.query.StringBasedR2dbcQuery;
+import org.springframework.data.relational.repository.query.RelationalEntityInformation;
+import org.springframework.data.relational.repository.support.MappingRelationalEntityInformation;
+import org.springframework.data.repository.core.NamedQueries;
+import org.springframework.data.repository.core.RepositoryInformation;
+import org.springframework.data.repository.core.RepositoryMetadata;
+import org.springframework.data.repository.core.support.ReactiveRepositoryFactorySupport;
+import org.springframework.data.repository.query.QueryLookupStrategy;
+import org.springframework.data.repository.query.QueryLookupStrategy.Key;
+import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.data.repository.query.RepositoryQuery;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Factory to create {@link R2dbcRepository} instances.
+ *
+ * @author Mark Paluch
+ */
+public class R2dbcRepositoryFactory extends ReactiveRepositoryFactorySupport {
+
+	private static final SpelExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
+
+	private final DatabaseClient databaseClient;
+	private final MappingContext<? extends JdbcPersistentEntity<?>, JdbcPersistentProperty> mappingContext;
+	private final MappingR2dbcConverter converter;
+
+	/**
+	 * Creates a new {@link R2dbcRepositoryFactory} given {@link DatabaseClient} and {@link MappingContext}.
+	 *
+	 * @param databaseClient must not be {@literal null}.
+	 * @param mappingContext must not be {@literal null}.
+	 */
+	public R2dbcRepositoryFactory(DatabaseClient databaseClient,
+			MappingContext<? extends JdbcPersistentEntity<?>, JdbcPersistentProperty> mappingContext) {
+
+		Assert.notNull(databaseClient, "DatabaseClient must not be null!");
+		Assert.notNull(mappingContext, "MappingContext must not be null!");
+
+		this.databaseClient = databaseClient;
+		this.mappingContext = mappingContext;
+		this.converter = new MappingR2dbcConverter(mappingContext);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getRepositoryBaseClass(org.springframework.data.repository.core.RepositoryMetadata)
+	 */
+	@Override
+	protected Class<?> getRepositoryBaseClass(RepositoryMetadata metadata) {
+		return SimpleR2dbcRepository.class;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getTargetRepository(org.springframework.data.repository.core.RepositoryInformation)
+	 */
+	@Override
+	protected Object getTargetRepository(RepositoryInformation information) {
+
+		RelationalEntityInformation<?, ?> entityInformation = getEntityInformation(information.getDomainType(),
+				information);
+
+		return getTargetRepositoryViaReflection(information, entityInformation, databaseClient, converter);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getQueryLookupStrategy(org.springframework.data.repository.query.QueryLookupStrategy.Key, org.springframework.data.repository.query.EvaluationContextProvider)
+	 */
+	@Override
+	protected Optional<QueryLookupStrategy> getQueryLookupStrategy(@Nullable Key key,
+			QueryMethodEvaluationContextProvider evaluationContextProvider) {
+		return Optional.of(new R2dbcQueryLookupStrategy(databaseClient, evaluationContextProvider, converter));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getEntityInformation(java.lang.Class)
+	 */
+	public <T, ID> RelationalEntityInformation<T, ID> getEntityInformation(Class<T> domainClass) {
+		return getEntityInformation(domainClass, null);
+	}
+
+	@SuppressWarnings("unchecked")
+	private <T, ID> RelationalEntityInformation<T, ID> getEntityInformation(Class<T> domainClass,
+			@Nullable RepositoryInformation information) {
+
+		JdbcPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(domainClass);
+
+		return new MappingRelationalEntityInformation<>((JdbcPersistentEntity<T>) entity);
+	}
+
+	/**
+	 * {@link QueryLookupStrategy} to create R2DBC queries..
+	 *
+	 * @author Mark Paluch
+	 */
+	@RequiredArgsConstructor(access = AccessLevel.PACKAGE)
+	private static class R2dbcQueryLookupStrategy implements QueryLookupStrategy {
+
+		private final DatabaseClient databaseClient;
+		private final QueryMethodEvaluationContextProvider evaluationContextProvider;
+		private final MappingR2dbcConverter converter;
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.repository.query.QueryLookupStrategy#resolveQuery(java.lang.reflect.Method, org.springframework.data.repository.core.RepositoryMetadata, org.springframework.data.projection.ProjectionFactory, org.springframework.data.repository.core.NamedQueries)
+		 */
+		@Override
+		public RepositoryQuery resolveQuery(Method method, RepositoryMetadata metadata, ProjectionFactory factory,
+				NamedQueries namedQueries) {
+
+			R2dbcQueryMethod queryMethod = new R2dbcQueryMethod(method, metadata, factory, converter.getMappingContext());
+			String namedQueryName = queryMethod.getNamedQueryName();
+
+			if (namedQueries.hasQuery(namedQueryName)) {
+				String namedQuery = namedQueries.getQuery(namedQueryName);
+				return new StringBasedR2dbcQuery(namedQuery, queryMethod, databaseClient, converter, EXPRESSION_PARSER,
+						evaluationContextProvider);
+			} else if (queryMethod.hasAnnotatedQuery()) {
+				return new StringBasedR2dbcQuery(queryMethod, databaseClient, converter, EXPRESSION_PARSER,
+						evaluationContextProvider);
+			}
+
+			throw new UnsupportedOperationException("Query derivation not yet supported!");
+
+		}
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java
@@ -0,0 +1,340 @@
+/*
+ * 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.r2dbc.repository.support;
+
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.reactivestreams.Publisher;
+import org.springframework.data.r2dbc.function.DatabaseClient;
+import org.springframework.data.r2dbc.function.DatabaseClient.BindSpec;
+import org.springframework.data.r2dbc.function.DatabaseClient.GenericExecuteSpec;
+import org.springframework.data.r2dbc.function.FetchSpec;
+import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter;
+import org.springframework.data.relational.repository.query.RelationalEntityInformation;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+import org.springframework.util.Assert;
+
+/**
+ * Simple {@link ReactiveCrudRepository} implementation using R2DBC through {@link DatabaseClient}.
+ *
+ * @author Mark Paluch
+ */
+@RequiredArgsConstructor
+public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, ID> {
+
+	private final @NonNull RelationalEntityInformation<T, ID> entity;
+	private final @NonNull DatabaseClient databaseClient;
+	private final @NonNull MappingR2dbcConverter converter;
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#save(S)
+	 */
+	@Override
+	public <S extends T> Mono<S> save(S objectToSave) {
+
+		Assert.notNull(objectToSave, "Object to save must not be null!");
+
+		if (entity.isNew(objectToSave)) {
+
+			return databaseClient.insert() //
+					.into(entity.getJavaType()) //
+					.using(objectToSave) //
+					.exchange() //
+					.flatMap(it -> it.extract(converter.populateIdIfNecessary(objectToSave)).one());
+		}
+
+		// TODO: Extract in some kind of SQL generator
+		Object id = entity.getRequiredId(objectToSave);
+
+		Map<String, Optional<Object>> fields = converter.getFieldsToUpdate(objectToSave);
+
+		String setClause = getSetClause(fields);
+
+		GenericExecuteSpec exec = databaseClient.execute()
+				.sql(String.format("UPDATE %s SET %s WHERE %s = $1", entity.getTableName(), setClause, getIdColumnName())) //
+				.bind(0, id);
+
+		int index = 1;
+		for (Optional<Object> setValue : fields.values()) {
+
+			Object value = setValue.orElse(null);
+			if (value != null) {
+				exec = exec.bind(index++, value);
+			} else {
+				exec = exec.bindNull(index++);
+			}
+		}
+
+		return exec.as(entity.getJavaType()) //
+				.exchange() //
+				.flatMap(FetchSpec::rowsUpdated) //
+				.thenReturn(objectToSave);
+	}
+
+	private static String getSetClause(Map<String, Optional<Object>> fields) {
+
+		StringBuilder setClause = new StringBuilder();
+
+		int index = 2;
+		for (String field : fields.keySet()) {
+
+			if (setClause.length() != 0) {
+				setClause.append(", ");
+			}
+
+			setClause.append(field).append('=').append('$').append(index++);
+		}
+
+		return setClause.toString();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#saveAll(java.lang.Iterable)
+	 */
+	@Override
+	public <S extends T> Flux<S> saveAll(Iterable<S> objectsToSave) {
+
+		Assert.notNull(objectsToSave, "Objects to save must not be null!");
+
+		return Flux.fromIterable(objectsToSave).flatMap(this::save);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#saveAll(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public <S extends T> Flux<S> saveAll(Publisher<S> objectsToSave) {
+
+		Assert.notNull(objectsToSave, "Object publisher must not be null!");
+
+		return Flux.from(objectsToSave).flatMap(this::save);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#findById(java.lang.Object)
+	 */
+	@Override
+	public Mono<T> findById(ID id) {
+
+		Assert.notNull(id, "Id must not be null!");
+
+		// TODO: Generate proper SQL (select, where clause, parameter binding).
+		return databaseClient.execute()
+				.sql(String.format("SELECT * FROM %s WHERE %s = $1", entity.getTableName(), getIdColumnName())) //
+				.bind("$1", id) //
+				.as(entity.getJavaType()) //
+				.fetch() //
+				.one();
+
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#findById(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Mono<T> findById(Publisher<ID> publisher) {
+		return Mono.from(publisher).flatMap(this::findById);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#existsById(java.lang.Object)
+	 */
+	@Override
+	public Mono<Boolean> existsById(ID id) {
+
+		Assert.notNull(id, "Id must not be null!");
+
+		// TODO: Generate proper SQL (select, where clause, parameter binding).
+		return databaseClient.execute()
+				.sql(String.format("SELECT %s FROM %s WHERE %s = $1 LIMIT 1", getIdColumnName(), entity.getTableName(),
+						getIdColumnName())) //
+				.bind("$1", id) //
+				.exchange() //
+				.flatMap(it -> it.extract((r, md) -> r).first()).hasElement();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#existsById(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Mono<Boolean> existsById(Publisher<ID> publisher) {
+		return Mono.from(publisher).flatMap(this::findById).hasElement();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#findAll()
+	 */
+	@Override
+	public Flux<T> findAll() {
+		return databaseClient.select().from(entity.getJavaType()).fetch().all();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#findAllById(java.lang.Iterable)
+	 */
+	@Override
+	public Flux<T> findAllById(Iterable<ID> iterable) {
+
+		Assert.notNull(iterable, "The iterable of Id's must not be null!");
+
+		return findAllById(Flux.fromIterable(iterable));
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#findAllById(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Flux<T> findAllById(Publisher<ID> idPublisher) {
+
+		Assert.notNull(idPublisher, "The Id Publisher must not be null!");
+
+		return Flux.from(idPublisher).buffer().filter(ids -> !ids.isEmpty()).flatMap(ids -> {
+
+			String bindings = getInBinding(ids);
+
+			GenericExecuteSpec exec = databaseClient.execute()
+					.sql(String.format("SELECT * FROM %s WHERE %s IN (%s)", entity.getTableName(), getIdColumnName(), bindings));
+
+			return bind(ids, exec).as(entity.getJavaType()).fetch().all();
+		});
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#count()
+	 */
+	@Override
+	public Mono<Long> count() {
+
+		return databaseClient.execute()
+				.sql(String.format("SELECT COUNT(%s) FROM %s", getIdColumnName(), entity.getTableName())) //
+				.exchange() //
+				.flatMap(it -> it.extract((r, md) -> r.get(0, Long.class)).first()) //
+				.defaultIfEmpty(0L);
+
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#deleteById(java.lang.Object)
+	 */
+	@Override
+	public Mono<Void> deleteById(ID id) {
+
+		Assert.notNull(id, "Id must not be null!");
+
+		return databaseClient.execute()
+				.sql(String.format("DELETE FROM %s WHERE %s = $1", entity.getTableName(), getIdColumnName())) //
+				.bind("$1", id) //
+				.fetch() //
+				.rowsUpdated() //
+				.then();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#deleteById(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Mono<Void> deleteById(Publisher<ID> idPublisher) {
+
+		Assert.notNull(idPublisher, "The Id Publisher must not be null!");
+
+		return Flux.from(idPublisher).buffer().filter(ids -> !ids.isEmpty()).flatMap(ids -> {
+
+			String bindings = getInBinding(ids);
+
+			GenericExecuteSpec exec = databaseClient.execute()
+					.sql(String.format("DELETE FROM %s WHERE %s IN (%s)", entity.getTableName(), getIdColumnName(), bindings));
+
+			return bind(ids, exec).as(entity.getJavaType()).fetch().rowsUpdated();
+		}).then();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#delete(java.lang.Object)
+	 */
+	@Override
+	@SuppressWarnings("unchecked")
+	public Mono<Void> delete(T objectToDelete) {
+
+		Assert.notNull(objectToDelete, "Object to delete must not be null!");
+
+		return deleteById(entity.getRequiredId(objectToDelete));
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#deleteAll(java.lang.Iterable)
+	 */
+	@Override
+	public Mono<Void> deleteAll(Iterable<? extends T> iterable) {
+
+		Assert.notNull(iterable, "The iterable of Id's must not be null!");
+
+		return deleteAll(Flux.fromIterable(iterable));
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#deleteAll(org.reactivestreams.Publisher)
+	 */
+	@Override
+	@SuppressWarnings("unchecked")
+	public Mono<Void> deleteAll(Publisher<? extends T> objectPublisher) {
+
+		Assert.notNull(objectPublisher, "The Object Publisher must not be null!");
+
+		Flux<ID> idPublisher = Flux.from(objectPublisher) //
+				.map(entity::getRequiredId);
+
+		return deleteById(idPublisher);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#deleteAll()
+	 */
+	@Override
+	public Mono<Void> deleteAll() {
+
+		return databaseClient.execute().sql(String.format("DELETE FROM %s", entity.getTableName())) //
+				.exchange() //
+				.then();
+	}
+
+	private String getInBinding(List<ID> ids) {
+		return IntStream.range(1, ids.size() + 1).mapToObj(i -> "$" + i).collect(Collectors.joining(", "));
+	}
+
+	@SuppressWarnings("unchecked")
+	private <S extends BindSpec<?>> S bind(List<ID> it, S bindSpec) {
+
+		for (int i = 0; i < it.size(); i++) {
+			bindSpec = (S) bindSpec.bind(i, it.get(i));
+		}
+
+		return bindSpec;
+	}
+
+	private String getIdColumnName() {
+		return converter.getMappingContext().getRequiredPersistentEntity(entity.getJavaType()).getRequiredIdProperty()
+				.getColumnName();
+	}
+}
--- a/src/main/java/org/springframework/data/r2dbc/repository/support/package-info.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/support/package-info.java
@@ -0,0 +1,7 @@
+/**
+ * Support infrastructure for query derivation of R2DBC-specific repositories.
+ */
+@NonNullApi
+package org.springframework.data.r2dbc.repository.support;
+
+import org.springframework.lang.NonNullApi;