From d4f1ef8704ac8907dce91ad0c14d194027a3487e Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Mon, 16 Feb 2015 16:02:29 +0100 Subject: [PATCH] DATAMONGO-1165 - Add support for Java 8 Stream as return type for repository methods. Added support for a MongoDB Cursor backed Iterator that allows the usage of a Java 8 Stream at the repository level. Original pull request: #274. --- .../data/mongodb/core/MongoOperations.java | 16 +++ .../data/mongodb/core/MongoTemplate.java | 103 +++++++++++++++++- .../repository/query/AbstractMongoQuery.java | 37 ++++++- ...tractPersonRepositoryIntegrationTests.java | 4 +- .../mongodb/repository/PersonRepository.java | 9 +- .../PersonRepositoryIntegrationTests.java | 30 ++++- 6 files changed, 187 insertions(+), 12 deletions(-) diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java index 6e3261db4..2c41d5e13 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java @@ -33,8 +33,10 @@ import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.NearQuery; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; +import org.springframework.data.util.CloseableIterator; import com.mongodb.CommandResult; +import com.mongodb.Cursor; import com.mongodb.DB; import com.mongodb.DBCollection; import com.mongodb.DBObject; @@ -165,6 +167,20 @@ public interface MongoOperations { @Deprecated T executeInSession(DbCallback action); + /** + * Executes the given {@link Query} on the entity collection of the specified {@code entityType} backed by a Mongo DB + * {@link Cursor}. + *

