DATAMONGO-2150 - Fixed broken auditing for entities using optimistic locking.

The previous implementation of ReactiveMongoTemplate.doSaveVersioned(…) prematurely initialized the version property so that the entity wasn't considered new by the auditing subsystem. Even worse, for primitive version properties, the initialization kept the property at a value of 0, so that the just persisted entity was still considered new. This mean that via the repository route, inserts are triggered even for subsequent attempts to save an entity which caused duplicate key exceptions.

We now make sure we fire the BeforeConvertEvent before the version property is initialized or updated. Also, the initialization of the property now sets primitive properties to 1 initially.

Added integration tests for the auditing via ReactiveMongoTemplate and repositories.

Related ticket: DATAMONGO-2139.

Original Pull Request: #627

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

@@ -1234,23 +1234,22 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 
 	protected <T> Mono<T> doInsert(String collectionName, T objectToSave, MongoWriter<Object> writer) {
 
-		assertUpdateableIdIfNotSet(objectToSave);
-
 		return Mono.defer(() -> {
 
-			AdaptibleEntity<T> entity = operations.forEntity(objectToSave, mongoConverter.getConversionService());
+			BeforeConvertEvent<T> event = new BeforeConvertEvent<>(objectToSave, collectionName);
-			T toSave = entity.initializeVersionProperty();
+			T toConvert = maybeEmitEvent(event).getSource();
-
+			AdaptibleEntity<T> entity = operations.forEntity(toConvert, mongoConverter.getConversionService());
-			maybeEmitEvent(new BeforeConvertEvent<>(toSave, collectionName));
 
+			entity.assertUpdateableIdIfNotSet();
+			T initialized = entity.initializeVersionProperty();
 			Document dbDoc = entity.toMappedDocument(writer).getDocument();
 
-			maybeEmitEvent(new BeforeSaveEvent<>(toSave, dbDoc, collectionName));
+			maybeEmitEvent(new BeforeSaveEvent<>(initialized, dbDoc, collectionName));
 
-			Mono<T> afterInsert = insertDBObject(collectionName, dbDoc, toSave.getClass()).map(id -> {
+			Mono<T> afterInsert = insertDBObject(collectionName, dbDoc, initialized.getClass()).map(id -> {
 
 				T saved = entity.populateIdIfNecessary(id);
-				maybeEmitEvent(new AfterSaveEvent<>(saved, dbDoc, collectionName));
+				maybeEmitEvent(new AfterSaveEvent<>(initialized, dbDoc, collectionName));
 				return saved;
 			});
 
@@ -1388,34 +1387,27 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		Assert.notNull(objectToSave, "Object to save must not be null!");
 		Assert.hasText(collectionName, "Collection name must not be null or empty!");
 
-		MongoPersistentEntity<?> mongoPersistentEntity = getPersistentEntity(objectToSave.getClass());
+		AdaptibleEntity<T> source = operations.forEntity(objectToSave, mongoConverter.getConversionService());
-
-		// No optimistic locking -> simple save
-		if (mongoPersistentEntity == null || !mongoPersistentEntity.hasVersionProperty()) {
-			return doSave(collectionName, objectToSave, this.mongoConverter);
-		}
 
-		return doSaveVersioned(objectToSave, mongoPersistentEntity, collectionName);
+		return source.isVersionedEntity() ? doSaveVersioned(source, collectionName)
+				: doSave(collectionName, objectToSave, this.mongoConverter);
 	}
 
-	private <T> Mono<T> doSaveVersioned(T objectToSave, MongoPersistentEntity<?> entity, String collectionName) {
+	private <T> Mono<T> doSaveVersioned(AdaptibleEntity<T> source, String collectionName) {
 
-		AdaptibleEntity<T> forEntity = operations.forEntity(objectToSave, mongoConverter.getConversionService());
+		if (source.isNew()) {
+			return doInsert(collectionName, source.getBean(), this.mongoConverter);
+		}
 
 		return createMono(collectionName, collection -> {
 
-			Number versionNumber = forEntity.getVersion();
+			// Create query for entity with the id and old version
-
+			Query query = source.getQueryForVersion();
-			// Fresh instance -> initialize version property
-			if (versionNumber == null) {
-				return doInsert(collectionName, objectToSave, mongoConverter);
-			}
-
-			forEntity.assertUpdateableIdIfNotSet();
 
-			Query query = forEntity.getQueryForVersion();
+			// Bump version number
+			T toSave = source.incrementVersion();
 
-			T toSave = forEntity.incrementVersion();
+			source.assertUpdateableIdIfNotSet();
 
 			BeforeConvertEvent<T> event = new BeforeConvertEvent<>(toSave, collectionName);
 			T afterEvent = ReactiveMongoTemplate.this.maybeEmitEvent(event).getSource();
@@ -1426,7 +1418,9 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 			ReactiveMongoTemplate.this.maybeEmitEvent(new BeforeSaveEvent<>(afterEvent, document, collectionName));
 
 			return doUpdate(collectionName, query, mapped.updateWithoutId(), afterEvent.getClass(), false, false)
-					.map(updateResult -> maybeEmitEvent(new AfterSaveEvent<T>(afterEvent, document, collectionName)).getSource());
+					.map(result -> {
+						return maybeEmitEvent(new AfterSaveEvent<T>(afterEvent, document, collectionName)).getSource();
+					});
 		});
 	}
 

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/config/ReactiveAuditingTests.java

@@ -0,0 +1,154 @@
+/*
+ * Copyright 2018 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.config;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.annotation.Version;
+import org.springframework.data.domain.AuditorAware;
+import org.springframework.data.mongodb.core.AuditablePerson;
+import org.springframework.data.mongodb.core.ReactiveMongoOperations;
+import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
+import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
+import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
+import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import com.mongodb.reactivestreams.client.MongoClient;
+import com.mongodb.reactivestreams.client.MongoClients;
+
+/**
+ * Integration test for the auditing support via {@link org.springframework.data.mongodb.core.ReactiveMongoTemplate}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(SpringRunner.class)
+@ContextConfiguration
+public class ReactiveAuditingTests {
+
+	@Autowired ReactiveAuditablePersonRepository auditablePersonRepository;
+	@Autowired AuditorAware<AuditablePerson> auditorAware;
+	@Autowired MongoMappingContext context;
+	@Autowired ReactiveMongoOperations operations;
+
+	@Configuration
+	@EnableMongoAuditing(auditorAwareRef = "auditorProvider")
+	@EnableReactiveMongoRepositories(basePackageClasses = ReactiveAuditingTests.class, considerNestedRepositories = true)
+	static class Config extends AbstractReactiveMongoConfiguration {
+
+		@Override
+		protected String getDatabaseName() {
+			return "database";
+		}
+
+		@Override
+		public MongoClient reactiveMongoClient() {
+			return MongoClients.create();
+		}
+
+		@Bean
+		@SuppressWarnings("unchecked")
+		public AuditorAware<AuditablePerson> auditorProvider() {
+			return mock(AuditorAware.class);
+		}
+	}
+
+	@Test // DATAMONGO-2139, DATAMONGO-2150
+	public void auditingWorksForVersionedEntityWithWrapperVersion() {
+
+		verifyAuditingViaVersionProperty(new VersionedAuditablePerson(), //
+				it -> it.version, //
+				auditablePersonRepository::save, //
+				null, 0L, 1L);
+	}
+
+	@Test // DATAMONGO-2139, DATAMONGO-2150
+	public void auditingWorksForVersionedEntityWithSimpleVersion() {
+
+		verifyAuditingViaVersionProperty(new SimpleVersionedAuditablePerson(), //
+				it -> it.version, //
+				auditablePersonRepository::save, //
+				0L, 1L, 2L);
+	}
+
+	@Test // DATAMONGO-2139, DATAMONGO-2150
+	public void auditingWorksForVersionedEntityWithWrapperVersionOnTemplate() {
+
+		verifyAuditingViaVersionProperty(new VersionedAuditablePerson(), //
+				it -> it.version, //
+				operations::save, //
+				null, 0L, 1L);
+	}
+
+	@Test // DATAMONGO-2139, DATAMONGO-2150
+	public void auditingWorksForVersionedEntityWithSimpleVersionOnTemplate() {
+		verifyAuditingViaVersionProperty(new SimpleVersionedAuditablePerson(), //
+				it -> it.version, //
+				operations::save, //
+				0L, 1L, 2L);
+	}
+
+	private <T extends AuditablePerson> void verifyAuditingViaVersionProperty(T instance,
+			Function<T, Object> versionExtractor, Function<T, Mono<T>> persister, Object... expectedValues) {
+
+		AtomicReference<T> instanceHolder = new AtomicReference<>(instance);
+		MongoPersistentEntity<?> entity = context.getRequiredPersistentEntity(instance.getClass());
+
+		assertThat(versionExtractor.apply(instance)).isEqualTo(expectedValues[0]);
+		assertThat(entity.isNew(instance)).isTrue();
+
+		persister.apply(instanceHolder.get()) //
+				.as(StepVerifier::create).consumeNextWith(actual -> {
+
+					instanceHolder.set(actual);
+
+					assertThat(versionExtractor.apply(actual)).isEqualTo(expectedValues[1]);
+					assertThat(entity.isNew(actual)).isFalse();
+				}).verifyComplete();
+
+		persister.apply(instanceHolder.get()) //
+				.as(StepVerifier::create).consumeNextWith(actual -> {
+
+					instanceHolder.set(actual);
+
+					assertThat(versionExtractor.apply(actual)).isEqualTo(expectedValues[2]);
+					assertThat(entity.isNew(actual)).isFalse();
+				}).verifyComplete();
+	}
+
+	interface ReactiveAuditablePersonRepository extends ReactiveMongoRepository<AuditablePerson, String> {}
+
+	static class VersionedAuditablePerson extends AuditablePerson {
+		@Version Long version;
+	}
+
+	static class SimpleVersionedAuditablePerson extends AuditablePerson {
+		@Version long version;
+	}
+}

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

@@ -774,12 +774,9 @@ public class ReactiveMongoTemplateTests {
 		StepVerifier.create(template.save(map, "maps")).expectNextCount(1).verifyComplete();
 	}
 
-	@Test // DATAMONGO-1444, DATAMONGO-1730
+	@Test(expected = MappingException.class) // DATAMONGO-1444, DATAMONGO-1730, DATAMONGO-2150
 	public void savesMongoPrimitiveObjectCorrectly() {
-
+		template.save(new Object(), "collection");
-		StepVerifier.create(template.save(new Object(), "collection")) //
-				.expectError(MappingException.class) //
-				.verify();
 	}
 
 	@Test // DATAMONGO-1444
@@ -852,12 +849,9 @@ public class ReactiveMongoTemplateTests {
 				.verifyComplete();
 	}
 
-	@Test // DATAMONGO-1444
+	@Test(expected = MappingException.class) // DATAMONGO-1444, DATAMONGO-2150
 	public void rejectsNonJsonStringForSave() {
-
+		template.save("Foobar!", "collection");
-		StepVerifier.create(template.save("Foobar!", "collection")) //
-				.expectError(MappingException.class) //
-				.verify();
 	}
 
 	@Test // DATAMONGO-1444