#2 - Add R2dbcRepositoryFactory and simple query subsystem.

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

@@ -37,9 +37,10 @@ import org.springframework.util.ClassUtils;
  */
 public class MappingR2dbcConverter {
 
-	private final MappingContext<JdbcPersistentEntity<?>, JdbcPersistentProperty> mappingContext;
+	private final MappingContext<? extends JdbcPersistentEntity<?>, JdbcPersistentProperty> mappingContext;
 
-	public MappingR2dbcConverter(MappingContext<JdbcPersistentEntity<?>, JdbcPersistentProperty> mappingContext) {
+	public MappingR2dbcConverter(
+			MappingContext<? extends JdbcPersistentEntity<?>, JdbcPersistentProperty> mappingContext) {
 		this.mappingContext = mappingContext;
 	}
 
@@ -97,4 +98,8 @@ public class MappingR2dbcConverter {
 			return object;
 		};
 	}
+
+	public MappingContext<? extends JdbcPersistentEntity<?>, JdbcPersistentProperty> getMappingContext() {
+		return mappingContext;
+	}
 }

src/main/java/org/springframework/data/jdbc/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.jdbc.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> {}

src/main/java/org/springframework/data/jdbc/repository/query/AbstractR2dbcQuery.java

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.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.jdbc.core.function.DatabaseClient;
+import org.springframework.data.jdbc.core.function.DatabaseClient.GenericExecuteSpec;
+import org.springframework.data.jdbc.core.function.FetchSpec;
+import org.springframework.data.jdbc.core.function.MappingR2dbcConverter;
+import org.springframework.data.jdbc.repository.query.R2dbcQueryExecution.ResultProcessingConverter;
+import org.springframework.data.jdbc.repository.query.R2dbcQueryExecution.ResultProcessingExecution;
+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 JdbcParametersParameterAccessor(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(JdbcParameterAccessor 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(JdbcParameterAccessor accessor);
+}

src/main/java/org/springframework/data/jdbc/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.jdbc.repository.query;
+
+import java.util.function.Supplier;
+
+import org.springframework.data.jdbc.core.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);
+}

