diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/MongoItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/MongoItemWriter.java index dbfc95763..0592b3565 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/MongoItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/MongoItemWriter.java @@ -58,6 +58,10 @@ import org.springframework.util.StringUtils; */ public class MongoItemWriter implements ItemWriter, InitializingBean { + public enum Mode { + INSERT, UPSERT, REMOVE; + } + private static final String ID_KEY = "_id"; private MongoOperations template; @@ -66,7 +70,7 @@ public class MongoItemWriter implements ItemWriter, InitializingBean { private String collection; - private boolean delete = false; + private Mode mode = Mode.UPSERT; public MongoItemWriter() { super(); @@ -78,9 +82,19 @@ public class MongoItemWriter implements ItemWriter, InitializingBean { * the data store. If set to false (default), the items will be saved. If set to true, * the items will be removed. * @param delete removal indicator + * @deprecated use {@link MongoItemWriter#setMode(Mode)} */ + @Deprecated public void setDelete(boolean delete) { - this.delete = delete; + this.mode = (delete) ? Mode.REMOVE : Mode.UPSERT; + } + + /** + * Set the operating {@link Mode} to be applied by this writer. + * @param mode the mode to be used. + */ + public void setMode(final Mode mode) { + this.mode = mode; } /** @@ -133,15 +147,31 @@ public class MongoItemWriter implements ItemWriter, InitializingBean { */ protected void doWrite(Chunk chunk) { if (!CollectionUtils.isEmpty(chunk.getItems())) { - if (this.delete) { - delete(chunk); - } - else { - saveOrUpdate(chunk); + switch (this.mode) { + case INSERT: + save(chunk); + break; + case REMOVE: + delete(chunk); + break; + default: + saveOrUpdate(chunk); + break; } } } + private void save(final Chunk chunk) { + final BulkOperations bulkOperations = initBulkOperations(BulkMode.ORDERED, chunk.getItems().get(0)); + final MongoConverter mongoConverter = this.template.getConverter(); + for (final Object item : chunk) { + final Document document = new Document(); + mongoConverter.write(item, document); + bulkOperations.insert(document); + } + bulkOperations.execute(); + } + private void delete(Chunk chunk) { BulkOperations bulkOperations = initBulkOperations(BulkMode.ORDERED, chunk.getItems().get(0)); MongoConverter mongoConverter = this.template.getConverter(); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/builder/MongoItemWriterBuilder.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/builder/MongoItemWriterBuilder.java index efffa4691..3e080fecd 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/builder/MongoItemWriterBuilder.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/builder/MongoItemWriterBuilder.java @@ -17,6 +17,7 @@ package org.springframework.batch.item.data.builder; import org.springframework.batch.item.data.MongoItemWriter; +import org.springframework.batch.item.data.MongoItemWriter.Mode; import org.springframework.data.mongodb.core.MongoOperations; import org.springframework.util.Assert; @@ -33,7 +34,7 @@ public class MongoItemWriterBuilder { private String collection; - private boolean delete = false; + private Mode mode = Mode.UPSERT; /** * Indicates if the items being passed to the writer are to be saved or removed from @@ -42,9 +43,23 @@ public class MongoItemWriterBuilder { * @param delete removal indicator * @return The current instance of the builder * @see MongoItemWriter#setDelete(boolean) + * @deprecated use {@link MongoItemWriterBuilder#mode(Mode)} */ + @Deprecated public MongoItemWriterBuilder delete(boolean delete) { - this.delete = delete; + this.mode = (delete) ? Mode.REMOVE : Mode.UPSERT; + + return this; + } + + /** + * Set the operating {@link Mode} to be applied by this writer. + * @param mode the mode to be used. + * @return The current instance of the builder + * @see MongoItemWriter#setMode(Mode) + */ + public MongoItemWriterBuilder mode(final Mode mode) { + this.mode = mode; return this; } @@ -83,7 +98,7 @@ public class MongoItemWriterBuilder { MongoItemWriter writer = new MongoItemWriter<>(); writer.setTemplate(this.template); - writer.setDelete(this.delete); + writer.setMode(this.mode); writer.setCollection(this.collection); return writer; diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/MongoItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/MongoItemWriterTests.java index eaedf4316..55919f242 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/MongoItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/MongoItemWriterTests.java @@ -26,6 +26,7 @@ import org.mockito.junit.jupiter.MockitoSettings; import org.mockito.quality.Strictness; import org.springframework.batch.item.Chunk; +import org.springframework.batch.item.data.MongoItemWriter.Mode; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.data.mapping.context.MappingContext; import org.springframework.data.mongodb.core.BulkOperations; @@ -297,6 +298,155 @@ class MongoItemWriterTests { } } + // BATCH-4149 + + @Test + void testInsertModeNoTransactionNoCollection() throws Exception { + Chunk items = Chunk.of(new Item("Foo"), new Item("Bar")); + + writer.setMode(Mode.INSERT); + writer.write(items); + + verify(template).bulkOps(any(), any(Class.class)); + verify(bulkOperations, times(2)).insert(any(Object.class)); + } + + @Test + void testInsertModeNoTransactionWithCollection() throws Exception { + Chunk items = Chunk.of(new Item("Foo"), new Item("Bar")); + + writer.setMode(Mode.INSERT); + writer.setCollection("collection"); + + writer.write(items); + + verify(template).bulkOps(any(), eq("collection")); + verify(bulkOperations, times(2)).insert(any(Object.class)); + } + + @Test + void testInsertModeNoTransactionNoItems() throws Exception { + writer.setMode(Mode.INSERT); + writer.write(new Chunk<>()); + + verifyNoInteractions(template); + verifyNoInteractions(bulkOperations); + } + + @Test + void testInsertModeTransactionNoCollection() { + final Chunk items = Chunk.of(new Item("Foo"), new Item("Bar")); + + writer.setMode(Mode.INSERT); + + new TransactionTemplate(transactionManager).execute((TransactionCallback) status -> { + assertDoesNotThrow(() -> writer.write(items)); + return null; + }); + + verify(template).bulkOps(any(), any(Class.class)); + verify(bulkOperations, times(2)).insert(any(Object.class)); + } + + @Test + void testInsertModeTransactionWithCollection() { + final Chunk items = Chunk.of(new Item("Foo"), new Item("Bar")); + + writer.setMode(Mode.INSERT); + writer.setCollection("collection"); + + new TransactionTemplate(transactionManager).execute((TransactionCallback) status -> { + assertDoesNotThrow(() -> writer.write(items)); + return null; + }); + + verify(template).bulkOps(any(), eq("collection")); + verify(bulkOperations, times(2)).insert(any(Object.class)); + } + + @Test + void testInsertModeTransactionFails() { + final Chunk items = Chunk.of(new Item("Foo"), new Item("Bar")); + + writer.setMode(Mode.INSERT); + writer.setCollection("collection"); + + Exception exception = assertThrows(RuntimeException.class, + () -> new TransactionTemplate(transactionManager).execute((TransactionCallback) status -> { + assertDoesNotThrow(() -> writer.write(items)); + throw new RuntimeException("force rollback"); + })); + assertEquals(exception.getMessage(), "force rollback"); + + verifyNoInteractions(template); + verifyNoInteractions(bulkOperations); + } + + @Test + void testInsertModeTransactionReadOnly() { + final Chunk items = Chunk.of(new Item("Foo"), new Item("Bar")); + + writer.setMode(Mode.INSERT); + writer.setCollection("collection"); + + TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager); + transactionTemplate.setReadOnly(true); + transactionTemplate.execute((TransactionCallback) status -> { + assertDoesNotThrow(() -> writer.write(items)); + return null; + }); + + verifyNoInteractions(template); + verifyNoInteractions(bulkOperations); + } + + @Test + void testRemoveModeNoObjectIdNoCollection() throws Exception { + writer.setMode(Mode.REMOVE); + Chunk items = Chunk.of(new Item("Foo"), new Item("Bar")); + + writer.write(items); + + verify(template).bulkOps(any(), any(Class.class)); + verify(bulkOperations, never()).remove(any(Query.class)); + } + + @Test + void testRemoveModeNoObjectIdWithCollection() throws Exception { + writer.setMode(Mode.REMOVE); + Chunk items = Chunk.of(new Item("Foo"), new Item("Bar")); + + writer.setCollection("collection"); + writer.write(items); + + verify(template).bulkOps(any(), eq("collection")); + verify(bulkOperations, never()).remove(any(Query.class)); + } + + @Test + void testRemoveModeNoTransactionNoCollection() throws Exception { + writer.setMode(Mode.REMOVE); + Chunk items = Chunk.of(new Item(1), new Item(2)); + + writer.write(items); + + verify(template).bulkOps(any(), any(Class.class)); + verify(bulkOperations, times(2)).remove(any(Query.class)); + } + + @Test + void testRemoveModeNoTransactionWithCollection() throws Exception { + writer.setMode(Mode.REMOVE); + Chunk items = Chunk.of(new Item(1), new Item(2)); + + writer.setCollection("collection"); + + writer.write(items); + + verify(template).bulkOps(any(), eq("collection")); + verify(bulkOperations, times(2)).remove(any(Query.class)); + } + static class Item { Integer id;