You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2018/03/05 07:20:26 UTC
[camel] 02/05: CAMEL-12302 : MongoDB3 - add bulk writes operation
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit c08c333b408652a230f7688292add52c04113c9e
Author: Farès Hassak <fa...@gmail.com>
AuthorDate: Sun Mar 4 15:38:37 2018 +0100
CAMEL-12302 : MongoDB3 - add bulk writes operation
---
.../camel/component/mongodb3/MongoDbConstants.java | 2 +-
.../camel/component/mongodb3/MongoDbOperation.java | 5 +-
.../camel/component/mongodb3/MongoDbProducer.java | 24 ++++
.../mongodb3/MongoDbBulkWriteOperationTest.java | 122 +++++++++++++++++++++
4 files changed, 151 insertions(+), 2 deletions(-)
diff --git a/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbConstants.java b/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbConstants.java
index 70fe95c..32a3354 100644
--- a/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbConstants.java
+++ b/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbConstants.java
@@ -41,7 +41,7 @@ public final class MongoDbConstants {
public static final String OID = "CamelMongoOid";
public static final String DISTINCT_QUERY_FIELD = "CamelMongoDbDistinctQueryField";
public static final String ALLOW_DISK_USE = "CamelMongoDbAllowDiskUse";
-
+ public static final String BULK_ORDERED = "CamelMongoDbBulkOrdered";
public static final String MONGO_ID = "_id"; // default id field
private MongoDbConstants() {
diff --git a/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbOperation.java b/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbOperation.java
index 6ddc7d3..6f1b127 100644
--- a/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbOperation.java
+++ b/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbOperation.java
@@ -33,7 +33,10 @@ public enum MongoDbOperation {
// delete operations
remove,
-
+
+ //Bulk operations
+ bulkWrite,
+
// aggregate
aggregate,
diff --git a/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbProducer.java b/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbProducer.java
index ad558a8..ee3ec76 100644
--- a/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbProducer.java
+++ b/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbProducer.java
@@ -24,13 +24,16 @@ import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
+import com.mongodb.bulk.BulkWriteResult;
import com.mongodb.client.AggregateIterable;
import com.mongodb.client.DistinctIterable;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.model.BulkWriteOptions;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.UpdateOptions;
+import com.mongodb.client.model.WriteModel;
import com.mongodb.client.result.DeleteResult;
import com.mongodb.client.result.UpdateResult;
@@ -78,6 +81,7 @@ public class MongoDbProducer extends DefaultProducer {
{
bind(MongoDbOperation.aggregate, createDoAggregate());
+ bind(MongoDbOperation.bulkWrite, createDoBulkWrite());
bind(MongoDbOperation.command, createDoCommand());
bind(MongoDbOperation.count, createDoCount());
bind(MongoDbOperation.findDistinct, createDoDistinct());
@@ -612,4 +616,24 @@ public class MongoDbProducer extends DefaultProducer {
}
};
}
+
+ private Function<Exchange, Object> createDoBulkWrite() {
+ return exchange -> {
+ try {
+ MongoCollection<Document> dbCol = calculateCollection(exchange);
+
+ Boolean ordered = exchange.getIn().getHeader(MongoDbConstants.BULK_ORDERED, Boolean.TRUE, Boolean.class);
+ BulkWriteOptions options = new BulkWriteOptions().ordered(ordered);
+
+ @SuppressWarnings("unchecked")
+ List<WriteModel<Document>> requests = (List<WriteModel<Document>>) exchange.getIn().getMandatoryBody((Class<List<WriteModel<Document>>>)Class.class.cast(List.class));
+
+ BulkWriteResult result = dbCol.bulkWrite(requests, options);
+ return result;
+
+ } catch (InvalidPayloadException e) {
+ throw new CamelMongoDbException("Invalid payload for bulk write", e);
+ }
+ };
+ }
}
diff --git a/components/camel-mongodb3/src/test/java/org/apache/camel/component/mongodb3/MongoDbBulkWriteOperationTest.java b/components/camel-mongodb3/src/test/java/org/apache/camel/component/mongodb3/MongoDbBulkWriteOperationTest.java
new file mode 100644
index 0000000..dc27e81
--- /dev/null
+++ b/components/camel-mongodb3/src/test/java/org/apache/camel/component/mongodb3/MongoDbBulkWriteOperationTest.java
@@ -0,0 +1,122 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.mongodb3;
+
+import java.util.Arrays;
+import java.util.List;
+
+import com.mongodb.bulk.BulkWriteResult;
+import com.mongodb.client.model.DeleteManyModel;
+import com.mongodb.client.model.DeleteOneModel;
+import com.mongodb.client.model.InsertOneModel;
+import com.mongodb.client.model.ReplaceOneModel;
+import com.mongodb.client.model.UpdateManyModel;
+import com.mongodb.client.model.UpdateOneModel;
+import com.mongodb.client.model.WriteModel;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.builder.RouteBuilder;
+import org.bson.Document;
+import org.junit.Test;
+
+public class MongoDbBulkWriteOperationTest extends AbstractMongoDbTest {
+
+ @Test
+ public void testBulkWrite() throws Exception {
+ // Test that the collection has 0 documents in it
+ assertEquals(0, testCollection.count());
+ pumpDataIntoTestCollection();
+ List<WriteModel<Document>> bulkOperations = Arrays
+ .asList(new InsertOneModel<>(new Document("scientist", "Pierre Curie")),
+ new UpdateOneModel<>(new Document("_id", "2"),
+ new Document("$set", new Document("scientist", "Charles Darwin"))),
+ new UpdateManyModel<>(new Document("scientist", "Curie"),
+ new Document("$set", new Document("scientist", "Marie Curie"))),
+ new ReplaceOneModel<>(new Document("_id", "1"), new Document("scientist", "Albert Einstein")),
+ new DeleteOneModel<>(new Document("_id", "3")),
+ new DeleteManyModel<>(new Document("scientist", "Bohr")));
+
+ BulkWriteResult result = template.requestBody("direct:bulkWrite", bulkOperations, BulkWriteResult.class);
+
+ assertNotNull(result);
+ // 1 insert
+ assertEquals("Records inserted should be 2 : ", 1, result.getInsertedCount());
+ // 1 updateOne + 100 updateMany + 1 replaceOne
+ assertEquals("Records matched should be 102 : ", 102, result.getMatchedCount());
+ assertEquals("Records modified should be 102 : ", 102, result.getModifiedCount());
+ // 1 updateOne + 100 updateMany
+ assertEquals("Records deleted should be 101 : ", 101, result.getDeletedCount());
+ }
+
+ @Test
+ public void testOrderedBulkWriteWithError() throws Exception {
+ // Test that the collection has 0 documents in it
+ assertEquals(0, testCollection.count());
+ pumpDataIntoTestCollection();
+
+ List<WriteModel<Document>> bulkOperations = Arrays
+ .asList(new InsertOneModel<>(new Document("scientist", "Pierre Curie")),
+ // this insert failed and bulk stop
+ new InsertOneModel<>(new Document("_id", "1")),
+ new InsertOneModel<>(new Document("scientist", "Descartes")),
+ new UpdateOneModel<>(new Document("_id", "5"), new Document("$set", new Document("scientist", "Marie Curie"))),
+ new DeleteOneModel<>(new Document("_id", "2")));
+
+ try {
+ template.requestBody("direct:bulkWrite", bulkOperations, BulkWriteResult.class);
+ fail("Bulk operation should throw Exception");
+ } catch (CamelExecutionException e) {
+ extractAndAssertCamelMongoDbException(e, "duplicate key error");
+ // count = 1000 records + 1 inserted
+ assertEquals(1001, testCollection.count());
+ }
+ }
+
+ @Test
+ public void testUnorderedBulkWriteWithError() throws Exception {
+ // Test that the collection has 0 documents in it
+ assertEquals(0, testCollection.count());
+ pumpDataIntoTestCollection();
+
+ List<WriteModel<Document>> bulkOperations = Arrays
+ .asList(new InsertOneModel<>(new Document("scientist", "Pierre Curie")),
+ // this insert failed and bulk continue
+ new InsertOneModel<>(new Document("_id", "1")),
+ new InsertOneModel<>(new Document("scientist", "Descartes")),
+ new UpdateOneModel<>(new Document("_id", "5"), new Document("$set", new Document("scientist", "Marie Curie"))),
+ new DeleteOneModel<>(new Document("_id", "2")));
+ try {
+ template.requestBody("direct:unorderedBulkWrite", bulkOperations, BulkWriteResult.class);
+ fail("Bulk operation should throw Exception");
+ } catch (CamelExecutionException e) {
+ extractAndAssertCamelMongoDbException(e, "duplicate key error");
+ // count = 1000 + 2 inserted + 1 deleted
+ assertEquals(1001, testCollection.count());
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() {
+ from("direct:bulkWrite").to("mongodb3:myDb?database={{mongodb.testDb}}&collection={{mongodb.testCollection}}&operation=bulkWrite");
+ from("direct:unorderedBulkWrite").setHeader(MongoDbConstants.BULK_ORDERED).constant(false)
+ .to("mongodb3:myDb?database={{mongodb.testDb}}&collection={{mongodb.testCollection}}&operation=bulkWrite");
+ }
+ };
+ }
+}
--
To stop receiving notification emails like this one, please contact
acosentino@apache.org.