src/main/java/org/springframework/data/jdbc/repository/query/DtoInstantiatingConverter.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.repository.query;
+
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.convert.EntityInstantiator;
+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.PersistentEntity;
+import org.springframework.data.mapping.PersistentProperty;
+import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.mapping.PreferredConstructor;
+import org.springframework.data.mapping.PreferredConstructor.Parameter;
+import org.springframework.data.mapping.SimplePropertyHandler;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.mapping.model.ParameterValueProvider;
+import org.springframework.util.Assert;
+
+/**
+ * {@link Converter} to instantiate DTOs from fully equipped domain objects.
+ *
+ * @author Mark Paluch
+ */
+class DtoInstantiatingConverter implements Converter<Object, Object> {
+
+	private final Class<?> targetType;
+	private final MappingContext<? extends PersistentEntity<?, ?>, ? extends PersistentProperty<?>> context;
+	private final EntityInstantiator instantiator;
+
+	/**
+	 * Creates a new {@link Converter} to instantiate DTOs.
+	 *
+	 * @param dtoType must not be {@literal null}.
+	 * @param context must not be {@literal null}.
+	 * @param instantiators must not be {@literal null}.
+	 */
+	public DtoInstantiatingConverter(Class<?> dtoType,
+			MappingContext<? extends JdbcPersistentEntity<?>, JdbcPersistentProperty> context,
+			EntityInstantiators instantiator) {
+
+		Assert.notNull(dtoType, "DTO type must not be null!");
+		Assert.notNull(context, "MappingContext must not be null!");
+		Assert.notNull(instantiator, "EntityInstantiators must not be null!");
+
+		this.targetType = dtoType;
+		this.context = context;
+		this.instantiator = instantiator.getInstantiatorFor(context.getRequiredPersistentEntity(dtoType));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.core.convert.converter.Converter#convert(java.lang.Object)
+	 */
+	@Override
+	public Object convert(Object source) {
+
+		if (targetType.isInterface()) {
+			return source;
+		}
+
+		final PersistentEntity<?, ?> sourceEntity = context.getRequiredPersistentEntity(source.getClass());
+		final PersistentPropertyAccessor sourceAccessor = sourceEntity.getPropertyAccessor(source);
+		final PersistentEntity<?, ?> targetEntity = context.getRequiredPersistentEntity(targetType);
+		final PreferredConstructor<?, ? extends PersistentProperty<?>> constructor = targetEntity
+				.getPersistenceConstructor();
+
+		@SuppressWarnings({ "rawtypes", "unchecked" })
+		Object dto = instantiator.createInstance(targetEntity, new ParameterValueProvider() {
+
+			@Override
+			public Object getParameterValue(Parameter parameter) {
+				return sourceAccessor.getProperty(sourceEntity.getPersistentProperty(parameter.getName()));
+			}
+		});
+
+		final PersistentPropertyAccessor dtoAccessor = targetEntity.getPropertyAccessor(dto);
+
+		targetEntity.doWithProperties(new SimplePropertyHandler() {
+
+			@Override
+			public void doWithPersistentProperty(PersistentProperty<?> property) {
+
+				if (constructor.isConstructorParameter(property)) {
+					return;
+				}
+
+				dtoAccessor.setProperty(property,
+						sourceAccessor.getProperty(sourceEntity.getPersistentProperty(property.getName())));
+			}
+		});
+
+		return dto;
+	}
+}

src/main/java/org/springframework/data/jdbc/repository/query/JdbcEntityInformation.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.repository.query;
+
+import org.springframework.data.repository.core.EntityInformation;
+
+/**
+ * JDBC specific {@link EntityInformation}.
+ *
+ * @author Mark Paluch
+ */
+public interface JdbcEntityInformation<T, ID> extends EntityInformation<T, ID> {
+
+	/**
+	 * Returns the name of the table the entity shall be persisted to.
+	 *
+	 * @return
+	 */
+	String getTableName();
+}

src/main/java/org/springframework/data/jdbc/repository/query/JdbcEntityMetadata.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.repository.query;
+
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentEntity;
+import org.springframework.data.repository.core.EntityMetadata;
+
+/**
+ * Extension of {@link EntityMetadata} to additionally expose the collection name an entity shall be persisted to.
+ *
+ * @author Mark Paluch
+ */
+public interface JdbcEntityMetadata<T> extends EntityMetadata<T> {
+
+	/**
+	 * Returns the name of the table the entity shall be persisted to.
+	 *
+	 * @return
+	 */
+	String getTableName();
+
+	/**
+	 * Returns the {@link JdbcPersistentEntity} that supposed to determine the table to be queried.
+	 *
+	 * @return
+	 */
+	JdbcPersistentEntity<?> getTableEntity();
+}

src/main/java/org/springframework/data/jdbc/repository/query/JdbcParameterAccessor.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.repository.query;
+
+import org.springframework.data.repository.query.ParameterAccessor;
+
+/**
+ * JDBC-specific {@link ParameterAccessor}.
+ *
+ * @author Mark Paluch
+ */
+public interface JdbcParameterAccessor extends ParameterAccessor {
+
+	/**
+	 * Returns the raw parameter values of the underlying query method.
+	 */
+	Object[] getValues();
+}

src/main/java/org/springframework/data/jdbc/repository/query/JdbcParameters.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.repository.query;
+
+import java.lang.reflect.Method;
+import java.util.List;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.data.jdbc.repository.query.JdbcParameters.JdbcParameter;
+import org.springframework.data.repository.query.Parameter;
+import org.springframework.data.repository.query.Parameters;
+
+/**
+ * Custom extension of {@link Parameters}.
+ *
+ * @author Mark Paluch
+ */
+public class JdbcParameters extends Parameters<JdbcParameters, JdbcParameter> {
+
+	/**
+	 * Creates a new {@link JdbcParameters} instance from the given {@link Method}.
+	 *
+	 * @param method must not be {@literal null}.
+	 */
+	public JdbcParameters(Method method) {
+		super(method);
+	}
+
+	private JdbcParameters(List<JdbcParameter> parameters) {
+		super(parameters);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.Parameters#createParameter(org.springframework.core.MethodParameter)
+	 */
+	@Override
+	protected JdbcParameter createParameter(MethodParameter parameter) {
+		return new JdbcParameter(parameter);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.query.Parameters#createFrom(java.util.List)
+	 */
+	@Override
+	protected JdbcParameters createFrom(List<JdbcParameter> parameters) {
+		return new JdbcParameters(parameters);
+	}
+
+	/**
+	 * Custom {@link Parameter} implementation.
+	 *
+	 * @author Mark Paluch
+	 */
+	class JdbcParameter extends Parameter {
+
+		/**
+		 * Creates a new {@link JdbcParameter}.
+		 *
+		 * @param parameter must not be {@literal null}.
+		 */
+		JdbcParameter(MethodParameter parameter) {
+			super(parameter);
+		}
+	}
+}

src/main/java/org/springframework/data/jdbc/repository/query/JdbcParametersParameterAccessor.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.repository.query;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.springframework.data.repository.query.ParametersParameterAccessor;
+import org.springframework.data.repository.query.QueryMethod;
+
+/**
+ * JDBC-specific {@link ParametersParameterAccessor}.
+ *
+ * @author Mark Paluch
+ */
+public class JdbcParametersParameterAccessor extends ParametersParameterAccessor implements JdbcParameterAccessor {
+
+	private final List<Object> values;
+
+	/**
+	 * Creates a new {@link JdbcParametersParameterAccessor}.
+	 *
+	 * @param method must not be {@literal null}.
+	 * @param values must not be {@literal null}.
+	 */
+	public JdbcParametersParameterAccessor(QueryMethod method, Object[] values) {
+
+		super(method.getParameters(), values);
+		this.values = Arrays.asList(values);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.repository.query.JdbcParameterAccessor#getValues()
+	 */
+	@Override
+	public Object[] getValues() {
+		return values.toArray();
+	}
+}

src/main/java/org/springframework/data/jdbc/repository/query/R2dbcParameterAccessor.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.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.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 JdbcParametersParameterAccessor {
+
+	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());
+	}
+}

src/main/java/org/springframework/data/jdbc/repository/query/R2dbcQueryExecution.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.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.function.FetchSpec;
+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.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);
+		}
+	}
+}

