diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java index ad891b3d3..870152ba1 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java @@ -422,7 +422,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati Mono> 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> 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 callback = new GeoNearResultDbObjectCallback( new ReadDocumentCallback(mongoConverter, entityClass, collectionName), near.getMetric()); - return executeCommand(command, this.readPreference).flatMap(document -> { + return executeCommand(command, this.readPreference).flatMapMany(document -> { List l = document.get("results", List.class); if (l == null) { @@ -766,7 +766,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati */ @Override public Mono insert(Mono 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(objectToSave, dbDoc, collectionName)); - Mono afterInsert = insertDBObject(collectionName, dbDoc, objectToSave.getClass()).then(id -> { + Mono afterInsert = insertDBObject(collectionName, dbDoc, objectToSave.getClass()).flatMap(id -> { populateIdIfNecessary(objectToSave, id); maybeEmitEvent(new AfterSaveEvent(objectToSave, dbDoc, collectionName)); return Mono.just(objectToSave); @@ -897,7 +897,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati } }).collectList(); - Flux> insertDocuments = prepareDocuments.flatMap(tuples -> { + Flux> insertDocuments = prepareDocuments.flatMapMany(tuples -> { List dbObjects = tuples.stream().map(Tuple2::getT2).collect(Collectors.toList()); @@ -917,7 +917,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati */ @Override public Mono save(Mono 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 Mono save(Mono 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 remove(Mono 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 remove(Mono 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 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))); } diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/support/SimpleReactiveMongoRepository.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/support/SimpleReactiveMongoRepository.java index a1ae5c790..bc3e9b250 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/support/SimpleReactiveMongoRepository.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/support/SimpleReactiveMongoRepository.java @@ -77,7 +77,7 @@ public class SimpleReactiveMongoRepository 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 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 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 implement // TODO: should this one really be void? public Mono deleteAll() { return mongoOperations.remove(new Query(), entityInformation.getCollectionName()) - .then(deleteResult -> Mono.empty()); + .then(Mono.empty()); } private Query getIdQuery(Object id) { diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java index c06121319..9a6e0d06b 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java @@ -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 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) //