diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/BulkOperationException.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/BulkOperationException.java new file mode 100644 index 000000000..033861afa --- /dev/null +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/BulkOperationException.java @@ -0,0 +1,50 @@ +/* + * Copyright 2015 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; + +import java.util.List; + +import org.springframework.dao.DataAccessException; + +import com.mongodb.BulkWriteError; +import com.mongodb.BulkWriteException; +import com.mongodb.BulkWriteResult; + +/** + * Is thrown when errors occur during bulk operations. + * + * @author Tobias Trelle + */ +public class BulkOperationException extends DataAccessException { + private static final long serialVersionUID = 73929601661154421L; + private final List errors; + private final BulkWriteResult result; + + public BulkOperationException(String msg, BulkWriteException e) { + super(msg, e); + this.errors = e.getWriteErrors(); + this.result = e.getWriteResult(); + } + + public List getErrors() { + return errors; + } + + public BulkWriteResult getResult() { + return result; + } + +} diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/BulkOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/BulkOperations.java new file mode 100644 index 000000000..57d33d22d --- /dev/null +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/BulkOperations.java @@ -0,0 +1,167 @@ +/* + * Copyright 2015 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.core; + +import java.util.List; + +import org.springframework.data.mongodb.BulkOperationException; +import org.springframework.data.mongodb.core.query.Query; +import org.springframework.data.mongodb.core.query.Update; +import org.springframework.data.mongodb.util.Tuple; + +import com.mongodb.BulkWriteResult; +import com.mongodb.WriteConcern; + +/** + * Bulk operations for insert/update/remove actions on a collection. + *

+ * These bulks operation are available since MongoDB 2.6 and make use of low level bulk commands on the protocol level. + *

+ * This interface defines a fluent API add multiple single operations or list of similar operations in sequence. + * + * @author Tobias Trelle + */ +public interface BulkOperations { + + /** Mode for bulk operation. */ + public enum BulkMode { + + /** Perform bulk operations in sequence. The first error will cancel processing. */ + ORDERED, + + /** Perform bulk operations in parallel. Processing will continue on errors. */ + UNORDERED + }; + /** + * Add a single insert to the bulk operation. + * + * @param documents List of documents to insert. + * + * @return The bulk operation. + * + * @throws BulkOperationException if an error occured during bulk processing. + */ + BulkOperations insert(Object documents); + + /** + * Add a list of inserts to the bulk operation. + * + * @param documents List of documents to insert. + * + * @return The bulk operation. + * + * @throws BulkOperationException if an error occured during bulk processing. + */ + BulkOperations insert(List documents); + + /** + * Add a single update to the bulk operation. For the update request, only the first matching document is updated. + * + * @param query Update criteria. + * @param update Update operation to perform. + * + * @return The bulk operation. + */ + BulkOperations updateOne(Query query, Update update); + + /** + * Add a list of updates to the bulk operation. For each update request, only the first matching document is updated. + * + * @param updates Update operations to perform. + * + * @return The bulk operation. + */ + BulkOperations updateOne(List> updates); + + /** + * Add a single update to the bulk operation. For the update request, all matching documents are updated. + * + * @param query Update criteria. + * @param update Update operation to perform. + * + * @return The bulk operation. + */ + BulkOperations updateMulti(Query query, Update update); + + /** + * Add a list of updates to the bulk operation. For each update request, all matching documents are updated. + * + * @param updates Update operations to perform. + * + * @return The bulk operation. + */ + BulkOperations updateMulti(List> updates); + + /** + * Add a single upsert to the bulk operation. An upsert is an update if the set of matching documents is not empty, + * else an insert. + * + * @param query Update criteria. + * @param update Update operation to perform. + * + * @return The bulk operation. + */ + BulkOperations upsert(Query query, Update update); + + /** + * Add a list of upserts to the bulk operation. An upsert is an update if the set of matching documents is not empty, else an + * insert. + * + * @param updates Updates/insert operations to perform. + * + * @return The bulk operation. + */ + BulkOperations upsert(List> updates); + + /** + * Add a single remove operation to the bulk operation. + * + * @param remove operations to perform. + * + * @return The bulk operation. + */ + BulkOperations remove(Query remove); + + /** + * Add a list of remove operations to the bulk operation. + * + * @param remove operations to perform. + * + * @return The bulk operation. + */ + BulkOperations remove(List removes); + + /** + * Execute all bulk operations using the default write concern. + * + * @return Result of the bulk operation providing counters for inserts/updates etc. + * + * @throws BulkOperationException if errors occur during the exection of the bulk operations. + */ + BulkWriteResult executeBulk(); + + /** + * Execute all bulk operations using the given write concern. + * + * @param writeConcern Write concern to use. + * + * @return Result of the bulk operation providing counters for inserts/updates etc. + * + * @throws BulkOperationException if errors occur during the exection of the bulk operations. + */ + BulkWriteResult executeBulk(WriteConcern writeConcern); + +} 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 new file mode 100644 index 000000000..85c7521c2 --- /dev/null +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/DefaultBulkOperations.java @@ -0,0 +1,205 @@ +/* + * Copyright 2015 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.core; + +import java.util.List; + +import org.springframework.data.mongodb.BulkOperationException; +import org.springframework.data.mongodb.core.query.Query; +import org.springframework.data.mongodb.core.query.Update; +import org.springframework.data.mongodb.util.Tuple; +import org.springframework.util.Assert; + +import com.mongodb.BulkWriteException; +import com.mongodb.BulkWriteOperation; +import com.mongodb.BulkWriteResult; +import com.mongodb.DBCollection; +import com.mongodb.DBObject; +import com.mongodb.WriteConcern; + +/** + * Default implementation for {@link BulkOperations}. + * + * @author Tobias Trelle + */ +public class DefaultBulkOperations implements BulkOperations { + + private final MongoOperations mongoOperations; + private final BulkMode bulkMode; + private final String collectionName; + private final WriteConcern writeConcernDefault; + private BulkWriteOperation bulk; + + /** + * Creates a new {@link DefaultBulkOperations}. + * + * @param mongoOperations The underlying Mongo operations. + * @param bulkMode The bulk mode (ordered or unordered). + * @param collectionName Name of the collection to work on. + * @param writeConcernDefault The default write concern for all executions. + */ + public DefaultBulkOperations(MongoOperations mongoOperations, BulkMode bulkMode, String collectionName, + WriteConcern writeConcernDefault) { + + Assert.notNull(mongoOperations, "MongoOperations must not be null!"); + Assert.notNull(collectionName, "Collection name can not be null!"); + + this.mongoOperations = mongoOperations; + this.bulkMode = bulkMode; + this.collectionName = collectionName; + this.writeConcernDefault = writeConcernDefault; + initBulkOp(); + } + + @Override + public BulkOperations insert(Object document) { + bulk.insert((DBObject) mongoOperations.getConverter().convertToMongoType(document)); + + return this; + } + + @Override + public BulkOperations insert(List documents) { + for (Object document : documents) { + insert(document); + } + + return this; + } + + @Override + public BulkOperations updateOne(Query query, Update update) { + return update(query, update, false, false); + } + + @Override + public BulkOperations updateOne(List> updates) { + for (Tuple update : updates) { + update(update.getFirst(), update.getSecond(), false, false); + } + return this; + } + + @Override + public BulkOperations updateMulti(Query query, Update update) { + return update(query, update, false, true); + } + + @Override + public BulkOperations updateMulti(List> updates) { + for (Tuple update : updates) { + update(update.getFirst(), update.getSecond(), false, true); + } + return this; + } + + @Override + public BulkOperations upsert(Query query, Update update) { + return update(query, update, true, true); + } + + @Override + public BulkOperations upsert(List> updates) { + for (Tuple update : updates) { + upsert(update.getFirst(), update.getSecond()); + } + return this; + } + + /** + * Performs update and upsert bulk operations. + * + * @param query Criteria to match documents. + * @param update Update to perform. + * @param upsert Upsert flag. + * @param multi Multi update flag. + * @param writeConcern The write concern to use. + * + * @return Self reference. + * + * @throws BulkOperationException if an error occured during bulk processing. + */ + protected BulkOperations update(Query query, Update pdate, boolean upsert, boolean multi) { + if (upsert) { + if (multi) { + bulk.find(query.getQueryObject()).upsert().update(pdate.getUpdateObject()); + } else { + bulk.find(query.getQueryObject()).upsert().updateOne(pdate.getUpdateObject()); + } + } else { + if (multi) { + bulk.find(query.getQueryObject()).update(pdate.getUpdateObject()); + } else { + bulk.find(query.getQueryObject()).updateOne(pdate.getUpdateObject()); + } + } + + return this; + } + + @Override + public BulkOperations remove(Query remove) { + bulk.find(remove.getQueryObject()).remove(); + + return this; + } + + @Override + public BulkOperations remove(List removes) { + for (Query query : removes) { + remove(query); + } + return this; + } + + @Override + public BulkWriteResult executeBulk() { + return executeBulk(writeConcernDefault); + } + + @Override + public BulkWriteResult executeBulk(WriteConcern writeConcern) { + try { + if (writeConcern != null) { + return bulk.execute(writeConcern); + } else { + return bulk.execute(); + } + } catch (BulkWriteException e) { + throw new BulkOperationException("Bulk operation did not complete", e); + } finally { + // reset bulk for future use + initBulkOp(); + } + + } + + private void initBulkOp() { + this.bulk = createBulkOperation(bulkMode, mongoOperations.getCollection(collectionName)); + } + + private BulkWriteOperation createBulkOperation(BulkMode mode, DBCollection collection) { + switch (mode) { + case ORDERED: + return collection.initializeOrderedBulkOperation(); + case UNORDERED: + return collection.initializeUnorderedBulkOperation(); + default: + return null; + } + } + +} diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java index d166bf1f1..63d1ab6f1 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java @@ -20,6 +20,7 @@ import java.util.List; import java.util.Set; import org.springframework.data.geo.GeoResults; +import org.springframework.data.mongodb.core.BulkOperations.BulkMode; import org.springframework.data.mongodb.core.aggregation.Aggregation; import org.springframework.data.mongodb.core.aggregation.AggregationResults; import org.springframework.data.mongodb.core.aggregation.TypedAggregation; @@ -292,6 +293,26 @@ public interface MongoOperations { */ ScriptOperations scriptOps(); + /** + * Returns the bulk operations. + * + * @param bulkMode Mode to use for bulk operations (ordered, unordered). + * @param collectionsName Name of the collection to work on. +v * + * @return index operations on the named collection + */ + BulkOperations bulkOps(BulkMode bulkMode, String collectionsName); + + /** + * Returns the bulk operations. + * + * @param bulkMode Mode to use for bulk operations (ordered, unordered). + * @param entityClass Name of the entity class. + * + * @return index operations on the named collection associated with the given entity class + */ + BulkOperations bulkOps(BulkMode bulkMode, Class entityClass); + /** * Query for a list of objects of type T from the collection used by the entity class. *

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 271286722..a30e57b10 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 @@ -60,6 +60,7 @@ import org.springframework.data.mapping.context.MappingContext; import org.springframework.data.mapping.model.ConvertingPropertyAccessor; import org.springframework.data.mapping.model.MappingException; import org.springframework.data.mongodb.MongoDbFactory; +import org.springframework.data.mongodb.core.BulkOperations.BulkMode; import org.springframework.data.mongodb.core.aggregation.Aggregation; import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext; import org.springframework.data.mongodb.core.aggregation.AggregationResults; @@ -544,6 +545,14 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware { return new DefaultIndexOperations(this, determineCollectionName(entityClass)); } + public BulkOperations bulkOps(BulkMode bulkMode, String collectionName) { + return new DefaultBulkOperations(this, bulkMode, collectionName, writeConcern); + } + + public BulkOperations bulkOps(BulkMode bulkMode, Class entityClass) { + return new DefaultBulkOperations(this, bulkMode, determineCollectionName(entityClass), writeConcern); + } + /* * (non-Javadoc) * @see org.springframework.data.mongodb.core.MongoOperations#scriptOps() diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/Tuple.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/Tuple.java new file mode 100644 index 000000000..422afc1bd --- /dev/null +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/Tuple.java @@ -0,0 +1,44 @@ +/* + * Copyright 2015 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.util; + +/** + * A tuple of things. + * + * @author Tobias Trelle + * + * @param Type of the first thing. + * @param Type of the second thing. + */ +public class Tuple { + + private T t; + private S s; + + public Tuple(T t, S s) { + this.t = t; + this.s = s; + } + + public T getFirst() { + return t; + } + + public S getSecond() { + return s; + } + +} 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 new file mode 100644 index 000000000..bffad502e --- /dev/null +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsIntegrationTests.java @@ -0,0 +1,319 @@ +/* + * Copyright 2015 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.core; + +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.CoreMatchers.notNullValue; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.mongodb.BulkOperationException; +import org.springframework.data.mongodb.core.BulkOperations.BulkMode; +import org.springframework.data.mongodb.core.query.Criteria; +import org.springframework.data.mongodb.core.query.Query; +import org.springframework.data.mongodb.core.query.Update; +import org.springframework.data.mongodb.util.Tuple; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import com.mongodb.BasicDBObject; +import com.mongodb.BulkWriteResult; +import com.mongodb.DBCollection; +import com.mongodb.DBObject; +import com.mongodb.WriteConcern; + +/** + * Integration tests for {@link DefaultBulkOperations}. + * + * @author Tobias Trelle + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration("classpath:infrastructure.xml") +public class DefaultBulkOperationsIntegrationTests { + + private static final String COLLECTION_NAME = "bulk_ops"; + + @Autowired + private MongoTemplate template; + + private DBCollection collection; + private BulkOperations bulkOps; + + @Before + public void setUp() { + + this.collection = this.template.getDb().getCollection(COLLECTION_NAME); + this.collection.remove(new BasicDBObject()); + } + + @Test + public void insertOrdered() { + // given + List documents = Arrays.asList(newDoc("1"), newDoc("2")); + bulkOps = createBulkOps(BulkMode.ORDERED); + + // when + int n = bulkOps.insert(documents).executeBulk().getInsertedCount(); + + // then + assertThat(n, is(2)); + } + + @Test + public void insertOrderedFails() { + // given + List documents = Arrays.asList(newDoc("1"), newDoc("1"), newDoc("2")); + bulkOps = createBulkOps(BulkMode.ORDERED); + + // when + try { + bulkOps.insert(documents).executeBulk(); + fail(); + } catch (BulkOperationException e) { + // then + assertThat(e.getResult().getInsertedCount(), is(1)); // fails after first error + assertThat(e.getErrors(), notNullValue()); + assertThat(e.getErrors().size(), is(1)); + } + } + + @Test + public void insertUnOrdered() { + // given + List documents = Arrays.asList(newDoc("1"), newDoc("2")); + bulkOps = createBulkOps(BulkMode.UNORDERED); + + // when + int n = bulkOps.insert(documents).executeBulk().getInsertedCount(); + + // then + assertThat(n, is(2)); + } + + @Test + public void insertUnOrderedContinuesOnError() { + // given + List documents = Arrays.asList(newDoc("1"), newDoc("1"), newDoc("2")); + bulkOps = createBulkOps(BulkMode.UNORDERED); + + // when + try { + bulkOps.insert(documents).executeBulk(); + fail(); + } catch (BulkOperationException e) { + // then + assertThat(e.getResult().getInsertedCount(), is(2)); // two docs were inserted + assertThat(e.getErrors(), notNullValue()); + assertThat(e.getErrors().size(), is(1)); + } + + } + + @Test + public void upsertDoesUpdate() { + // given + bulkOps = createBulkOps(BulkMode.ORDERED); + insertSomeDocuments(); + + // when + BulkWriteResult result = bulkOps.upsert(where("value", "value1"), set("value", "value2")).executeBulk(); + + // then + assertThat(result, notNullValue()); + assertThat(result.getMatchedCount(), is(2)); + assertThat(result.getModifiedCount(), is(2)); + assertThat(result.getInsertedCount(), is(0)); + assertThat(result.getUpserts(), notNullValue()); + assertThat(result.getUpserts().size(), is(0)); + } + + @Test + public void upsertDoesInsert() { + // given + bulkOps = createBulkOps(BulkMode.ORDERED); + + // when + BulkWriteResult result = bulkOps.upsert(where("_id", "1"), set("value", "v1")).executeBulk(); + + // then + assertThat(result, notNullValue()); + assertThat(result.getMatchedCount(), is(0)); + assertThat(result.getModifiedCount(), is(0)); + assertThat(result.getUpserts(), notNullValue()); + assertThat(result.getUpserts().size(), is(1)); + } + + @Test + public void updateOneOrdered() { + testUpdate(BulkMode.ORDERED, false, 2); + } + + @Test + public void updateMultiOrdered() { + testUpdate(BulkMode.ORDERED, true, 4); + } + + @Test + public void updateOneUnOrdered() { + testUpdate(BulkMode.UNORDERED, false, 2); + } + + @Test + public void updateMultiUnOrdered() { + testUpdate(BulkMode.UNORDERED, true, 4); + } + + @Test + public void removeOrdered() { + testRemove(BulkMode.ORDERED); + } + + @Test + public void removeUnordered() { + testRemove(BulkMode.UNORDERED); + } + + /** + * If working on the same set of documents, only an ordered bulk operation will yield predictable results. + */ + @Test + public void mixedBulkOrdered() { + // given + bulkOps = createBulkOps(BulkMode.ORDERED); + + // when + BulkWriteResult result = bulkOps.insert(newDoc("1", "v1")).updateOne(where("_id", "1"), set("value", "v2")) + .remove(where("value", "v2")).executeBulk(); + + // then + assertThat(result, notNullValue()); + assertThat(result.getInsertedCount(), is(1)); + assertThat(result.getModifiedCount(), is(1)); + assertThat(result.getRemovedCount(), is(1)); + } + + /** + * If working on the same set of documents, only an ordered bulk operation will yield predictable results. + */ + @Test + public void mixedBulkOrderedWithList() { + // given + bulkOps = createBulkOps(BulkMode.ORDERED); + List inserts = Arrays.asList(newDoc("1", "v1"), newDoc("2", "v2"), newDoc("3", "v2")); + List> updates = new ArrayList>(); + updates.add(new Tuple(where("value", "v2"), set("value", "v3"))); + List removes = Arrays.asList(where("_id", "1")); + + // when + BulkWriteResult result = bulkOps.insert(inserts).updateMulti(updates).remove(removes).executeBulk(); + + // then + assertThat(result, notNullValue()); + assertThat(result.getInsertedCount(), is(3)); + assertThat(result.getModifiedCount(), is(2)); + assertThat(result.getRemovedCount(), is(1)); + } + + private void testUpdate(BulkMode mode, boolean multi, int expectedUpdates) { + // given + bulkOps = createBulkOps(mode); + insertSomeDocuments(); + List> updates = new ArrayList>(); + updates.add(new Tuple(where("value", "value1"), set("value", "value3"))); + updates.add(new Tuple(where("value", "value2"), set("value", "value4"))); + + // when + int n; + if (multi) { + n = bulkOps.updateMulti(updates).executeBulk().getModifiedCount(); + } else { + n = bulkOps.updateOne(updates).executeBulk().getModifiedCount(); + } + + // then + assertThat(n, is(expectedUpdates)); + } + + private void testRemove(BulkMode mode) { + // given + bulkOps = createBulkOps(mode); + insertSomeDocuments(); + List removes = Arrays.asList(where("_id", "1"), where("value", "value2")); + + // when + int n = bulkOps.remove(removes).executeBulk().getRemovedCount(); + + // then + assertThat(n, is(3)); + } + + private BulkOperations createBulkOps(BulkMode mode) { + return new DefaultBulkOperations(template, mode, COLLECTION_NAME, WriteConcern.ACKNOWLEDGED); + } + + private void insertSomeDocuments() { + final DBCollection coll = template.getCollection(COLLECTION_NAME); + + coll.insert(rawDoc("1", "value1")); + coll.insert(rawDoc("2", "value1")); + coll.insert(rawDoc("3", "value2")); + coll.insert(rawDoc("4", "value2")); + } + + private static BaseDoc newDoc(String id) { + final BaseDoc doc = new BaseDoc(); + doc.id = id; + + return doc; + } + + private static BaseDoc newDoc(String id, String value) { + final BaseDoc doc = newDoc(id); + + doc.value = value; + + return doc; + } + + private static Query where(String field, String value) { + return new Query().addCriteria(Criteria.where(field).is(value)); + } + + private static Update set(String field, String value) { + Update u = new Update(); + u.set(field, value); + + return u; + } + + private static DBObject rawDoc(String id, String value) { + final DBObject o = new BasicDBObject(); + + o.put("_id", id); + o.put("value", value); + + return o; + } + +}