DATAMONGO-2182 - Add Querydsl support for reactive repositories.

We now support execution of Querydsl Predicates using reactive MongoDB repositories. Reactive repositories are only required to add ReactiveQuerydslPredicateExecutor to their declaration to enable Querydsl.

public interface PersonRepository extends ReactiveMongoRepository<Person, String>, ReactiveQuerydslPredicateExecutor<Person> {

   // additional query methods go here
}

PersonRepository repository = …;

QPerson person = new QPerson("person");

Flux<Person> result = repository.findAll(person.address.zipCode.eq("C0123"));

Original Pull Request: #635

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/support/ReactiveMongoRepositoryFactory.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.mongodb.repository.support;
 
+import static org.springframework.data.querydsl.QuerydslUtils.*;
+
 import lombok.AccessLevel;
 import lombok.RequiredArgsConstructor;
 
@@ -32,10 +34,13 @@ import org.springframework.data.mongodb.repository.query.ReactiveMongoQueryMetho
 import org.springframework.data.mongodb.repository.query.ReactivePartTreeMongoQuery;
 import org.springframework.data.mongodb.repository.query.ReactiveStringBasedMongoQuery;
 import org.springframework.data.projection.ProjectionFactory;
+import org.springframework.data.querydsl.ReactiveQuerydslPredicateExecutor;
 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.core.support.RepositoryComposition.RepositoryFragments;
+import org.springframework.data.repository.core.support.RepositoryFragment;
 import org.springframework.data.repository.query.QueryLookupStrategy;
 import org.springframework.data.repository.query.QueryLookupStrategy.Key;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
@@ -81,6 +86,30 @@ public class ReactiveMongoRepositoryFactory extends ReactiveRepositoryFactorySup
 		return SimpleReactiveMongoRepository.class;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getRepositoryFragments(org.springframework.data.repository.core.RepositoryMetadata)
