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

DATAMONGO-1992 - Add mutation support for immutable objects through ReactiveMongoTemplate.

Browse Source 
Persisting methods of ReactiveMongoTemplate now return potentially new object instances of immutable objects. New instances are created using wither methods/Kotlin copy(…) methods if an immutable object requires association with an Id or the version number needs to be incremented.
This commit is contained in:
Mark Paluch
2018-06-04 11:01:11 +02:00
committed by Oliver Gierke
parent 1eab66aff4
commit ba2ab183ed
3 changed files with 106 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java
View File

@@ -927,7 +927,7 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
* Insert is used to initially store the object into the database. To update an existing object use the save method. * Insert is used to initially store the object into the database. To update an existing object use the save method.
* *
* @param objectToSave the object to store in the collection. Must not be {@literal null}. * @param objectToSave the object to store in the collection. Must not be {@literal null}.
* @return the saved object. * @return the inserted object.
*/ */
<T> Mono<T> insert(T objectToSave); <T> Mono<T> insert(T objectToSave);
@@ -941,7 +941,7 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
* *
* @param objectToSave the object to store in the collection. Must not be {@literal null}. * @param objectToSave the object to store in the collection. Must not be {@literal null}.
* @param collectionName name of the collection to store the object in. Must not be {@literal null}. * @param collectionName name of the collection to store the object in. Must not be {@literal null}.
* @return the saved object. * @return the inserted object.
*/ */
<T> Mono<T> insert(T objectToSave, String collectionName); <T> Mono<T> insert(T objectToSave, String collectionName);
@@ -950,7 +950,7 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
* *
* @param batchToSave the batch of objects to save. Must not be {@literal null}. * @param batchToSave the batch of objects to save. Must not be {@literal null}.
* @param entityClass class that determines the collection to use. Must not be {@literal null}. * @param entityClass class that determines the collection to use. Must not be {@literal null}.
* @return the saved objects. * @return the inserted objects .
*/ */
<T> Flux<T> insert(Collection<? extends T> batchToSave, Class<?> entityClass); <T> Flux<T> insert(Collection<? extends T> batchToSave, Class<?> entityClass);
@@ -959,7 +959,7 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
* *
* @param batchToSave the list of objects to save. Must not be {@literal null}. * @param batchToSave the list of objects to save. Must not be {@literal null}.
* @param collectionName name of the collection to store the object in. Must not be {@literal null}. * @param collectionName name of the collection to store the object in. Must not be {@literal null}.
* @return the saved objects. * @return the inserted objects.
*/ */
<T> Flux<T> insert(Collection<? extends T> batchToSave, String collectionName); <T> Flux<T> insert(Collection<? extends T> batchToSave, String collectionName);
@@ -987,7 +987,7 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
* Insert is used to initially store the object into the database. To update an existing object use the save method. * Insert is used to initially store the object into the database. To update an existing object use the save method.
* *
* @param objectToSave the object to store in the collection. Must not be {@literal null}. * @param objectToSave the object to store in the collection. Must not be {@literal null}.
* @return the saved object. * @return the inserted objects.
*/ */
<T> Mono<T> insert(Mono<? extends T> objectToSave); <T> Mono<T> insert(Mono<? extends T> objectToSave);
@@ -996,7 +996,7 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
* *
* @param batchToSave the publisher which provides objects to save. Must not be {@literal null}. * @param batchToSave the publisher which provides objects to save. Must not be {@literal null}.
* @param entityClass class that determines the collection to use. Must not be {@literal null}. * @param entityClass class that determines the collection to use. Must not be {@literal null}.
* @return the saved objects. * @return the inserted objects.
*/ */
<T> Flux<T> insertAll(Mono<? extends Collection<? extends T>> batchToSave, Class<?> entityClass); <T> Flux<T> insertAll(Mono<? extends Collection<? extends T>> batchToSave, Class<?> entityClass);
@@ -1005,7 +1005,7 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
* *
* @param batchToSave the publisher which provides objects to save. Must not be {@literal null}. * @param batchToSave the publisher which provides objects to save. Must not be {@literal null}.
* @param collectionName name of the collection to store the object in. Must not be {@literal null}. * @param collectionName name of the collection to store the object in. Must not be {@literal null}.
* @return the saved objects. * @return the inserted objects.
*/ */
<T> Flux<T> insertAll(Mono<? extends Collection<? extends T>> batchToSave, String collectionName); <T> Flux<T> insertAll(Mono<? extends Collection<? extends T>> batchToSave, String collectionName);
@@ -1014,7 +1014,7 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
* class. * class.
* *
* @param objectsToSave the publisher which provides objects to save. Must not be {@literal null}. * @param objectsToSave the publisher which provides objects to save. Must not be {@literal null}.
* @return the saved objects. * @return the inserted objects.
*/ */
<T> Flux<T> insertAll(Mono<? extends Collection<? extends T>> objectsToSave); <T> Flux<T> insertAll(Mono<? extends Collection<? extends T>> objectsToSave);

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