+ * Returns a {@link CloseableIterator} that wraps the a Mongo DB {@link Cursor} that needs to be closed. + * + * @param element return type + * @param query + * @param entityType + * @return + * @since 1.7 + */ + CloseableIterator executeAsStream(Query query, Class entityType); + /** * Create an uncapped collection with a name based on the provided entity class. * diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java index 04c140ace..6b99b8b89 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java @@ -95,6 +95,8 @@ import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.NearQuery; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; +import org.springframework.data.util.CloseableIterator; +import org.springframework.data.mongodb.util.CloseableIterator; import org.springframework.data.mongodb.util.MongoClientVersion; import org.springframework.jca.cci.core.ConnectionCallback; import org.springframework.util.Assert; @@ -106,6 +108,7 @@ import org.springframework.util.StringUtils; import com.mongodb.BasicDBObject; import com.mongodb.Bytes; import com.mongodb.CommandResult; +import com.mongodb.Cursor; import com.mongodb.DB; import com.mongodb.DBCollection; import com.mongodb.DBCursor; @@ -315,6 +318,31 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware { return this.mongoConverter; } + /* (non-Javadoc) + * @see org.springframework.data.mongodb.core.MongoOperations#executeAsStream(org.springframework.data.mongodb.core.query.Query, java.lang.Class) + */ + @Override + public CloseableIterator executeAsStream(final Query query, final Class entityType) { + + return execute(entityType, new CollectionCallback>() { + + @Override + public CloseableIterator doInCollection(DBCollection collection) throws MongoException, DataAccessException { + + MongoPersistentEntity persistentEntity = mappingContext.getPersistentEntity(entityType); + + DBObject mappedFields = queryMapper.getMappedFields(query.getFieldsObject(), persistentEntity); + DBObject mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), persistentEntity); + + DBCursor cursor = collection.find(mappedQuery, mappedFields); + ReadDbObjectCallback readCallback = new ReadDbObjectCallback(mongoConverter, entityType); + + return new CloseableIterableCusorAdapter(cursor, exceptionTranslator, readCallback); + } + }); + + } + public String getCollectionName(Class entityClass) { return this.determineCollectionName(entityClass); } @@ -2122,9 +2150,10 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware { * Simple internal callback to allow operations on a {@link DBObject}. * * @author Oliver Gierke + * @author Thomas Darimont */ - private interface DbObjectCallback { + static interface DbObjectCallback { T doWith(DBObject object); } @@ -2287,4 +2316,76 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware { } } + /** + * A {@link CloseableIterator} that is backed by a MongoDB {@link Cursor}. + * + * @since 1.7 + * @author Thomas Darimont + */ + static class CloseableIterableCusorAdapter implements CloseableIterator { + + private volatile Cursor cursor; + private PersistenceExceptionTranslator exceptionTranslator; + private DbObjectCallback objectReadCallback; + + /** + * Creates a new {@link CloseableIterableCusorAdapter} backed by the given {@link Cursor}. + * + * @param cursor + * @param exceptionTranslator + * @param objectReadCallback + */ + public CloseableIterableCusorAdapter(Cursor cursor, PersistenceExceptionTranslator exceptionTranslator, + DbObjectCallback objectReadCallback) { + + this.cursor = cursor; + this.exceptionTranslator = exceptionTranslator; + this.objectReadCallback = objectReadCallback; + } + + @Override + public boolean hasNext() { + + if (cursor == null) { + return false; + } + + try { + return cursor.hasNext(); + } catch (RuntimeException ex) { + throw exceptionTranslator.translateExceptionIfPossible(ex); + } + } + + @Override + public T next() { + + if (cursor == null) { + return null; + } + + try { + DBObject item = cursor.next(); + T converted = objectReadCallback.doWith(item); + return converted; + } catch (RuntimeException ex) { + throw exceptionTranslator.translateExceptionIfPossible(ex); + } + } + + @Override + public void close() { + + Cursor c = cursor; + try { + c.close(); + } catch (RuntimeException ex) { + throw exceptionTranslator.translateExceptionIfPossible(ex); + } finally { + cursor = null; + exceptionTranslator = null; + objectReadCallback = null; + } + } + } } diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/AbstractMongoQuery.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/AbstractMongoQuery.java index f6676fecb..dce6f110d 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/AbstractMongoQuery.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/AbstractMongoQuery.java @@ -1,5 +1,5 @@ /* - * Copyright 2010-2014 the original author or authors. + * Copyright 2010-2015 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. @@ -17,9 +17,10 @@ package org.springframework.data.mongodb.repository.query; import java.util.Collections; import java.util.List; +import java.util.Spliterator; +import java.util.Spliterators; +import java.util.stream.StreamSupport; -import org.springframework.core.convert.ConversionService; -import org.springframework.core.convert.support.DefaultConversionService; import org.springframework.data.domain.PageImpl; import org.springframework.data.domain.Pageable; import org.springframework.data.domain.Slice; @@ -34,6 +35,8 @@ import org.springframework.data.mongodb.core.query.NearQuery; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.repository.query.ParameterAccessor; import org.springframework.data.repository.query.RepositoryQuery; +import org.springframework.data.util.CloseableIterator; +import org.springframework.data.util.CloseableIteratorDisposingRunnable; import org.springframework.data.util.TypeInformation; import org.springframework.util.Assert; @@ -48,8 +51,6 @@ import com.mongodb.WriteResult; */ public abstract class AbstractMongoQuery implements RepositoryQuery { - private static final ConversionService CONVERSION_SERVICE = new DefaultConversionService(); - private final MongoQueryMethod method; private final MongoOperations operations; @@ -87,7 +88,9 @@ public abstract class AbstractMongoQuery implements RepositoryQuery { applyQueryMetaAttributesWhenPresent(query); - if (isDeleteQuery()) { + if (method.isStreamQuery()) { + return new StreamExecution().execute(query); + } else if (isDeleteQuery()) { return new DeleteExecution().execute(query); } else if (method.isGeoNearQuery() && method.isPageQuery()) { @@ -415,4 +418,26 @@ public abstract class AbstractMongoQuery implements RepositoryQuery { return writeResult != null ? writeResult.getN() : 0L; } } + + /** + * @author Thomas Darimont + * @since 1.7 + */ + final class StreamExecution extends Execution { + + /* (non-Javadoc) + * @see org.springframework.data.mongodb.repository.query.AbstractMongoQuery.Execution#execute(org.springframework.data.mongodb.core.query.Query) + */ + @Override + Object execute(Query query) { + + Class entityType = getQueryMethod().getEntityInformation().getJavaType(); + + @SuppressWarnings("unchecked") + CloseableIterator result = (CloseableIterator) operations.executeAsStream(query, entityType); + Spliterator spliterator = Spliterators.spliteratorUnknownSize(result, Spliterator.NONNULL); + + return StreamSupport.stream(spliterator, false).onClose(new CloseableIteratorDisposingRunnable(result)); + } + } } diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/AbstractPersonRepositoryIntegrationTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/AbstractPersonRepositoryIntegrationTests.java index 5423ad68c..3f2c07723 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/AbstractPersonRepositoryIntegrationTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/AbstractPersonRepositoryIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2011-2014 the original author or authors. + * Copyright 2011-2015 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. @@ -65,7 +65,7 @@ public abstract class AbstractPersonRepositoryIntegrationTests { @Autowired MongoOperations operations; - Person dave, oliver, carter, boyd, stefan, leroi, alicia; + protected Person dave, oliver, carter, boyd, stefan, leroi, alicia; QPerson person; List all; diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepository.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepository.java index 9d9593742..2d5d0c12a 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepository.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepository.java @@ -1,5 +1,5 @@ /* - * Copyright 2010-2014 the original author or authors. + * Copyright 2010-2015 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. @@ -18,6 +18,7 @@ package org.springframework.data.mongodb.repository; import java.util.Collection; import java.util.Date; import java.util.List; +import java.util.stream.Stream; import org.springframework.data.domain.Page; import org.springframework.data.domain.Pageable; @@ -320,4 +321,10 @@ public interface PersonRepository extends MongoRepository, Query @Query("{ ?0 : ?1 }") List findByKeyValue(String key, String value); + + /** + * @see DATAMONGO-1165 + */ + @Query("{firstname:{$in:?0}}") + Stream findByCustomQueryWithStreamingCursorByFirstnames(List firstnames); } diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepositoryIntegrationTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepositoryIntegrationTests.java index 8dfde08b7..98fdc43ae 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepositoryIntegrationTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepositoryIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2010-2013 the original author or authors. + * Copyright 2010-2015 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. @@ -15,6 +15,15 @@ */ package org.springframework.data.mongodb.repository; +import static org.hamcrest.Matchers.*; +import static org.junit.Assert.*; + +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import org.junit.Test; import org.springframework.test.context.ContextConfiguration; /** @@ -24,4 +33,21 @@ import org.springframework.test.context.ContextConfiguration; * @author Thomas Darimont */ @ContextConfiguration -public class PersonRepositoryIntegrationTests extends AbstractPersonRepositoryIntegrationTests {} +public class PersonRepositoryIntegrationTests extends AbstractPersonRepositoryIntegrationTests { + + /** + * @see DATAMONGO-1165 + */ + @Test + public void shouldAllowReturningJava8StreamInCustomQuery() throws Exception { + + Stream result = repository.findByCustomQueryWithStreamingCursorByFirstnames(Arrays.asList("Dave")); + + try { + List readPersons = result.collect(Collectors. toList()); + assertThat(readPersons, hasItems(dave)); + } finally { + result.close(); + } + } +}