camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acosent...@apache.org
Subject [camel] 01/05: CAMEL-12302 : add bulk writes operation
Date Mon, 05 Mar 2018 07:20:25 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 89c0d87607cc5bb4c5626b4dec682467179f7d6a
Author: Far├Ęs Hassak <fares.hassak@gmail.com>
AuthorDate: Wed Feb 28 22:23:53 2018 +0100

    CAMEL-12302 : add bulk writes operation
---
 .../camel/component/mongodb/MongoDbComponent.java  |   3 +-
 .../camel/component/mongodb/MongoDbConstants.java  |   1 +
 .../camel/component/mongodb/MongoDbOperation.java  |   3 +
 .../camel/component/mongodb/MongoDbProducer.java   |  24 ++++
 .../mongodb/MongoDbBulkWriteOperationTest.java     | 123 +++++++++++++++++++++
 5 files changed, 153 insertions(+), 1 deletion(-)

diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbComponent.java
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbComponent.java
index e2e8fee..dee4340 100644
--- a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbComponent.java
+++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbComponent.java
@@ -33,7 +33,8 @@ public class MongoDbComponent extends UriEndpointComponent {
     
     public static final Set<MongoDbOperation> WRITE_OPERATIONS = 
             new HashSet<MongoDbOperation>(Arrays.asList(MongoDbOperation.insert, MongoDbOperation.save,

-                    MongoDbOperation.update, MongoDbOperation.remove));
+                    MongoDbOperation.update, MongoDbOperation.remove, MongoDbOperation.bulkWrite));
+    
     private static final Logger LOG = LoggerFactory.getLogger(MongoDbComponent.class);
 
     public MongoDbComponent() {
diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbConstants.java
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbConstants.java
index ba3e71b..1bd0e98 100644
--- a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbConstants.java
+++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbConstants.java
@@ -40,6 +40,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";
 
     private MongoDbConstants() {
     }
diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbOperation.java
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbOperation.java
index 00228cd..4af38ff 100644
--- a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbOperation.java
+++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbOperation.java
@@ -34,6 +34,9 @@ public enum MongoDbOperation {
     // delete operations
     remove, 
     
+    //Bulk operations
+    bulkWrite,
+    
     // aggregate
     aggregate,
     
diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbProducer.java
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbProducer.java
index 63568e7..d6944a3 100644
--- a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbProducer.java
+++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbProducer.java
@@ -27,12 +27,15 @@ import java.util.stream.Collectors;
 import com.mongodb.BasicDBList;
 import com.mongodb.BasicDBObject;
 import com.mongodb.DBObject;
+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.UpdateOptions;
+import com.mongodb.client.model.WriteModel;
 import com.mongodb.client.result.DeleteResult;
 import com.mongodb.client.result.UpdateResult;
 
@@ -69,6 +72,7 @@ public class MongoDbProducer extends DefaultProducer {
         bind(MongoDbOperation.remove, createDoRemove());
         bind(MongoDbOperation.save, createDoSave());
         bind(MongoDbOperation.update, createDoUpdate());
+        bind(MongoDbOperation.bulkWrite, createDoBulkWrite());
     }
 
     public MongoDbProducer(MongoDbEndpoint endpoint) {
@@ -588,4 +592,24 @@ public class MongoDbProducer extends DefaultProducer {
         };
     }
 
+    private Function<Exchange, Object> createDoBulkWrite() {
+        return exchange -> {
+            try {
+                MongoCollection<BasicDBObject> 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<BasicDBObject>> requests = exchange.getIn().getMandatoryBody((Class<List<WriteModel<BasicDBObject>>>)(Class<?>)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-mongodb/src/test/java/org/apache/camel/component/mongodb/MongoDbBulkWriteOperationTest.java
b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/MongoDbBulkWriteOperationTest.java
new file mode 100644
index 0000000..82f69a4
--- /dev/null
+++ b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/MongoDbBulkWriteOperationTest.java
@@ -0,0 +1,123 @@
+/**
+ * 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.mongodb;
+
+import java.util.Arrays;
+import java.util.List;
+
+import com.mongodb.BasicDBObject;
+import com.mongodb.DBObject;
+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.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<DBObject>> bulkOperations = Arrays
+            .asList(new InsertOneModel<>(new BasicDBObject("scientist", "Pierre Curie")),
+                    new UpdateOneModel<>(new BasicDBObject("_id", "2"), 
+                                         new BasicDBObject("$set", new BasicDBObject("scientist",
"Charles Darwin"))),
+                    new UpdateManyModel<>(new BasicDBObject("scientist", "Curie"),

+                            new BasicDBObject("$set", new BasicDBObject("scientist", "Marie
Curie"))),
+                    new ReplaceOneModel<>(new BasicDBObject("_id", "1"), new BasicDBObject("scientist",
"Albert Einstein")),
+                    new DeleteOneModel<>(new BasicDBObject("_id", "3")),
+                    new DeleteManyModel<>(new BasicDBObject("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<DBObject>> bulkOperations = Arrays
+            .asList(new InsertOneModel<>(new BasicDBObject("scientist", "Pierre Curie")),
+                    // this insert failed and bulk stop
+                    new InsertOneModel<>(new BasicDBObject("_id", "1")), 
+                    new InsertOneModel<>(new BasicDBObject("scientist", "Descartes")),
+                    new UpdateOneModel<>(new BasicDBObject("_id", "5"), new BasicDBObject("$set",
new BasicDBObject("scientist", "Marie Curie"))),
+                    new DeleteOneModel<>(new BasicDBObject("_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<DBObject>> bulkOperations = Arrays
+            .asList(new InsertOneModel<>(new BasicDBObject("scientist", "Pierre Curie")),
+                    // this insert failed and bulk continue
+                    new InsertOneModel<>(new BasicDBObject("_id", "1")),
+                    new InsertOneModel<>(new BasicDBObject("scientist", "Descartes")),
+                    new UpdateOneModel<>(new BasicDBObject("_id", "5"), new BasicDBObject("$set",
new BasicDBObject("scientist", "Marie Curie"))),
+                    new DeleteOneModel<>(new BasicDBObject("_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("mongodb:myDb?database={{mongodb.testDb}}&collection={{mongodb.testCollection}}&operation=bulkWrite");
+                from("direct:unorderedBulkWrite").setHeader(MongoDbConstants.BULK_ORDERED).constant(false)
+                    .to("mongodb: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