+	 */
+	@Override
+	protected RepositoryFragments getRepositoryFragments(RepositoryMetadata metadata) {
+
+		RepositoryFragments fragments = RepositoryFragments.empty();
+
+		boolean isQueryDslRepository = QUERY_DSL_PRESENT
+				&& ReactiveQuerydslPredicateExecutor.class.isAssignableFrom(metadata.getRepositoryInterface());
+
+		if (isQueryDslRepository) {
+
+			MongoEntityInformation<?, Serializable> entityInformation = getEntityInformation(metadata.getDomainType(),
+					metadata);
+
+			fragments = fragments.append(RepositoryFragment.implemented(getTargetRepositoryViaReflection(
+					ReactiveQuerydslMongoPredicateExecutor.class, entityInformation, operations)));
+		}
+
+		return fragments;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getTargetRepository(org.springframework.data.repository.core.RepositoryInformation)
@@ -113,12 +142,12 @@ public class ReactiveMongoRepositoryFactory extends ReactiveRepositoryFactorySup
 
 	@SuppressWarnings("unchecked")
 	private <T, ID> MongoEntityInformation<T, ID> getEntityInformation(Class<T> domainClass,
-			@Nullable RepositoryInformation information) {
+			@Nullable RepositoryMetadata metadata) {
 
 		MongoPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(domainClass);
 
-		return new MappingMongoEntityInformation<T, ID>((MongoPersistentEntity<T>) entity,
-				information != null ? (Class<ID>) information.getIdType() : null);
+		return new MappingMongoEntityInformation<>((MongoPersistentEntity<T>) entity,
+				metadata != null ? (Class<ID>) metadata.getIdType() : null);
 	}
 
 	/**

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/support/ReactiveQuerydslMongoPredicateExecutor.java

@@ -0,0 +1,232 @@
+/*
+ * Copyright 2019 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.mongodb.repository.support;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.List;
+
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Order;
+import org.springframework.data.mongodb.core.ReactiveMongoOperations;
+import org.springframework.data.mongodb.repository.query.MongoEntityInformation;
+import org.springframework.data.querydsl.EntityPathResolver;
+import org.springframework.data.querydsl.QSort;
+import org.springframework.data.querydsl.QuerydslPredicateExecutor;
+import org.springframework.data.querydsl.ReactiveQuerydslPredicateExecutor;
+import org.springframework.data.querydsl.SimpleEntityPathResolver;
+import org.springframework.data.repository.core.EntityInformation;
+import org.springframework.util.Assert;
+
+import com.querydsl.core.types.EntityPath;
+import com.querydsl.core.types.Expression;
+import com.querydsl.core.types.OrderSpecifier;
+import com.querydsl.core.types.Predicate;
+import com.querydsl.core.types.dsl.PathBuilder;
+
+/**
+ * MongoDB-specific {@link QuerydslPredicateExecutor} that allows execution {@link Predicate}s in various forms.
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ */
+public class ReactiveQuerydslMongoPredicateExecutor<T> implements ReactiveQuerydslPredicateExecutor<T> {
+
+	private final PathBuilder<T> builder;
+	private final EntityInformation<T, ?> entityInformation;
+	private final ReactiveMongoOperations mongoOperations;
+
+	/**
+	 * Creates a new {@link ReactiveQuerydslMongoPredicateExecutor} for the given {@link MongoEntityInformation} and
+	 * {@link ReactiveMongoOperations}. Uses the {@link SimpleEntityPathResolver} to create an {@link EntityPath} for the
+	 * given domain class.
+	 *
+	 * @param entityInformation must not be {@literal null}.
+	 * @param mongoOperations must not be {@literal null}.
+	 */
+	public ReactiveQuerydslMongoPredicateExecutor(MongoEntityInformation<T, ?> entityInformation,
+			ReactiveMongoOperations mongoOperations) {
+		this(entityInformation, mongoOperations, SimpleEntityPathResolver.INSTANCE);
+	}
+
+	/**
+	 * Creates a new {@link ReactiveQuerydslMongoPredicateExecutor} for the given {@link MongoEntityInformation},
+	 * {@link ReactiveMongoOperations} and {@link EntityPathResolver}.
+	 *
+	 * @param entityInformation must not be {@literal null}.
+	 * @param mongoOperations must not be {@literal null}.
+	 * @param resolver must not be {@literal null}.
+	 */
+	public ReactiveQuerydslMongoPredicateExecutor(MongoEntityInformation<T, ?> entityInformation,
+			ReactiveMongoOperations mongoOperations, EntityPathResolver resolver) {
+
+		Assert.notNull(resolver, "EntityPathResolver must not be null!");
+
+		EntityPath<T> path = resolver.createPath(entityInformation.getJavaType());
+
+		this.builder = new PathBuilder<T>(path.getType(), path.getMetadata());
+		this.entityInformation = entityInformation;
+		this.mongoOperations = mongoOperations;
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.querydsl.ReactiveQuerydslPredicateExecutor#findOne(com.querydsl.core.types.Predicate)
+	 */
+	@Override
+	public Mono<T> findOne(Predicate predicate) {
+
+		Assert.notNull(predicate, "Predicate must not be null!");
+
+		return createQueryFor(predicate).fetchOne();
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.querydsl.ReactiveQuerydslPredicateExecutor#findAll(com.querydsl.core.types.Predicate)
+	 */
+	@Override
+	public Flux<T> findAll(Predicate predicate) {
+
+		Assert.notNull(predicate, "Predicate must not be null!");
+
+		return createQueryFor(predicate).fetch();
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.querydsl.ReactiveQuerydslPredicateExecutor#findAll(com.querydsl.core.types.Predicate, com.querydsl.core.types.OrderSpecifier[])
+	 */
+	@Override
+	public Flux<T> findAll(Predicate predicate, OrderSpecifier<?>... orders) {
+
+		Assert.notNull(predicate, "Predicate must not be null!");
+		Assert.notNull(orders, "Order specifiers must not be null!");
+
+		return createQueryFor(predicate).orderBy(orders).fetch();
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.querydsl.ReactiveQuerydslPredicateExecutor#findAll(com.querydsl.core.types.Predicate, org.springframework.data.domain.Sort)
+	 */
+	@Override
+	public Flux<T> findAll(Predicate predicate, Sort sort) {
+
+		Assert.notNull(predicate, "Predicate must not be null!");
+		Assert.notNull(sort, "Sort must not be null!");
+
+		return applySorting(createQueryFor(predicate), sort).fetch();
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.querydsl.ReactiveQuerydslPredicateExecutor#findAll(com.querydsl.core.types.OrderSpecifier[])
+	 */
+	@Override
+	public Flux<T> findAll(OrderSpecifier<?>... orders) {
+
+		Assert.notNull(orders, "Order specifiers must not be null!");
+
+		return createQuery().orderBy(orders).fetch();
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.querydsl.ReactiveQuerydslPredicateExecutor#count(com.querydsl.core.types.Predicate)
+	 */
+	@Override
+	public Mono<Long> count(Predicate predicate) {
+
+		Assert.notNull(predicate, "Predicate must not be null!");
+
+		return createQueryFor(predicate).fetchCount();
+	}
+
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.querydsl.ReactiveQuerydslPredicateExecutor#exists(com.querydsl.core.types.Predicate)
+	 */
+	@Override
+	public Mono<Boolean> exists(Predicate predicate) {
+
+		Assert.notNull(predicate, "Predicate must not be null!");
+
+		return createQueryFor(predicate).fetchCount().map(it -> it != 0);
+	}
+
+	/**
+	 * Creates a {@link ReactiveSpringDataMongodbQuery} for the given {@link Predicate}.
+	 *
+	 * @param predicate
+	 * @return
+	 */
+	private ReactiveSpringDataMongodbQuery<T> createQueryFor(Predicate predicate) {
+		return createQuery().where(predicate);
+	}
+
+	/**
+	 * Creates a {@link ReactiveSpringDataMongodbQuery}.
+	 *
+	 * @return
+	 */
+	private ReactiveSpringDataMongodbQuery<T> createQuery() {
+		SpringDataMongodbSerializer serializer = new SpringDataMongodbSerializer(mongoOperations.getConverter());
+
+		Class<T> javaType = entityInformation.getJavaType();
+		return new ReactiveSpringDataMongodbQuery<>(serializer, mongoOperations, javaType,
+				mongoOperations.getCollectionName(javaType));
+	}
+
+	/**
+	 * Applies the given {@link Sort} to the given {@link ReactiveSpringDataMongodbQuery}.
+	 *
+	 * @param query
+	 * @param sort
+	 * @return
+	 */
+	private ReactiveSpringDataMongodbQuery<T> applySorting(ReactiveSpringDataMongodbQuery<T> query, Sort sort) {
+
+		// TODO: find better solution than instanceof check
+		if (sort instanceof QSort) {
+
+			List<OrderSpecifier<?>> orderSpecifiers = ((QSort) sort).getOrderSpecifiers();
+			query.orderBy(orderSpecifiers.toArray(new OrderSpecifier<?>[orderSpecifiers.size()]));
+
+			return query;
+		}
+
+		sort.stream().map(this::toOrder).forEach(query::orderBy);
+
+		return query;
+	}
+
+	/**
+	 * Transforms a plain {@link Order} into a Querydsl specific {@link OrderSpecifier}.
+	 *
+	 * @param order
+	 * @return
+	 */
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	private OrderSpecifier<?> toOrder(Order order) {
+
+		Expression<Object> property = builder.get(order.getProperty());
+
+		return new OrderSpecifier(
+				order.isAscending() ? com.querydsl.core.types.Order.ASC : com.querydsl.core.types.Order.DESC, property);
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/support/ReactiveSpringDataMongodbQuery.java

@@ -0,0 +1,282 @@
+/*
+ * Copyright 2019 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.mongodb.repository.support;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.data.mongodb.core.ReactiveFindOperation.FindWithProjection;
+import org.springframework.data.mongodb.core.ReactiveMongoOperations;
+import org.springframework.data.mongodb.core.query.BasicQuery;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.lang.Nullable;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+
+import com.querydsl.core.JoinExpression;
+import com.querydsl.core.QueryMetadata;
+import com.querydsl.core.QueryModifiers;
+import com.querydsl.core.types.Expression;
+import com.querydsl.core.types.ExpressionUtils;
+import com.querydsl.core.types.Operation;
+import com.querydsl.core.types.OrderSpecifier;
+import com.querydsl.core.types.Path;
+import com.querydsl.core.types.Predicate;
+import com.querydsl.core.types.dsl.CollectionPathBase;
+
+/**
+ * MongoDB query with utilizing {@link ReactiveMongoOperations} for command execution.
+ *
+ * @param <K> result type
+ * @param <Q> concrete subtype
+ * @author Mark Paluch
+ * @since 2.2
+ */
+class ReactiveSpringDataMongodbQuery<K> extends QuerydslAbstractMongodbQuery<K, ReactiveSpringDataMongodbQuery<K>> {
+
+	private final Class<K> entityClass;
+	private final ReactiveMongoOperations mongoOperations;
+	private final FindWithProjection<K> find;
+
+	ReactiveSpringDataMongodbQuery(ReactiveMongoOperations mongoOperations, Class<? extends K> entityClass) {
+
+		super(new SpringDataMongodbSerializer(mongoOperations.getConverter()));
+
+		this.entityClass = (Class<K>) entityClass;
+		this.mongoOperations = mongoOperations;
+		this.find = mongoOperations.query(this.entityClass);
+	}
+
+	ReactiveSpringDataMongodbQuery(MongodbDocumentSerializer serializer, ReactiveMongoOperations mongoOperations,
+			Class<? extends K> entityClass, String collection) {
+
+		super(serializer);
+
+		this.entityClass = (Class<K>) entityClass;
+		this.mongoOperations = mongoOperations;
+		this.find = mongoOperations.query(this.entityClass).inCollection(collection);
+	}
+
+	/**
+	 * Fetch all matching query results.
+	 *
+	 * @return {@link Flux} emitting all query results or {@link Flux#empty()} if there are none.
+	 */
+	public Flux<K> fetch() {
+		return createQuery().flatMapMany(it -> find.matching(it).all());
+	}
+
+	/**
+	 * Fetch the first matching query result.
+	 *
+	 * @return {@link Mono} emitting the first query result or {@link Mono#empty()} if there are none.
+	 * @throws org.springframework.dao.IncorrectResultSizeDataAccessException if more than one match found.
+	 */
+	public Mono<K> fetchOne() {
+		return createQuery().flatMap(it -> find.matching(it).one());
+	}
+
+	/**
+	 * Fetch the count of matching query results.
+	 *
+	 * @return {@link Mono} emitting the first query result count. Emits always a count even item.
+	 */
+	public Mono<Long> fetchCount() {
+		return createQuery().flatMap(it -> find.matching(it).count());
+	}
+
+	/**
+	 * Define a join.
+	 *
+	 * @param ref reference
+	 * @param target join target
+	 * @return new instance of {@link QuerydslJoinBuilder}.
+	 */
+	public <T> QuerydslJoinBuilder<ReactiveSpringDataMongodbQuery<K>, K, T> join(Path<T> ref, Path<T> target) {
+		return new QuerydslJoinBuilder<>(getQueryMixin(), ref, target);
+	}
+
+	/**
+	 * Define a join.
+	 *
+	 * @param ref reference
+	 * @param target join target
+	 * @return new instance of {@link QuerydslJoinBuilder}.
+	 */
+	public <T> QuerydslJoinBuilder<ReactiveSpringDataMongodbQuery<K>, K, T> join(CollectionPathBase<?, T, ?> ref,
+			Path<T> target) {
+		return new QuerydslJoinBuilder<>(getQueryMixin(), ref, target);
+	}
+
+	/**
+	 * Define a constraint for an embedded object.
+	 *
+	 * @param collection collection must not be {@literal null}.
+	 * @param target target must not be {@literal null}.
+	 * @return new instance of {@link QuerydslAnyEmbeddedBuilder}.
+	 */
+	public <T> QuerydslAnyEmbeddedBuilder<ReactiveSpringDataMongodbQuery<K>, K> anyEmbedded(
+			Path<? extends Collection<T>> collection, Path<T> target) {
+		return new QuerydslAnyEmbeddedBuilder<>(getQueryMixin(), collection);
+	}
+
+	protected Mono<Query> createQuery() {
+
+		QueryMetadata metadata = getQueryMixin().getMetadata();
+
+		return createQuery(createFilter(metadata), metadata.getProjection(), metadata.getModifiers(),
+				metadata.getOrderBy());
+	}
+
+	/**
+	 * Creates a MongoDB query that is emitted through a {@link Mono} given {@link Mono} of {@link Predicate}.
+	 *
+	 * @param filter must not be {@literal null}.
+	 * @param projection can be {@literal null} if no projection is given. Query requests all fields in such case.
+	 * @param modifiers must not be {@literal null}.
+	 * @param orderBy must not be {@literal null}.
+	 * @return {@link Mono} emitting the {@link Query}.
+	 */
+	protected Mono<Query> createQuery(Mono<Predicate> filter, @Nullable Expression<?> projection,
+			QueryModifiers modifiers, List<OrderSpecifier<?>> orderBy) {
+
+		return filter.map(this::createQuery) //
+				.defaultIfEmpty(createQuery(null)) //
+				.map(it -> {
+
+					BasicQuery basicQuery = new BasicQuery(it, createProjection(projection));
+
+					Integer limit = modifiers.getLimitAsInteger();
+					Integer offset = modifiers.getOffsetAsInteger();
+
+					if (limit != null) {
+						basicQuery.limit(limit);
+					}
+					if (offset != null) {
+						basicQuery.skip(offset);
+					}
+					if (orderBy.size() > 0) {
+						basicQuery.setSortObject(createSort(orderBy));
+					}
+
+					return basicQuery;
+				});
+	}
+
+	protected Mono<Predicate> createFilter(QueryMetadata metadata) {
+
+		if (!metadata.getJoins().isEmpty()) {
+
+			return createJoinFilter(metadata).map(it -> ExpressionUtils.allOf(metadata.getWhere(), it))
+					.switchIfEmpty(Mono.justOrEmpty(metadata.getWhere()));
+		}
+
+		return Mono.justOrEmpty(metadata.getWhere());
+	}
+
+	/**
+	 * Creates a Join filter by querying {@link com.mongodb.DBRef references}.
+	 *
+	 * @param metadata
+	 * @return
+	 */
+	@SuppressWarnings("unchecked")
+	protected Mono<Predicate> createJoinFilter(QueryMetadata metadata) {
+
+		MultiValueMap<Expression<?>, Mono<Predicate>> predicates = new LinkedMultiValueMap<>();
+		List<JoinExpression> joins = metadata.getJoins();
+
+		for (int i = joins.size() - 1; i >= 0; i--) {
+
+			JoinExpression join = joins.get(i);
+			Path<?> source = (Path) ((Operation<?>) join.getTarget()).getArg(0);
+			Path<?> target = (Path) ((Operation<?>) join.getTarget()).getArg(1);
+			Collection<Mono<Predicate>> extraFilters = predicates.get(target.getRoot());
+
+			Mono<Predicate> filter = allOf(extraFilters).map(it -> ExpressionUtils.allOf(join.getCondition(), it))
+					.switchIfEmpty(Mono.justOrEmpty(join.getCondition()));
+
+			Mono<Predicate> predicate = getIds(target.getType(), filter) //
+					.collectList() //
+					.handle((it, sink) -> {
+
+						if (it.isEmpty()) {
+							sink.error(new NoMatchException(source));
+							return;
+						}
+
+						Path<?> path = ExpressionUtils.path(String.class, source, "$id");
+						sink.next(ExpressionUtils.in((Path<Object>) path, it));
+					});
+
+			predicates.add(source.getRoot(), predicate);
+		}
+
+		Path<?> source = (Path) ((Operation) joins.get(0).getTarget()).getArg(0);
+		return allOf(predicates.get(source.getRoot())).onErrorResume(NoMatchException.class,
+				e -> Mono.just(ExpressionUtils.predicate(QuerydslMongoOps.NO_MATCH, e.source)));
+	}
+
+	private Mono<Predicate> allOf(@Nullable Collection<Mono<Predicate>> predicates) {
+		return predicates != null ? Flux.concat(predicates).collectList().map(ExpressionUtils::allOf) : Mono.empty();
+	}
+
+	/**
+	 * Fetch the list of ids matching a given condition.
+	 *
+	 * @param targetType must not be {@literal null}.
+	 * @param condition must not be {@literal null}.
+	 * @return empty {@link List} if none found.
+	 */
+	protected Flux<Object> getIds(Class<?> targetType, Mono<Predicate> condition) {
+
+		return condition.flatMapMany(it -> getIds(targetType, it))
+				.switchIfEmpty(Flux.defer(() -> getIds(targetType, (Predicate) null)));
+	}
+
+	/**
+	 * Fetch the list of ids matching a given condition.
+	 *
+	 * @param targetType must not be {@literal null}.
+	 * @param condition must not be {@literal null}.
+	 * @return empty {@link List} if none found.
+	 */
+	protected Flux<Object> getIds(Class<?> targetType, @Nullable Predicate condition) {
+		return createQuery(Mono.justOrEmpty(condition), null, QueryModifiers.EMPTY, Collections.emptyList())
+				.flatMapMany(query -> mongoOperations.findDistinct(query, "_id", targetType, Object.class));
+	}
+
+	/**
+	 * Marker exception to indicate no matches for a query using reference Id's.
+	 */
+	static class NoMatchException extends RuntimeException {
+
+		final Path<?> source;
+
+		public NoMatchException(Path<?> source) {
+			this.source = source;
+		}
+
+		@Override
+		public synchronized Throwable fillInStackTrace() {
+			return null;
+		}
+	}
+}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/ReactiveMongoRepositoryTests.java

@@ -56,6 +56,7 @@ import org.springframework.data.mongodb.core.mapping.Document;
 import org.springframework.data.mongodb.repository.Person.Sex;
 import org.springframework.data.mongodb.repository.support.ReactiveMongoRepositoryFactory;
 import org.springframework.data.mongodb.repository.support.SimpleReactiveMongoRepository;
+import org.springframework.data.querydsl.ReactiveQuerydslPredicateExecutor;
 import org.springframework.data.repository.Repository;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
 import org.springframework.test.context.ContextConfiguration;
@@ -82,6 +83,7 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF
 	ReactiveCappedCollectionRepository cappedRepository;
 
 	Person dave, oliver, carter, boyd, stefan, leroi, alicia;
+	QPerson person = new QPerson("person");
 
 	@Override
 	public void setBeanClassLoader(ClassLoader classLoader) {
@@ -376,7 +378,19 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF
 				.verifyComplete();
 	}
 
-	interface ReactivePersonRepository extends ReactiveMongoRepository<Person, String> {
+	@Test // DATAMONGO-2182
+	public void shouldFindPersonsWhenUsingQueryDslPerdicatedOnIdProperty() {
+
+		repository.findAll(person.id.in(Arrays.asList(dave.id, carter.id))) //
+				.collectList() //
+				.as(StepVerifier::create) //
+				.assertNext(actual -> {
+					assertThat(actual).containsExactlyInAnyOrder(dave, carter);
+				}).verifyComplete();
+	}
+
+	interface ReactivePersonRepository
+			extends ReactiveMongoRepository<Person, String>, ReactiveQuerydslPredicateExecutor<Person> {
 
 		Flux<Person> findByLastname(String lastname);
 

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/support/ReactiveQuerydslMongoPredicateExecutorIntegrationTests.java

@@ -0,0 +1,278 @@
+/*
+ * Copyright 2019 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.mongodb.repository.support;
+
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.dao.PermissionDeniedDataAccessException;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Direction;
+import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
+import org.springframework.data.mongodb.core.ReactiveMongoOperations;
+import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
+import org.springframework.data.mongodb.repository.Address;
+import org.springframework.data.mongodb.repository.Person;
+import org.springframework.data.mongodb.repository.QAddress;
+import org.springframework.data.mongodb.repository.QPerson;
+import org.springframework.data.mongodb.repository.QUser;
+import org.springframework.data.mongodb.repository.User;
+import org.springframework.data.mongodb.repository.query.MongoEntityInformation;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import com.mongodb.MongoException;
+import com.mongodb.reactivestreams.client.MongoDatabase;
+
+/**
+ * Tests for {@link ReactiveQuerydslMongoPredicateExecutor}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(SpringRunner.class)
+@ContextConfiguration("classpath:reactive-infrastructure.xml")
+public class ReactiveQuerydslMongoPredicateExecutorIntegrationTests {
+
+	@Autowired ReactiveMongoOperations operations;
+	@Autowired ReactiveMongoDatabaseFactory dbFactory;
+
+	ReactiveQuerydslMongoPredicateExecutor<Person> repository;
+
+	Person dave, oliver, carter;
+	QPerson person;
+
+	@Before
+	public void setup() {
+
+		ReactiveMongoRepositoryFactory factory = new ReactiveMongoRepositoryFactory(operations);
+		MongoEntityInformation<Person, String> entityInformation = factory.getEntityInformation(Person.class);
+		repository = new ReactiveQuerydslMongoPredicateExecutor<>(entityInformation, operations);
+
+		operations.dropCollection(Person.class) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		dave = new Person("Dave", "Matthews", 42);
+		oliver = new Person("Oliver August", "Matthews", 4);
+		carter = new Person("Carter", "Beauford", 49);
+
+		person = new QPerson("person");
+
+		operations.insertAll(Arrays.asList(oliver, dave, carter)).as(StepVerifier::create) //
+				.expectNextCount(3) //
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-2182
+	public void shouldSupportExistsWithPredicate() {
+
+		repository.exists(person.firstname.eq("Dave")) //
+				.as(StepVerifier::create) //
+				.expectNext(true) //
+				.verifyComplete();
+
+		repository.exists(person.firstname.eq("Unknown")) //
+				.as(StepVerifier::create) //
+				.expectNext(false) //
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-2182
+	public void shouldSupportFindAllWithPredicateAndSort() {
+
+		repository.findAll(person.lastname.isNotNull(), Sort.by(Direction.ASC, "firstname")) //
+				.as(StepVerifier::create) //
+				.expectNext(carter, dave, oliver) //
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-2182
+	public void findOneWithPredicateReturnsResultCorrectly() {
+
+		repository.findOne(person.firstname.eq(dave.getFirstname())) //
+				.as(StepVerifier::create) //
+				.expectNext(dave) //
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-2182
+	public void findOneWithPredicateReturnsEmptyWhenNoDataFound() {
+
+		repository.findOne(person.firstname.eq("batman")) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-2182
+	public void findOneWithPredicateThrowsExceptionForNonUniqueResults() {
+
+		repository.findOne(person.firstname.contains("e")) //
+				.as(StepVerifier::create) //
+				.expectError(IncorrectResultSizeDataAccessException.class) //
+				.verify();
+	}
+
+	@Test // DATAMONGO-2182
+	public void findUsingAndShouldWork() {
+
+		repository
+				.findAll(person.lastname.startsWith(oliver.getLastname()).and(person.firstname.startsWith(dave.getFirstname()))) //
+				.as(StepVerifier::create) //
+				.expectNext(dave) //
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-2182
+	public void queryShouldTerminateWithUnsupportedOperationWithJoinOnDBref() {
+
+		User user1 = new User();
+		user1.setUsername("user-1");
+
+		User user2 = new User();
+		user2.setUsername("user-2");
+
+		User user3 = new User();
+		user3.setUsername("user-3");
+
+		operations.insertAll(Arrays.asList(user1, user2, user3)) //
+				.as(StepVerifier::create) //
+				.expectNextCount(3) //
+				.verifyComplete();
+
+		Person person1 = new Person("Max", "The Mighty");
+		person1.setCoworker(user1);
+
+		Person person2 = new Person("Jack", "The Ripper");
+		person2.setCoworker(user2);
+
+		Person person3 = new Person("Bob", "The Builder");
+		person3.setCoworker(user3);
+
+		operations.save(person1) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+		operations.save(person2)//
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+		operations.save(person3) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		Flux<Person> result = new ReactiveSpringDataMongodbQuery<>(operations, Person.class).where()
+				.join(person.coworker, QUser.user).on(QUser.user.username.eq("user-2")).fetch();
+
+		result.as(StepVerifier::create) //
+				.expectError(UnsupportedOperationException.class) //
+				.verify();
+	}
+
+	@Test // DATAMONGO-2182
+	public void queryShouldTerminateWithUnsupportedOperationOnJoinWithNoResults() {
+
+		User user1 = new User();
+		user1.setUsername("user-1");
+
+		User user2 = new User();
+		user2.setUsername("user-2");
+
+		operations.insertAll(Arrays.asList(user1, user2)) //
+				.as(StepVerifier::create) //
+				.expectNextCount(2) //
+				.verifyComplete();
+
+		Person person1 = new Person("Max", "The Mighty");
+		person1.setCoworker(user1);
+
+		Person person2 = new Person("Jack", "The Ripper");
+		person2.setCoworker(user2);
+
+		operations.save(person1) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+		;
+		operations.save(person2) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+		;
+
+		Flux<Person> result = new ReactiveSpringDataMongodbQuery<>(operations, Person.class).where()
+				.join(person.coworker, QUser.user).on(QUser.user.username.eq("does-not-exist")).fetch();
+
+		result.as(StepVerifier::create) //
+				.expectError(UnsupportedOperationException.class) //
+				.verify();
+	}
+
+	@Test // DATAMONGO-2182
+	public void springDataMongodbQueryShouldAllowElemMatchOnArrays() {
+
+		Address adr1 = new Address("Hauptplatz", "4020", "Linz");
+		Address adr2 = new Address("Stephansplatz", "1010", "Wien");
+		Address adr3 = new Address("Tower of London", "EC3N 4AB", "London");
+
+		Person person1 = new Person("Max", "The Mighty");
+		person1.setShippingAddresses(new LinkedHashSet<>(Arrays.asList(adr1, adr2)));
+
+		Person person2 = new Person("Jack", "The Ripper");
+		person2.setShippingAddresses(new LinkedHashSet<>(Arrays.asList(adr2, adr3)));
+
+		operations.insertAll(Arrays.asList(person1, person2)) //
+				.as(StepVerifier::create) //
+				.expectNextCount(2) //
+				.verifyComplete();
+
+		Flux<Person> result = new ReactiveSpringDataMongodbQuery<>(operations, Person.class).where()
+				.anyEmbedded(person.shippingAddresses, QAddress.address).on(QAddress.address.city.eq("London")).fetch();
+
+		result.as(StepVerifier::create) //
+				.expectNext(person2) //
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-2182
+	public void translatesExceptionsCorrectly() {
+
+		ReactiveMongoOperations ops = new ReactiveMongoTemplate(dbFactory) {
+
+			@Override
+			protected MongoDatabase doGetDatabase() {
+				throw new MongoException(18, "Authentication Failed");
+			}
+		};
+
+		ReactiveMongoRepositoryFactory factory = new ReactiveMongoRepositoryFactory(ops);
+		MongoEntityInformation<Person, String> entityInformation = factory.getEntityInformation(Person.class);
+		repository = new ReactiveQuerydslMongoPredicateExecutor<>(entityInformation, ops);
+
+		repository.findOne(person.firstname.contains("batman")) //
+				.as(StepVerifier::create) //
+				.expectError(PermissionDeniedDataAccessException.class) //
+				.verify();
+	}
+}

src/main/asciidoc/new-features.adoc

@@ -4,6 +4,7 @@
 [[new-features.2-2-0]]
 == What's New in Spring Data MongoDB 2.2
 * <<mongo.query.kotlin-support,Type-safe Queries for Kotlin>>
+* <<mongodb.reactive.repositories.queries.type-safe,Querydsl support for reactive repositories>> via `ReactiveQuerydslPredicateExecutor`.
 
 [[new-features.2-1-0]]
 == What's New in Spring Data MongoDB 2.1

src/main/asciidoc/reference/reactive-mongo-repositories.adoc

@@ -129,6 +129,7 @@ It supports the following features:
 * <<mongodb.repositories.queries.delete>>
 * <<mongodb.repositories.queries.json-based>>
 * <<mongodb.repositories.queries.full-text>>
+* <<mongodb.reactive.repositories.queries.type-safe>>
 * <<projections>>
 
 WARNING: Reactive Repositories do not support type-safe query methods that use `Querydsl`.
@@ -198,3 +199,61 @@ public interface PersonRepository extends ReactiveMongoRepository<Person, String
   Flux<GeoResult<Person>> findByLocationNear(Point location);
 }
 ----
+
+[[mongodb.reactive.repositories.queries.type-safe]]
+=== Type-safe Query Methods
+
+Reactive MongoDB repository support integrates with the http://www.querydsl.com/[Querydsl] project, which provides a way to perform type-safe queries. To quote from the project description, "Instead of writing queries as inline strings or externalizing them into XML files they are constructed via a fluent API." It provides the following features:
+
+* Code completion in the IDE (all properties, methods, and operations can be expanded in your favorite Java IDE).
+* Almost no syntactically invalid queries allowed (type-safe on all levels).
+* Domain types and properties can be referenced safely -- no strings involved!
+* Adapts better to refactoring changes in domain types.
+* Incremental query definition is easier.
+
+See the http://www.querydsl.com/static/querydsl/latest/reference/html/[QueryDSL documentation] for how to bootstrap your environment for APT-based code generation using Maven or Ant.
+
+QueryDSL lets you write queries such as the following:
+
+[source,java]
+----
+QPerson person = new QPerson("person");
+
+Flux<Person> result = repository.findAll(person.address.zipCode.eq("C0123"));
+----
+
+`QPerson` is a class that is generated by the Java annotation post-processing tool. It is a `Predicate` that lets you write type-safe queries. Notice that there are no strings in the query other than the `C0123` value.
+
+You can use the generated `Predicate` class by using the `ReactiveQuerydslPredicateExecutor` interface, which the following listing shows:
+
+[source,java]
+----
+public interface ReactiveQuerydslPredicateExecutor<T> {
+
+	Mono<T> findOne(Predicate predicate);
+
+	Flux<T> findAll(Predicate predicate);
+
+	Flux<T> findAll(Predicate predicate, Sort sort);
+
+	Flux<T> findAll(Predicate predicate, OrderSpecifier<?>... orders);
+
+	Flux<T> findAll(OrderSpecifier<?>... orders);
+
+	Mono<Long> count(Predicate predicate);
+
+	Mono<Boolean> exists(Predicate predicate);
+}
+----
+
+To use this in your repository implementation, add it to the list of repository interfaces from which your interface inherits, as the following example shows:
+
+[source,java]
+----
+public interface PersonRepository extends ReactiveMongoRepository<Person, String>, ReactiveQuerydslPredicateExecutor<Person> {
+
+   // additional query methods go here
+}
+----
+
+NOTE: Please note that joins (DBRef's) are not supported with Reactive MongoDB support.