From c09188561cb528f0878871cde1a778687353623b Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Tue, 11 Jun 2019 13:59:17 +0200 Subject: [PATCH] DATAMONGO-2290 - Trigger (convert & save) lifecycle events for bulk operations. Original pull request: #762. --- .../mongodb/core/DefaultBulkOperations.java | 61 +++++++++++- .../data/mongodb/core/MongoTemplate.java | 2 +- ...DefaultBulkOperationsIntegrationTests.java | 8 +- .../core/DefaultBulkOperationsUnitTests.java | 98 +++++++++++++++++-- 4 files changed, 153 insertions(+), 16 deletions(-) diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/DefaultBulkOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/DefaultBulkOperations.java index 30a1ce08b..17f22cae8 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/DefaultBulkOperations.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/DefaultBulkOperations.java @@ -26,13 +26,18 @@ import java.util.stream.Collectors; import org.bson.Document; import org.bson.conversions.Bson; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.dao.support.PersistenceExceptionTranslator; import org.springframework.data.mapping.callback.EntityCallbacks; import org.springframework.data.mongodb.core.convert.QueryMapper; import org.springframework.data.mongodb.core.convert.UpdateMapper; import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity; +import org.springframework.data.mongodb.core.mapping.event.AfterSaveEvent; import org.springframework.data.mongodb.core.mapping.event.BeforeConvertCallback; +import org.springframework.data.mongodb.core.mapping.event.BeforeConvertEvent; import org.springframework.data.mongodb.core.mapping.event.BeforeSaveCallback; +import org.springframework.data.mongodb.core.mapping.event.BeforeSaveEvent; +import org.springframework.data.mongodb.core.mapping.event.MongoMappingEvent; import org.springframework.data.mongodb.core.query.Collation; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; @@ -115,6 +120,7 @@ class DefaultBulkOperations implements BulkOperations { Assert.notNull(document, "Document must not be null!"); + maybeEmitEvent(new BeforeConvertEvent<>(document, collectionName)); Object source = maybeInvokeBeforeConvertCallback(document); addModel(source, new InsertOneModel<>(getMappedObject(source))); @@ -266,6 +272,7 @@ class DefaultBulkOperations implements BulkOperations { replaceOptions.upsert(options.isUpsert()); query.getCollation().map(Collation::toMongoCollation).ifPresent(replaceOptions::collation); + maybeEmitEvent(new BeforeConvertEvent<>(replacement, collectionName)); Object source = maybeInvokeBeforeConvertCallback(replacement); addModel(source, new ReplaceOneModel<>(getMappedQuery(query.getQueryObject()), getMappedObject(source), replaceOptions)); @@ -282,19 +289,27 @@ class DefaultBulkOperations implements BulkOperations { try { - return mongoOperations.execute(collectionName, collection -> { + com.mongodb.bulk.BulkWriteResult result = mongoOperations.execute(collectionName, collection -> { return collection.bulkWrite(models.stream().map(it -> { + maybeEmitBeforeSaveEvent(it); + if (it.getModel() instanceof InsertOneModel) { - maybeInvokeBeforeSaveCallback(it.getSource(), ((InsertOneModel) it.getModel()).getDocument()); - } - if (it.getModel() instanceof ReplaceOneModel) { - maybeInvokeBeforeSaveCallback(it.getSource(), ((ReplaceOneModel) it.getModel()).getReplacement()); + + Document target = ((InsertOneModel) it.getModel()).getDocument(); + maybeInvokeBeforeSaveCallback(it.getSource(), target); + } else if (it.getModel() instanceof ReplaceOneModel) { + + Document target = ((ReplaceOneModel) it.getModel()).getReplacement(); + maybeInvokeBeforeSaveCallback(it.getSource(), target); } return mapWriteModel(it.getModel()); }).collect(Collectors.toList()), bulkOptions); }); + + models.stream().forEach(this::maybeEmitAfterSaveEvent); + return result; } finally { this.bulkOptions = getBulkWriteOptions(bulkOperationContext.getBulkMode()); } @@ -386,6 +401,41 @@ class DefaultBulkOperations implements BulkOperations { models.add(new SourceAwareWriteModelHolder(source, model)); } + private void maybeEmitBeforeSaveEvent(SourceAwareWriteModelHolder it) { + + if (it.getModel() instanceof InsertOneModel) { + + Document target = ((InsertOneModel) it.getModel()).getDocument(); + maybeEmitEvent(new BeforeSaveEvent<>(it.getSource(), target, collectionName)); + } else if (it.getModel() instanceof ReplaceOneModel) { + + Document target = ((ReplaceOneModel) it.getModel()).getReplacement(); + maybeEmitEvent(new BeforeSaveEvent<>(it.getSource(), target, collectionName)); + } + } + + private void maybeEmitAfterSaveEvent(SourceAwareWriteModelHolder it) { + + if (it.getModel() instanceof InsertOneModel) { + + Document target = ((InsertOneModel) it.getModel()).getDocument(); + maybeEmitEvent(new AfterSaveEvent<>(it.getSource(), target, collectionName)); + } else if (it.getModel() instanceof ReplaceOneModel) { + + Document target = ((ReplaceOneModel) it.getModel()).getReplacement(); + maybeEmitEvent(new AfterSaveEvent<>(it.getSource(), target, collectionName)); + } + } + + private , T> E maybeEmitEvent(E event) { + + if (null != bulkOperationContext.getEventPublisher()) { + bulkOperationContext.getEventPublisher().publishEvent(event); + } + + return event; + } + private Object maybeInvokeBeforeConvertCallback(Object value) { if (bulkOperationContext.getEntityCallbacks() == null) { @@ -434,6 +484,7 @@ class DefaultBulkOperations implements BulkOperations { @NonNull Optional> entity; @NonNull QueryMapper queryMapper; @NonNull UpdateMapper updateMapper; + ApplicationEventPublisher eventPublisher; EntityCallbacks entityCallbacks; } 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 eba273b9f..c71eada66 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 @@ -778,7 +778,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware, Assert.hasText(collectionName, "Collection name must not be null or empty!"); DefaultBulkOperations operations = new DefaultBulkOperations(this, collectionName, new BulkOperationContext(mode, - Optional.ofNullable(getPersistentEntity(entityType)), queryMapper, updateMapper, entityCallbacks)); + Optional.ofNullable(getPersistentEntity(entityType)), queryMapper, updateMapper, eventPublisher, entityCallbacks)); operations.setExceptionTranslator(exceptionTranslator); operations.setDefaultWriteConcern(writeConcern); diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsIntegrationTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsIntegrationTests.java index a252291c4..6933fb894 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsIntegrationTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsIntegrationTests.java @@ -72,19 +72,19 @@ public class DefaultBulkOperationsIntegrationTests { @Test(expected = IllegalArgumentException.class) // DATAMONGO-934 public void rejectsNullMongoOperations() { new DefaultBulkOperations(null, COLLECTION_NAME, - new BulkOperationContext(BulkMode.ORDERED, Optional.empty(), null, null, null)); + new BulkOperationContext(BulkMode.ORDERED, Optional.empty(), null, null, null, null)); } @Test(expected = IllegalArgumentException.class) // DATAMONGO-934 public void rejectsNullCollectionName() { new DefaultBulkOperations(operations, null, - new BulkOperationContext(BulkMode.ORDERED, Optional.empty(), null, null, null)); + new BulkOperationContext(BulkMode.ORDERED, Optional.empty(), null, null, null, null)); } @Test(expected = IllegalArgumentException.class) // DATAMONGO-934 public void rejectsEmptyCollectionName() { - new DefaultBulkOperations(operations, "", new BulkOperationContext(BulkMode.ORDERED, Optional.empty(), null, null, null)); + new DefaultBulkOperations(operations, "", new BulkOperationContext(BulkMode.ORDERED, Optional.empty(), null, null, null, null)); } @Test // DATAMONGO-934 @@ -341,7 +341,7 @@ public class DefaultBulkOperationsIntegrationTests { : Optional.empty(); BulkOperationContext bulkOperationContext = new BulkOperationContext(mode, entity, - new QueryMapper(operations.getConverter()), new UpdateMapper(operations.getConverter()), null); + new QueryMapper(operations.getConverter()), new UpdateMapper(operations.getConverter()), null, null); DefaultBulkOperations bulkOps = new DefaultBulkOperations(operations, COLLECTION_NAME, bulkOperationContext); bulkOps.setDefaultWriteConcern(WriteConcern.ACKNOWLEDGED); diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsUnitTests.java index 9115e3d09..edf120159 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsUnitTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsUnitTests.java @@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.*; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.*; import static org.mockito.Mockito.any; +import static org.mockito.Mockito.anyList; import static org.mockito.Mockito.eq; import static org.springframework.data.mongodb.core.query.Criteria.*; import static org.springframework.data.mongodb.core.query.Query.*; @@ -26,6 +27,8 @@ import static org.springframework.data.mongodb.core.query.Query.*; import java.util.List; import java.util.Optional; +import org.bson.BsonDocument; +import org.bson.BsonString; import org.bson.Document; import org.junit.Before; import org.junit.Test; @@ -34,6 +37,9 @@ import org.mockito.ArgumentCaptor; import org.mockito.Captor; import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.dao.DataAccessException; +import org.springframework.dao.support.PersistenceExceptionTranslator; import org.springframework.data.annotation.Id; import org.springframework.data.mapping.callback.EntityCallbacks; import org.springframework.data.mongodb.MongoDbFactory; @@ -46,12 +52,17 @@ import org.springframework.data.mongodb.core.convert.QueryMapper; import org.springframework.data.mongodb.core.convert.UpdateMapper; import org.springframework.data.mongodb.core.mapping.Field; import org.springframework.data.mongodb.core.mapping.MongoMappingContext; +import org.springframework.data.mongodb.core.mapping.event.AfterSaveEvent; import org.springframework.data.mongodb.core.mapping.event.BeforeConvertCallback; +import org.springframework.data.mongodb.core.mapping.event.BeforeConvertEvent; import org.springframework.data.mongodb.core.mapping.event.BeforeSaveCallback; +import org.springframework.data.mongodb.core.mapping.event.BeforeSaveEvent; import org.springframework.data.mongodb.core.query.BasicQuery; import org.springframework.data.mongodb.core.query.Collation; import org.springframework.data.mongodb.core.query.Update; +import com.mongodb.MongoWriteException; +import com.mongodb.WriteError; import com.mongodb.client.MongoCollection; import com.mongodb.client.MongoDatabase; import com.mongodb.client.model.DeleteManyModel; @@ -85,20 +96,20 @@ public class DefaultBulkOperationsUnitTests { @Before public void setUp() { + when(factory.getDb()).thenReturn(database); + when(factory.getExceptionTranslator()).thenReturn(new NullExceptionTranslator()); + when(database.getCollection(anyString(), eq(Document.class))).thenReturn(collection); + mappingContext = new MongoMappingContext(); mappingContext.afterPropertiesSet(); converter = new MappingMongoConverter(dbRefResolver, mappingContext); - template = new MongoTemplate(factory, converter); - when(factory.getDb()).thenReturn(database); - when(database.getCollection(anyString(), eq(Document.class))).thenReturn(collection); - ops = new DefaultBulkOperations(template, "collection-1", new BulkOperationContext(BulkMode.ORDERED, Optional.of(mappingContext.getPersistentEntity(SomeDomainType.class)), new QueryMapper(converter), - new UpdateMapper(converter), null)); + new UpdateMapper(converter), null, null)); } @Test // DATAMONGO-1518 @@ -199,7 +210,7 @@ public class DefaultBulkOperationsUnitTests { ops = new DefaultBulkOperations(template, "collection-1", new BulkOperationContext(BulkMode.ORDERED, Optional.of(mappingContext.getPersistentEntity(Person.class)), - new QueryMapper(converter), new UpdateMapper(converter), + new QueryMapper(converter), new UpdateMapper(converter), null, EntityCallbacks.create(beforeConvertCallback, beforeSaveCallback))); Person entity = new Person("init"); @@ -219,6 +230,73 @@ public class DefaultBulkOperationsUnitTests { assertThat(updateModel.getDocument()).containsEntry("firstName", "before-save"); } + @Test // DATAMONGO-2290 + public void bulkReplaceOneEmitsEventsCorrectly() { + + ApplicationEventPublisher eventPublisher = mock(ApplicationEventPublisher.class); + + ops = new DefaultBulkOperations(template, "collection-1", + new BulkOperationContext(BulkMode.ORDERED, Optional.of(mappingContext.getPersistentEntity(Person.class)), + new QueryMapper(converter), new UpdateMapper(converter), eventPublisher, null)); + + ops.replaceOne(query(where("firstName").is("danerys")), new SomeDomainType()); + + verify(eventPublisher).publishEvent(any(BeforeConvertEvent.class)); + verify(eventPublisher, never()).publishEvent(any(BeforeSaveEvent.class)); + verify(eventPublisher, never()).publishEvent(any(AfterSaveEvent.class)); + + ops.execute(); + + verify(eventPublisher).publishEvent(any(BeforeSaveEvent.class)); + verify(eventPublisher).publishEvent(any(AfterSaveEvent.class)); + } + + @Test // DATAMONGO-2290 + public void bulkInsertEmitsEventsCorrectly() { + + ApplicationEventPublisher eventPublisher = mock(ApplicationEventPublisher.class); + + ops = new DefaultBulkOperations(template, "collection-1", + new BulkOperationContext(BulkMode.ORDERED, Optional.of(mappingContext.getPersistentEntity(Person.class)), + new QueryMapper(converter), new UpdateMapper(converter), eventPublisher, null)); + + ops.insert(new SomeDomainType()); + + verify(eventPublisher).publishEvent(any(BeforeConvertEvent.class)); + verify(eventPublisher, never()).publishEvent(any(BeforeSaveEvent.class)); + verify(eventPublisher, never()).publishEvent(any(AfterSaveEvent.class)); + + ops.execute(); + + verify(eventPublisher).publishEvent(any(BeforeSaveEvent.class)); + verify(eventPublisher).publishEvent(any(AfterSaveEvent.class)); + } + + @Test + public void noAfterSaveEventOnFailure() { + + ApplicationEventPublisher eventPublisher = mock(ApplicationEventPublisher.class); + when(collection.bulkWrite(anyList(), any())).thenThrow(new MongoWriteException( + new WriteError(89, "NetworkTimeout", new BsonDocument("hi", new BsonString("there!"))), null)); + + ops = new DefaultBulkOperations(template, "collection-1", + new BulkOperationContext(BulkMode.ORDERED, Optional.of(mappingContext.getPersistentEntity(Person.class)), + new QueryMapper(converter), new UpdateMapper(converter), eventPublisher, null)); + + ops.insert(new SomeDomainType()); + + verify(eventPublisher).publishEvent(any(BeforeConvertEvent.class)); + + try { + ops.execute(); + } catch (MongoWriteException expected) { + + } + + verify(eventPublisher).publishEvent(any(BeforeSaveEvent.class)); + verify(eventPublisher, never()).publishEvent(any(AfterSaveEvent.class)); + } + class SomeDomainType { @Id String id; @@ -248,4 +326,12 @@ public class DefaultBulkOperationsUnitTests { return new Person("before-save"); } } + + static class NullExceptionTranslator implements PersistenceExceptionTranslator { + + @Override + public DataAccessException translateExceptionIfPossible(RuntimeException ex) { + return null; + } + } }