camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acosent...@apache.org
Subject [camel] 02/05: CAMEL-12302 : MongoDB3 - add bulk writes operation
Date Mon, 05 Mar 2018 07:20:26 GMT
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 <fares.hassak@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.

Mime
View raw message