Spring/spring-data-mongodb
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAMONGO-1665 - Upgraded to Reactor 3.1.0.BUILD-SNAPSHOT.

Browse Source 
Original pull request: #456.
Related ticket: spring-projects/spring-data-build#333
This commit is contained in:
Sebastien Deleuze
2017-04-11 23:23:17 +02:00
committed by Oliver Gierke
parent 314b95370f
commit 935db07511
3 changed files with 19 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java
View File

@@ -422,7 +422,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
Mono<MongoCollection<Document>> collectionPublisher = Mono
.fromCallable(() -> getAndPrepareCollection(getMongoDatabase(), collectionName));
return collectionPublisher.flatMap(callback::doInCollection).onErrorResumeWith(translateFluxException());
return collectionPublisher.flatMapMany(callback::doInCollection).onErrorResumeWith(translateFluxException());
}
/**
@@ -441,7 +441,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
Mono<MongoCollection<Document>> collectionPublisher = Mono
.fromCallable(() -> getAndPrepareCollection(getMongoDatabase(), collectionName));
return collectionPublisher.then(collection -> Mono.from(callback.doInCollection(collection)))
return collectionPublisher.flatMap(collection -> Mono.from(callback.doInCollection(collection)))
.otherwise(translateMonoException());
}
@@ -665,7 +665,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
GeoNearResultDbObjectCallback<T> callback = new GeoNearResultDbObjectCallback<T>(
new ReadDocumentCallback<T>(mongoConverter, entityClass, collectionName), near.getMetric());
return executeCommand(command, this.readPreference).flatMap(document -> {
return executeCommand(command, this.readPreference).flatMapMany(document -> {
List<Document> l = document.get("results", List.class);
if (l == null) {
@@ -766,7 +766,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
*/
@Override
public <T> Mono<T> insert(Mono<? extends T> objectToSave) {
return objectToSave.then(this::insert);
return objectToSave.flatMap(this::insert);
}
/* (non-Javadoc)
@@ -816,7 +816,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
maybeEmitEvent(new BeforeSaveEvent<T>(objectToSave, dbDoc, collectionName));
Mono<T> afterInsert = insertDBObject(collectionName, dbDoc, objectToSave.getClass()).then(id -> {
Mono<T> afterInsert = insertDBObject(collectionName, dbDoc, objectToSave.getClass()).flatMap(id -> {
populateIdIfNecessary(objectToSave, id);
maybeEmitEvent(new AfterSaveEvent<T>(objectToSave, dbDoc, collectionName));
return Mono.just(objectToSave);
@@ -897,7 +897,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
}
}).collectList();
Flux<Tuple2<T, Document>> insertDocuments = prepareDocuments.flatMap(tuples -> {
Flux<Tuple2<T, Document>> insertDocuments = prepareDocuments.flatMapMany(tuples -> {
List<Document> dbObjects = tuples.stream().map(Tuple2::getT2).collect(Collectors.toList());
@@ -917,7 +917,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
*/
@Override
public <T> Mono<T> save(Mono<? extends T> objectToSave) {
return objectToSave.then(this::save);
return objectToSave.flatMap(this::save);
}
/* (non-Javadoc)
@@ -925,7 +925,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
*/
@Override
public <T> Mono<T> save(Mono<? extends T> objectToSave, String collectionName) {
return objectToSave.then(o -> save(o, collectionName));
return objectToSave.flatMap(o -> save(o, collectionName));
}
/* (non-Javadoc)
@@ -1247,7 +1247,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
*/
@Override
public Mono<DeleteResult> remove(Mono<? extends Object> objectToRemove) {
return objectToRemove.then(this::remove);
return objectToRemove.flatMap(this::remove);
}
/* (non-Javadoc)
@@ -1255,7 +1255,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
*/
@Override
public Mono<DeleteResult> remove(Mono<? extends Object> objectToRemove, String collection) {
return objectToRemove.then(o -> remove(objectToRemove, collection));
return objectToRemove.flatMap(o -> remove(objectToRemove, collection));
}
/* (non-Javadoc)
@@ -1496,7 +1496,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
Flux<T> flux = find(query, entityClass, collectionName);
return Flux.from(flux).collectList()
.flatMap(list -> Flux.from(remove(getIdInQueryFor(list), entityClass, collectionName))
.flatMapMany(list -> Flux.from(remove(getIdInQueryFor(list), entityClass, collectionName))
.flatMap(deleteResult -> Flux.fromIterable(list)));
}

10
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/support/SimpleReactiveMongoRepository.java
View File

@@ -77,7 +77,7 @@ public class SimpleReactiveMongoRepository<T, ID extends Serializable> implement
Assert.notNull(mono, "The given id must not be null!");
return mono.then(
return mono.flatMap(
id -> mongoOperations.findById(id, entityInformation.getJavaType(), entityInformation.getCollectionName()));
}
@@ -101,7 +101,7 @@ public class SimpleReactiveMongoRepository<T, ID extends Serializable> implement
Assert.notNull(mono, "The given id must not be null!");
return mono.then(id -> mongoOperations.exists(getIdQuery(id), entityInformation.getJavaType(),
return mono.flatMap(id -> mongoOperations.exists(getIdQuery(id), entityInformation.getJavaType(),
entityInformation.getCollectionName()));
}
@@ -250,10 +250,10 @@ public class SimpleReactiveMongoRepository<T, ID extends Serializable> implement
return Flux.from(entityStream).flatMap(entity -> {
if (entityInformation.isNew(entity)) {
return mongoOperations.insert(entity, entityInformation.getCollectionName()).then(aVoid -> Mono.just(entity));
return mongoOperations.insert(entity, entityInformation.getCollectionName()).then(Mono.just(entity));
}
return mongoOperations.save(entity, entityInformation.getCollectionName()).then(aVoid -> Mono.just(entity));
return mongoOperations.save(entity, entityInformation.getCollectionName()).then(Mono.just(entity));
});
}
@@ -295,7 +295,7 @@ public class SimpleReactiveMongoRepository<T, ID extends Serializable> implement
// TODO: should this one really be void?
public Mono<Void> deleteAll() {
return mongoOperations.remove(new Query(), entityInformation.getCollectionName())
.then(deleteResult -> Mono.empty());
.then(Mono.empty());
}
private Query getIdQuery(Object id) {

6
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java
View File

@@ -298,7 +298,7 @@ public class ReactiveMongoTemplateTests {
Person person = new Person("Oliver2", 25);
StepVerifier.create(template.insert(person) //
.then(template.updateFirst(new Query(where("age").is(25)), new Update().set("firstName", "Sven"), Person.class)) //
.flatMap(p -> template.find(new Query(where("age").is(25)), Person.class))).consumeNextWith(actual -> {
.flatMapMany(p -> template.find(new Query(where("age").is(25)), Person.class))).consumeNextWith(actual -> {
assertThat(actual.getFirstName(), is(equalTo("Sven")));
}).verifyComplete();
@@ -311,7 +311,7 @@ public class ReactiveMongoTemplateTests {
StepVerifier
.create(template.insert(person, "people") //
.then(template.updateFirst(new Query(where("age").is(25)), new Update().set("firstName", "Sven"), "people")) //
.flatMap(p -> template.find(new Query(where("age").is(25)), Person.class, "people")))
.flatMapMany(p -> template.find(new Query(where("age").is(25)), Person.class, "people")))
.consumeNextWith(actual -> {
assertThat(actual.getFirstName(), is(equalTo("Sven")));
@@ -347,7 +347,7 @@ public class ReactiveMongoTemplateTests {
Flux<Person> personFlux = template.insertAll(Mono.just(people), "people") //
.collectList() //
.flatMap(a -> template.updateMulti(query, new Update().set("firstName", "Walt"), Person.class, "people")) //
.flatMap(p -> template.find(new Query(where("firstName").is("Walt")), Person.class, "people"));
.flatMapMany(p -> template.find(new Query(where("firstName").is("Walt")), Person.class, "people"));
StepVerifier.create(personFlux) //
.expectNextCount(2) //