@@ -144,6 +144,7 @@ import com.mongodb.reactivestreams.client.MongoCollection;
import com.mongodb.reactivestreams.client.MongoDatabase; import com.mongodb.reactivestreams.client.MongoDatabase;
import com.mongodb.reactivestreams.client.Success; import com.mongodb.reactivestreams.client.Success;
import com.mongodb.util.JSONParseException; import com.mongodb.util.JSONParseException;
import reactor.util.function.Tuples;
/** /**
* Primary implementation of {@link ReactiveMongoOperations}. It simplifies the use of Reactive MongoDB usage and helps * Primary implementation of {@link ReactiveMongoOperations}. It simplifies the use of Reactive MongoDB usage and helps
@@ -1252,17 +1253,18 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
return Mono.defer(() -> { return Mono.defer(() -> {
initializeVersionProperty(objectToSave); T toSave = (T) initializeVersionProperty(objectToSave);
maybeEmitEvent(new BeforeConvertEvent<>(objectToSave, collectionName)); maybeEmitEvent(new BeforeConvertEvent<>(toSave, collectionName));
Document dbDoc = toDocument(objectToSave, writer); Document dbDoc = toDocument(toSave, writer);
maybeEmitEvent(new BeforeSaveEvent<>(objectToSave, dbDoc, collectionName)); maybeEmitEvent(new BeforeSaveEvent<>(toSave, dbDoc, collectionName));
Mono<T> afterInsert = insertDBObject(collectionName, dbDoc, objectToSave.getClass()).flatMap(id -> { Mono<T> afterInsert = insertDBObject(collectionName, dbDoc, toSave.getClass()).map(id -> {
populateIdIfNecessary(objectToSave, id);
maybeEmitEvent(new AfterSaveEvent<>(objectToSave, dbDoc, collectionName)); T saved = (T) populateIdIfNecessary(toSave, id);
return Mono.just(objectToSave); maybeEmitEvent(new AfterSaveEvent<>(saved, dbDoc, collectionName));
return saved;
}); });
return afterInsert; return afterInsert;
@@ -1326,18 +1328,15 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
Assert.notNull(writer, "MongoWriter must not be null!"); Assert.notNull(writer, "MongoWriter must not be null!");
Mono<List<Tuple2<T, Document>>> prepareDocuments = Flux.fromIterable(batchToSave) Mono<List<Tuple2<T, Document>>> prepareDocuments = Flux.fromIterable(batchToSave)
.flatMap(new Function<T, Flux<Tuple2<T, Document>>>() { .map(o -> {
@Override
public Flux<Tuple2<T, Document>> apply(T o) {
initializeVersionProperty(o); T toSave = (T) initializeVersionProperty(o);
maybeEmitEvent(new BeforeConvertEvent<>(o, collectionName)); maybeEmitEvent(new BeforeConvertEvent<>(toSave, collectionName));
Document dbDoc = toDocument(o, writer); Document dbDoc = toDocument(toSave, writer);
maybeEmitEvent(new BeforeSaveEvent<>(o, dbDoc, collectionName)); maybeEmitEvent(new BeforeSaveEvent<>(toSave, dbDoc, collectionName));
return Flux.zip(Mono.just(o), Mono.just(dbDoc)); return Tuples.of(toSave, dbDoc);
}
}).collectList(); }).collectList();
Flux<Tuple2<T, Document>> insertDocuments = prepareDocuments.flatMapMany(tuples -> { Flux<Tuple2<T, Document>> insertDocuments = prepareDocuments.flatMapMany(tuples -> {
@@ -1349,9 +1348,9 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
return insertDocuments.map(tuple -> { return insertDocuments.map(tuple -> {
populateIdIfNecessary(tuple.getT1(), tuple.getT2().get(ID_FIELD)); T saved = (T) populateIdIfNecessary(tuple.getT1(), tuple.getT2().get(ID_FIELD));
maybeEmitEvent(new AfterSaveEvent<>(tuple.getT1(), tuple.getT2(), collectionName)); maybeEmitEvent(new AfterSaveEvent<>(saved, tuple.getT2(), collectionName));
return tuple.getT1(); return saved;
}); });
} }
@@ -1438,17 +1437,19 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
// Bump version number // Bump version number
convertingAccessor.setProperty(versionProperty, versionNumber.longValue() + 1); convertingAccessor.setProperty(versionProperty, versionNumber.longValue() + 1);
ReactiveMongoTemplate.this.maybeEmitEvent(new BeforeConvertEvent<>(objectToSave, collectionName)); T toSave = (T) convertingAccessor.getBean();
Document document = ReactiveMongoTemplate.this.toDocument(objectToSave, mongoConverter); ReactiveMongoTemplate.this.maybeEmitEvent(new BeforeConvertEvent<T>(toSave, collectionName));
ReactiveMongoTemplate.this.maybeEmitEvent(new BeforeSaveEvent<>(objectToSave, document, collectionName)); Document document = ReactiveMongoTemplate.this.toDocument(toSave, mongoConverter);
ReactiveMongoTemplate.this.maybeEmitEvent(new BeforeSaveEvent<>(toSave, document, collectionName));
Update update = Update.fromDocument(document, ID_FIELD); Update update = Update.fromDocument(document, ID_FIELD);
return doUpdate(collectionName, query, update, objectToSave.getClass(), false, false).map(updateResult -> { return doUpdate(collectionName, query, update, toSave.getClass(), false, false).map(updateResult -> {
maybeEmitEvent(new AfterSaveEvent<>(objectToSave, document, collectionName)); maybeEmitEvent(new AfterSaveEvent<>(toSave, document, collectionName));
return objectToSave; return toSave;
}); });
}); });
} }
@@ -1465,9 +1466,9 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
return saveDocument(collectionName, dbDoc, objectToSave.getClass()).map(id -> { return saveDocument(collectionName, dbDoc, objectToSave.getClass()).map(id -> {
populateIdIfNecessary(objectToSave, id); T saved = (T) populateIdIfNecessary(objectToSave, id);
maybeEmitEvent(new AfterSaveEvent<>(objectToSave, dbDoc, collectionName)); maybeEmitEvent(new AfterSaveEvent<>(saved, dbDoc, collectionName));
return objectToSave; return saved;
}); });
}); });
} }
@@ -2521,10 +2522,10 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
* @param id * @param id
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
private void populateIdIfNecessary(Object savedObject, @Nullable Object id) { private Object populateIdIfNecessary(Object savedObject, @Nullable Object id) {
if (id == null) { if (id == null) {
return; return null;
} }
if (savedObject instanceof Map) { if (savedObject instanceof Map) {
@@ -2532,13 +2533,13 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
Map<String, Object> map = (Map<String, Object>) savedObject; Map<String, Object> map = (Map<String, Object>) savedObject;
map.put(ID_FIELD, id); map.put(ID_FIELD, id);
return; return map;
} }
MongoPersistentProperty idProp = getIdPropertyFor(savedObject.getClass()); MongoPersistentProperty idProp = getIdPropertyFor(savedObject.getClass());
if (idProp == null) { if (idProp == null) {
return; return savedObject;
} }
ConversionService conversionService = mongoConverter.getConversionService(); ConversionService conversionService = mongoConverter.getConversionService();
@@ -2546,10 +2547,12 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
PersistentPropertyAccessor accessor = entity.getPropertyAccessor(savedObject); PersistentPropertyAccessor accessor = entity.getPropertyAccessor(savedObject);
if (accessor.getProperty(idProp) != null) { if (accessor.getProperty(idProp) != null) {
return; return accessor.getBean();
} }
new ConvertingPropertyAccessor(accessor, conversionService).setProperty(idProp, id); new ConvertingPropertyAccessor(accessor, conversionService).setProperty(idProp, id);
return accessor.getBean();
} }
private MongoCollection<Document> getAndPrepareCollection(MongoDatabase db, String collectionName) { private MongoCollection<Document> getAndPrepareCollection(MongoDatabase db, String collectionName) {
@@ -2807,7 +2810,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
} }
} }
private void initializeVersionProperty(Object entity) { private Object initializeVersionProperty(Object entity) {
MongoPersistentEntity<?> mongoPersistentEntity = getPersistentEntity(entity.getClass()); MongoPersistentEntity<?> mongoPersistentEntity = getPersistentEntity(entity.getClass());
@@ -2815,7 +2818,11 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
ConvertingPropertyAccessor accessor = new ConvertingPropertyAccessor( ConvertingPropertyAccessor accessor = new ConvertingPropertyAccessor(
mongoPersistentEntity.getPropertyAccessor(entity), mongoConverter.getConversionService()); mongoPersistentEntity.getPropertyAccessor(entity), mongoConverter.getConversionService());
accessor.setProperty(mongoPersistentEntity.getRequiredVersionProperty(), 0); accessor.setProperty(mongoPersistentEntity.getRequiredVersionProperty(), 0);
return accessor.getBean();
} }
return entity;
} }
// Callback implementations // Callback implementations

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

@@ -23,6 +23,7 @@ import static org.springframework.data.mongodb.test.util.Assertions.*;
import lombok.AllArgsConstructor; import lombok.AllArgsConstructor;
import lombok.Data; import lombok.Data;
import lombok.NoArgsConstructor; import lombok.NoArgsConstructor;
import lombok.experimental.Wither;
import reactor.core.Disposable; import reactor.core.Disposable;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
@@ -57,6 +58,7 @@ import org.springframework.dao.DuplicateKeyException;
import org.springframework.dao.InvalidDataAccessApiUsageException; import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.data.annotation.Id; import org.springframework.data.annotation.Id;
import org.springframework.data.annotation.Version;
import org.springframework.data.domain.Sort; import org.springframework.data.domain.Sort;
import org.springframework.data.domain.Sort.Direction; import org.springframework.data.domain.Sort.Direction;
import org.springframework.data.geo.Metrics; import org.springframework.data.geo.Metrics;
@@ -96,17 +98,16 @@ public class ReactiveMongoTemplateTests {
@Before @Before
public void setUp() { public void setUp() {
StepVerifier StepVerifier.create(template.dropCollection("people") //
.create(template.dropCollection("people") // .mergeWith(template.dropCollection("personX")) //
.mergeWith(template.dropCollection("personX")) // .mergeWith(template.dropCollection("collection")) //
.mergeWith(template.dropCollection("collection")) // .mergeWith(template.dropCollection(Person.class)) //
.mergeWith(template.dropCollection(Person.class)) // .mergeWith(template.dropCollection(Venue.class)) //
.mergeWith(template.dropCollection(Venue.class)) // .mergeWith(template.dropCollection(PersonWithAList.class)) //
.mergeWith(template.dropCollection(PersonWithAList.class)) // .mergeWith(template.dropCollection(PersonWithIdPropertyOfTypeObjectId.class)) //
.mergeWith(template.dropCollection(PersonWithIdPropertyOfTypeObjectId.class)) // .mergeWith(template.dropCollection(PersonWithVersionPropertyOfTypeInteger.class)) //
.mergeWith(template.dropCollection(PersonWithVersionPropertyOfTypeInteger.class)) // .mergeWith(template.dropCollection(Sample.class)) //
.mergeWith(template.dropCollection(Sample.class)) // .mergeWith(template.dropCollection(MyPerson.class))) //
.mergeWith(template.dropCollection(MyPerson.class))) //
.verifyComplete(); .verifyComplete();
} }
@@ -310,12 +311,9 @@ public class ReactiveMongoTemplateTests {
public void updateFirstByEntityTypeShouldUpdateObject() { public void updateFirstByEntityTypeShouldUpdateObject() {
Person person = new Person("Oliver2", 25); Person person = new Person("Oliver2", 25);
StepVerifier StepVerifier.create(template.insert(person) //
.create(template.insert(person) // .then(template.updateFirst(new Query(where("age").is(25)), new Update().set("firstName", "Sven"), Person.class)) //
.then(template.updateFirst(new Query(where("age").is(25)), new Update().set("firstName", "Sven"), .flatMapMany(p -> template.find(new Query(where("age").is(25)), Person.class))).consumeNextWith(actual -> {
Person.class)) //
.flatMapMany(p -> template.find(new Query(where("age").is(25)), Person.class)))
.consumeNextWith(actual -> {
assertThat(actual.getFirstName()).isEqualTo("Sven"); assertThat(actual.getFirstName()).isEqualTo("Sven");
}).verifyComplete(); }).verifyComplete();
@@ -325,10 +323,9 @@ public class ReactiveMongoTemplateTests {
public void updateFirstByCollectionNameShouldUpdateObjects() { public void updateFirstByCollectionNameShouldUpdateObjects() {
Person person = new Person("Oliver2", 25); Person person = new Person("Oliver2", 25);
StepVerifier StepVerifier.create(template.insert(person, "people") //
.create(template.insert(person, "people") // .then(template.updateFirst(new Query(where("age").is(25)), new Update().set("firstName", "Sven"), "people")) //
.then(template.updateFirst(new Query(where("age").is(25)), new Update().set("firstName", "Sven"), "people")) // .flatMapMany(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 -> { .consumeNextWith(actual -> {
assertThat(actual.getFirstName()).isEqualTo("Sven"); assertThat(actual.getFirstName()).isEqualTo("Sven");
@@ -741,9 +738,8 @@ public class ReactiveMongoTemplateTests {
Document dbObject = new Document(); Document dbObject = new Document();
dbObject.put("firstName", "Oliver"); dbObject.put("firstName", "Oliver");
StepVerifier StepVerifier.create(template.insert(dbObject, //
.create(template.insert(dbObject, // template.determineCollectionName(PersonWithVersionPropertyOfTypeInteger.class))) //
template.determineCollectionName(PersonWithVersionPropertyOfTypeInteger.class))) //
.expectNextCount(1) // .expectNextCount(1) //
.verifyComplete(); .verifyComplete();
} }
@@ -884,6 +880,23 @@ public class ReactiveMongoTemplateTests {
assertThat(person.version).isZero(); assertThat(person.version).isZero();
} }
@Test // DATAMONGO-1992
public void initializesIdAndVersionAndOfImmutableObject() {
ImmutableVersioned versioned = new ImmutableVersioned();
StepVerifier.create(template.insert(versioned)).consumeNextWith(actual -> {
assertThat(actual).isNotSameAs(versioned);
assertThat(versioned.id).isNull();
assertThat(versioned.version).isNull();
assertThat(actual.id).isNotNull();
assertThat(actual.version).isEqualTo(0);
}).verifyComplete();
}
@Test // DATAMONGO-1444 @Test // DATAMONGO-1444
public void queryCanBeNull() { public void queryCanBeNull() {
@@ -920,9 +933,8 @@ public class ReactiveMongoTemplateTests {
ReactiveMongoTemplate template = new ReactiveMongoTemplate(factory); ReactiveMongoTemplate template = new ReactiveMongoTemplate(factory);
template.setWriteResultChecking(WriteResultChecking.EXCEPTION); template.setWriteResultChecking(WriteResultChecking.EXCEPTION);
StepVerifier StepVerifier.create(template.indexOps(Person.class) //
.create(template.indexOps(Person.class) // .ensureIndex(new Index().on("firstName", Direction.DESC).unique())) //
.ensureIndex(new Index().on("firstName", Direction.DESC).unique())) //
.expectNextCount(1) // .expectNextCount(1) //
.verifyComplete(); .verifyComplete();
@@ -1040,9 +1052,8 @@ public class ReactiveMongoTemplateTests {
@Test // DATAMONGO-1444 @Test // DATAMONGO-1444
public void tailStreamsData() throws InterruptedException { public void tailStreamsData() throws InterruptedException {
StepVerifier.create(template.dropCollection("capped") StepVerifier.create(template.dropCollection("capped").then(template.createCollection("capped", //
.then(template.createCollection("capped", // CollectionOptions.empty().size(1000).maxDocuments(10).capped()))
CollectionOptions.empty().size(1000).maxDocuments(10).capped()))
.then(template.insert(new Document("random", Math.random()).append("key", "value"), // .then(template.insert(new Document("random", Math.random()).append("key", "value"), //
"capped"))) "capped")))
.expectNextCount(1).verifyComplete(); .expectNextCount(1).verifyComplete();
@@ -1062,9 +1073,8 @@ public class ReactiveMongoTemplateTests {
@Test // DATAMONGO-1444 @Test // DATAMONGO-1444
public void tailStreamsDataUntilCancellation() throws InterruptedException { public void tailStreamsDataUntilCancellation() throws InterruptedException {
StepVerifier.create(template.dropCollection("capped") StepVerifier.create(template.dropCollection("capped").then(template.createCollection("capped", //
.then(template.createCollection("capped", // CollectionOptions.empty().size(1000).maxDocuments(10).capped()))
CollectionOptions.empty().size(1000).maxDocuments(10).capped()))
.then(template.insert(new Document("random", Math.random()).append("key", "value"), // .then(template.insert(new Document("random", Math.random()).append("key", "value"), //
"capped"))) "capped")))
.expectNextCount(1).verifyComplete(); .expectNextCount(1).verifyComplete();
@@ -1397,6 +1407,19 @@ public class ReactiveMongoTemplateTests {
return p; return p;
} }
@AllArgsConstructor
@Wither
static class ImmutableVersioned {
final @Id String id;
final @Version Long version;
public ImmutableVersioned() {
id = null;
version = null;
}
}
@Data @Data
static class Sample { static class Sample {