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

DATAMONGO-2290 - Trigger (convert & save) lifecycle events for bulk operations.

Browse Source 
Original pull request: #762.
This commit is contained in:
Christoph Strobl
2019-06-11 13:59:17 +02:00
committed by Jens Schauder
parent c2ea595b8c
commit c09188561c
4 changed files with 153 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/DefaultBulkOperations.java
View File

@@ -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<Document>) it.getModel()).getDocument());
}
if (it.getModel() instanceof ReplaceOneModel) {
maybeInvokeBeforeSaveCallback(it.getSource(), ((ReplaceOneModel<Document>) it.getModel()).getReplacement());
Document target = ((InsertOneModel<Document>) it.getModel()).getDocument();
maybeInvokeBeforeSaveCallback(it.getSource(), target);
} else if (it.getModel() instanceof ReplaceOneModel) {
Document target = ((ReplaceOneModel<Document>) 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<Document>) it.getModel()).getDocument();
maybeEmitEvent(new BeforeSaveEvent<>(it.getSource(), target, collectionName));
} else if (it.getModel() instanceof ReplaceOneModel) {
Document target = ((ReplaceOneModel<Document>) it.getModel()).getReplacement();
maybeEmitEvent(new BeforeSaveEvent<>(it.getSource(), target, collectionName));
}
}
private void maybeEmitAfterSaveEvent(SourceAwareWriteModelHolder it) {
if (it.getModel() instanceof InsertOneModel) {
Document target = ((InsertOneModel<Document>) it.getModel()).getDocument();
maybeEmitEvent(new AfterSaveEvent<>(it.getSource(), target, collectionName));
} else if (it.getModel() instanceof ReplaceOneModel) {
Document target = ((ReplaceOneModel<Document>) it.getModel()).getReplacement();
maybeEmitEvent(new AfterSaveEvent<>(it.getSource(), target, collectionName));
}
}
private <E extends MongoMappingEvent<T>, 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<? extends MongoPersistentEntity<?>> entity;
@NonNull QueryMapper queryMapper;
@NonNull UpdateMapper updateMapper;
ApplicationEventPublisher eventPublisher;
EntityCallbacks entityCallbacks;
}

2
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java
View File

@@ -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);

8
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsIntegrationTests.java
View File

@@ -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);

98
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsUnitTests.java
View File

@@ -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;
}
}
}