src/main/java/org/springframework/data/jdbc/repository/query/R2dbcQueryMethod.java

@@ -0,0 +1,227 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.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.JdbcParameters.JdbcParameter;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.projection.ProjectionFactory;
+import org.springframework.data.repository.core.RepositoryMetadata;
+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 JdbcEntityMetadata<?> 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 JdbcParameters createParameters(Method method) {
+		return new JdbcParameters(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 JdbcEntityMetadata<?> getEntityInformation() {
+
+		if (metadata == null) {
+
+			Class<?> returnedObjectType = getReturnedObjectType();
+			Class<?> domainClass = getDomainClass();
+
+			if (ClassUtils.isPrimitiveOrWrapper(returnedObjectType)) {
+
+				this.metadata = new SimpleJdbcEntityMetadata<>((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 SimpleJdbcEntityMetadata<>((Class<Object>) returnedEntity.getType(), tableEntity);
+			}
+		}
+
+		return this.metadata;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.query.QueryMethod#getParameters()
+	 */
+	@Override
+	public JdbcParameters getParameters() {
+		return (JdbcParameters) 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 (JdbcParameter 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();
+	}
+}

src/main/java/org/springframework/data/jdbc/repository/query/SimpleJdbcEntityMetadata.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.repository.query;
+
+import lombok.Getter;
+
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentEntity;
+import org.springframework.util.Assert;
+
+/**
+ * Default implementation of {@link JdbcEntityMetadata}.
+ *
+ * @author Mark Paluch
+ */
+class SimpleJdbcEntityMetadata<T> implements JdbcEntityMetadata<T> {
+
+	private final Class<T> type;
+	private final @Getter JdbcPersistentEntity<?> tableEntity;
+
+	/**
+	 * Creates a new {@link SimpleJdbcEntityMetadata} using the given type and {@link JdbcPersistentEntity} to use for
+	 * table lookups.
+	 *
+	 * @param type must not be {@literal null}.
+	 * @param tableEntity must not be {@literal null}.
+	 */
+	SimpleJdbcEntityMetadata(Class<T> type, JdbcPersistentEntity<?> tableEntity) {
+
+		Assert.notNull(type, "Type must not be null!");
+		Assert.notNull(tableEntity, "Table entity must not be null!");
+
+		this.type = type;
+		this.tableEntity = tableEntity;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.core.EntityMetadata#getJavaType()
+	 */
+	public Class<T> getJavaType() {
+		return type;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.repository.query.JdbcEntityMetadata#getTableName()
+	 */
+	public String getTableName() {
+		return tableEntity.getTableName();
+	}
+}

src/main/java/org/springframework/data/jdbc/repository/query/StringBasedR2dbcQuery.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.repository.query;
+
+import org.springframework.data.jdbc.core.function.DatabaseClient;
+import org.springframework.data.jdbc.core.function.DatabaseClient.BindSpec;
+import org.springframework.data.jdbc.core.function.MappingR2dbcConverter;
+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(JdbcParameterAccessor 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;
+			}
+		};
+	}
+}

src/main/java/org/springframework/data/jdbc/repository/support/MappingJdbcEntityInformation.java

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.repository.support;
+
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentEntity;
+import org.springframework.data.jdbc.repository.query.JdbcEntityInformation;
+import org.springframework.data.repository.core.support.PersistentEntityInformation;
+import org.springframework.lang.Nullable;
+
+import com.sun.corba.se.spi.ior.ObjectId;
+
+/**
+ * {@link JdbcEntityInformation} implementation using a {@link JdbcPersistentEntity} instance to lookup the necessary
+ * information. Can be configured with a custom table name.
+ *
+ * @author Mark Paluch
+ */
+public class MappingJdbcEntityInformation<T, ID> extends PersistentEntityInformation<T, ID>
+		implements JdbcEntityInformation<T, ID> {
+
+	private final JdbcPersistentEntity<T> entityMetadata;
+	private final @Nullable String customTableName;
+	private final Class<ID> fallbackIdType;
+
+	/**
+	 * Creates a new {@link MappingJdbcEntityInformation} for the given {@link JdbcPersistentEntity}.
+	 *
+	 * @param entity must not be {@literal null}.
+	 */
+	public MappingJdbcEntityInformation(JdbcPersistentEntity<T> entity) {
+		this(entity, null, null);
+	}
+
+	/**
+	 * Creates a new {@link MappingJdbcEntityInformation} for the given {@link JdbcPersistentEntity} and fallback
+	 * identifier type.
+	 *
+	 * @param entity must not be {@literal null}.
+	 * @param fallbackIdType can be {@literal null}.
+	 */
+	public MappingJdbcEntityInformation(JdbcPersistentEntity<T> entity, @Nullable Class<ID> fallbackIdType) {
+		this(entity, null, fallbackIdType);
+	}
+
+	/**
+	 * Creates a new {@link MappingJdbcEntityInformation} for the given {@link JdbcPersistentEntity} and custom table
+	 * name.
+	 *
+	 * @param entity must not be {@literal null}.
+	 * @param customTableName can be {@literal null}.
+	 */
+	public MappingJdbcEntityInformation(JdbcPersistentEntity<T> entity, String customTableName) {
+		this(entity, customTableName, null);
+	}
+
+	/**
+	 * Creates a new {@link MappingJdbcEntityInformation} for the given {@link JdbcPersistentEntity}, collection name and
+	 * identifier type.
+	 *
+	 * @param entity must not be {@literal null}.
+	 * @param customTableName can be {@literal null}.
+	 * @param idType can be {@literal null}.
+	 */
+	@SuppressWarnings("unchecked")
+	private MappingJdbcEntityInformation(JdbcPersistentEntity<T> entity, @Nullable String customTableName,
+			@Nullable Class<ID> idType) {
+
+		super(entity);
+
+		this.entityMetadata = entity;
+		this.customTableName = customTableName;
+		this.fallbackIdType = idType != null ? idType : (Class<ID>) ObjectId.class;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.jdbc.repository.query.JdbcEntityInformation#getTableName()
+	 */
+	public String getTableName() {
+		return customTableName == null ? entityMetadata.getTableName() : customTableName;
+	}
+
+	public String getIdAttribute() {
+		return entityMetadata.getRequiredIdProperty().getName();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.core.support.PersistentEntityInformation#getIdType()
+	 */
+	@Override
+	public Class<ID> getIdType() {
+
+		if (this.entityMetadata.hasIdProperty()) {
+			return super.getIdType();
+		}
+
+		return fallbackIdType;
+	}
+}

src/main/java/org/springframework/data/jdbc/repository/support/R2dbcRepositoryFactory.java

@@ -0,0 +1,159 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.repository.support;
+
+import lombok.AccessLevel;
+import lombok.RequiredArgsConstructor;
+
+import java.lang.reflect.Method;
+import java.util.Optional;
+
+import org.springframework.data.jdbc.core.function.DatabaseClient;
+import org.springframework.data.jdbc.core.function.MappingR2dbcConverter;
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentEntity;
+import org.springframework.data.jdbc.core.mapping.JdbcPersistentProperty;
+import org.springframework.data.jdbc.repository.query.JdbcEntityInformation;
+import org.springframework.data.jdbc.repository.query.R2dbcQueryMethod;
+import org.springframework.data.jdbc.repository.query.StringBasedR2dbcQuery;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.projection.ProjectionFactory;
+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 org.springframework.data.jdbc.repository.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) {
+
+		JdbcEntityInformation<?, ?> 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> JdbcEntityInformation<T, ID> getEntityInformation(Class<T> domainClass) {
+		return getEntityInformation(domainClass, null);
+	}
+
+	@SuppressWarnings("unchecked")
+	private <T, ID> JdbcEntityInformation<T, ID> getEntityInformation(Class<T> domainClass,
+			@Nullable RepositoryInformation information) {
+
+		JdbcPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(domainClass);
+
+		return new MappingJdbcEntityInformation<>((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!");
+
+		}
+	}
+}

src/main/java/org/springframework/data/jdbc/repository/support/SimpleR2dbcRepository.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.jdbc.repository.support;
 
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -30,8 +32,7 @@ import org.springframework.data.jdbc.core.function.DatabaseClient.BindSpec;
 import org.springframework.data.jdbc.core.function.DatabaseClient.GenericExecuteSpec;
 import org.springframework.data.jdbc.core.function.FetchSpec;
 import org.springframework.data.jdbc.core.function.MappingR2dbcConverter;
-import org.springframework.data.jdbc.core.mapping.JdbcPersistentEntity;
-import org.springframework.data.mapping.IdentifierAccessor;
+import org.springframework.data.jdbc.repository.query.JdbcEntityInformation;
 import org.springframework.data.repository.reactive.ReactiveCrudRepository;
 import org.springframework.util.Assert;
 
@@ -40,30 +41,12 @@ import org.springframework.util.Assert;
  *
  * @author Mark Paluch
  */
+@RequiredArgsConstructor
 public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, ID> {
 
-	private final DatabaseClient databaseClient;
-	private final MappingR2dbcConverter converter;
-	private final JdbcPersistentEntity<T> entity;
-
-	/**
-	 * Create a new {@link SimpleR2dbcRepository} given {@link DatabaseClient} and {@link JdbcPersistentEntity}.
-	 *
-	 * @param databaseClient must not be {@literal null}.
-	 * @param converter must not be {@literal null}.
-	 * @param entity must not be {@literal null}.
-	 */
-	public SimpleR2dbcRepository(DatabaseClient databaseClient, MappingR2dbcConverter converter,
-			JdbcPersistentEntity<T> entity) {
-		this.converter = converter;
-
-		Assert.notNull(databaseClient, "DatabaseClient must not be null!");
-		Assert.notNull(converter, "MappingR2dbcConverter must not be null!");
-		Assert.notNull(entity, "PersistentEntity must not be null!");
-
-		this.databaseClient = databaseClient;
-		this.entity = entity;
-	}
+	private final @NonNull JdbcEntityInformation<T, ID> entity;
+	private final @NonNull DatabaseClient databaseClient;
+	private final @NonNull MappingR2dbcConverter converter;
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#save(S)
@@ -76,15 +59,14 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 		if (entity.isNew(objectToSave)) {
 
 			return databaseClient.insert() //
-					.into(entity.getType()) //
+					.into(entity.getJavaType()) //
 					.using(objectToSave) //
 					.exchange() //
 					.flatMap(it -> it.extract(converter.populateIdIfNecessary(objectToSave)).one());
 		}
 
 		// TODO: Extract in some kind of SQL generator
-		IdentifierAccessor identifierAccessor = entity.getIdentifierAccessor(objectToSave);
-		Object id = identifierAccessor.getRequiredIdentifier();
+		Object id = entity.getRequiredId(objectToSave);
 
 		Map<String, Optional<Object>> fields = converter.getFieldsToUpdate(objectToSave);
 
@@ -105,7 +87,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 			}
 		}
 
-		return exec.as(entity.getType()) //
+		return exec.as(entity.getJavaType()) //
 				.exchange() //
 				.flatMap(FetchSpec::rowsUpdated) //
 				.thenReturn(objectToSave);
@@ -162,7 +144,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 		return databaseClient.execute()
 				.sql(String.format("SELECT * FROM %s WHERE %s = $1", entity.getTableName(), getIdColumnName())) //
 				.bind("$1", id) //
-				.as(entity.getType()) //
+				.as(entity.getJavaType()) //
 				.fetch() //
 				.one();
 
@@ -206,7 +188,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 	 */
 	@Override
 	public Flux<T> findAll() {
-		return databaseClient.select().from(entity.getType()).fetch().all();
+		return databaseClient.select().from(entity.getJavaType()).fetch().all();
 	}
 
 	/* (non-Javadoc)
@@ -235,7 +217,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 			GenericExecuteSpec exec = databaseClient.execute()
 					.sql(String.format("SELECT * FROM %s WHERE %s IN (%s)", entity.getTableName(), getIdColumnName(), bindings));
 
-			return bind(ids, exec).as(entity.getType()).fetch().all();
+			return bind(ids, exec).as(entity.getJavaType()).fetch().all();
 		});
 	}
 
@@ -284,7 +266,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 			GenericExecuteSpec exec = databaseClient.execute()
 					.sql(String.format("DELETE FROM %s WHERE %s IN (%s)", entity.getTableName(), getIdColumnName(), bindings));
 
-			return bind(ids, exec).as(entity.getType()).fetch().rowsUpdated();
+			return bind(ids, exec).as(entity.getJavaType()).fetch().rowsUpdated();
 		}).then();
 	}
 
@@ -297,9 +279,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 
 		Assert.notNull(objectToDelete, "Object to delete must not be null!");
 
-		IdentifierAccessor identifierAccessor = entity.getIdentifierAccessor(objectToDelete);
-
-		return deleteById((ID) identifierAccessor.getRequiredIdentifier());
+		return deleteById(entity.getRequiredId(objectToDelete));
 	}
 
 	/* (non-Javadoc)
@@ -323,8 +303,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 		Assert.notNull(objectPublisher, "The Object Publisher must not be null!");
 
 		Flux<ID> idPublisher = Flux.from(objectPublisher) //
-				.map(entity::getIdentifierAccessor) //
-				.map(identifierAccessor -> (ID) identifierAccessor.getRequiredIdentifier());
+				.map(entity::getRequiredId);
 
 		return deleteById(idPublisher);
 	}
@@ -355,6 +334,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 	}
 
 	private String getIdColumnName() {
-		return entity.getRequiredIdProperty().getColumnName();
+		return converter.getMappingContext().getRequiredPersistentEntity(entity.getJavaType()).getRequiredIdProperty()
+				.getColumnName();
 	}
 }