🎉 New source: MongoDb ported to java

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/b2e713cd-cc36-4c0a-b5bd-b47cb8a0561e.json

@@ -0,0 +1,8 @@
+{
+  "sourceDefinitionId": "b2e713cd-cc36-4c0a-b5bd-b47cb8a0561e",
+  "name": "MongoDb",
+  "dockerRepository": "airbyte/source-mongodb-new",
+  "dockerImageTag": "0.1.0",
+  "documentationUrl": "https://docs.airbyte.io/integrations/sources/mongodb-new",
+  "icon": "mongodb.svg"
+}

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -458,3 +458,9 @@
   dockerRepository: airbyte/source-amazon-ads
   dockerImageTag: 0.1.0
   documentationUrl: https://docs.airbyte.io/integrations/sources/amazon-ads
+- sourceDefinitionId: b2e713cd-cc36-4c0a-b5bd-b47cb8a0561e
+  name: MongoDb
+  dockerRepository: airbyte/source-mongodb-new
+  dockerImageTag: 0.1.0
+  documentationUrl: https://docs.airbyte.io/integrations/sources/mongodb-new
+  icon: mongodb.svg

airbyte-db/lib/build.gradle

@@ -22,6 +22,9 @@ dependencies {
     // Lombok
     implementation 'org.projectlombok:lombok:1.18.20'
     annotationProcessor('org.projectlombok:lombok:1.18.20')
+
+    // MongoDB
+    implementation group: 'org.mongodb', name: 'mongo-java-driver', version: '3.12.8'
 }
 
 task(newConfigsMigration, dependsOn: 'classes', type: JavaExec) {

airbyte-db/lib/src/main/java/io/airbyte/db/Databases.java

@@ -30,6 +30,7 @@
 import io.airbyte.db.jdbc.JdbcDatabase;
 import io.airbyte.db.jdbc.JdbcStreamingQueryConfiguration;
 import io.airbyte.db.jdbc.StreamingJdbcDatabase;
+import io.airbyte.db.mongodb.MongoDatabase;
 import java.util.Optional;
 import java.util.function.Function;
 import org.apache.commons.dbcp2.BasicDataSource;
@@ -171,4 +172,8 @@ public static BigQueryDatabase createBigQueryDatabase(final String projectId, fi
     return new BigQueryDatabase(projectId, jsonCreds);
   }
 
+  public static MongoDatabase createMongoDatabase(final String connectionString, final String databaseName) {
+    return new MongoDatabase(connectionString, databaseName);
+  }
+
 }

airbyte-db/lib/src/main/java/io/airbyte/db/NoSqlDatabase.java

@@ -0,0 +1,50 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.db;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+public abstract class NoSqlDatabase implements AutoCloseable {
+
+  private JsonNode sourceConfig;
+  private JsonNode databaseConfig;
+
+  public JsonNode getSourceConfig() {
+    return sourceConfig;
+  }
+
+  public void setSourceConfig(JsonNode sourceConfig) {
+    this.sourceConfig = sourceConfig;
+  }
+
+  public JsonNode getDatabaseConfig() {
+    return databaseConfig;
+  }
+
+  public void setDatabaseConfig(JsonNode databaseConfig) {
+    this.databaseConfig = databaseConfig;
+  }
+
+}

airbyte-db/lib/src/main/java/io/airbyte/db/SqlDatabase.java

@@ -27,29 +27,10 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import java.util.stream.Stream;
 
-public abstract class SqlDatabase implements AutoCloseable {
-
-  private JsonNode sourceConfig;
-  private JsonNode databaseConfig;
+public abstract class SqlDatabase extends NoSqlDatabase {
 
   public abstract void execute(String sql) throws Exception;
 
   public abstract Stream<JsonNode> query(String sql, String... params) throws Exception;
 
-  public JsonNode getSourceConfig() {
-    return sourceConfig;
-  }
-
-  public void setSourceConfig(JsonNode sourceConfig) {
-    this.sourceConfig = sourceConfig;
-  }
-
-  public JsonNode getDatabaseConfig() {
-    return databaseConfig;
-  }
-
-  public void setDatabaseConfig(JsonNode databaseConfig) {
-    this.databaseConfig = databaseConfig;
-  }
-
 }

airbyte-db/lib/src/main/java/io/airbyte/db/mongodb/MongoDatabase.java

@@ -0,0 +1,141 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.db.mongodb;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.annotations.VisibleForTesting;
+import com.mongodb.MongoClient;
+import com.mongodb.MongoClientURI;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoCursor;
+import com.mongodb.client.MongoIterable;
+import io.airbyte.commons.functional.CheckedFunction;
+import io.airbyte.db.NoSqlDatabase;
+import java.util.List;
+import java.util.Optional;
+import java.util.Spliterator;
+import java.util.Spliterators;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+import org.bson.BsonDocument;
+import org.bson.Document;
+import org.bson.conversions.Bson;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MongoDatabase extends NoSqlDatabase {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(MongoDatabase.class);
+  private static final int BATCH_SIZE = 1000;
+
+  private final MongoClientURI mongoClientURI;
+  private final String databaseName;
+
+  private MongoClient mongoClient;
+
+  public MongoDatabase(String uri, String databaseName) {
+    try {
+      mongoClientURI = new MongoClientURI(uri);
+      mongoClient = new MongoClient(mongoClientURI);
+      this.databaseName = databaseName;
+    } catch (Exception e) {
+      LOGGER.error(e.getMessage());
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public void close() throws Exception {
+    mongoClient.close();
+  }
+
+  public com.mongodb.client.MongoDatabase getDatabase() {
+    return mongoClient.getDatabase(databaseName);
+  }
+
+  public MongoIterable<String> getCollectionNames() {
+    return getDatabase().listCollectionNames();
+  }
+
+  public MongoCollection<Document> getCollection(String collectionName) {
+    return getDatabase().getCollection(collectionName);
+  }
+
+  @VisibleForTesting
+  public MongoCollection<Document> createCollection(String name) {
+    getDatabase().createCollection(name);
+    return getDatabase().getCollection(name);
+  }
+
+  @VisibleForTesting
+  public String getName() {
+    return getDatabase().getName();
+  }
+
+  public Stream<JsonNode> read(String collectionName, List<String> columnNames, Optional<Bson> filter) {
+    try {
+      final MongoCollection<Document> collection = getDatabase().getCollection(collectionName);
+      final MongoCursor<Document> cursor = collection
+          .find(filter.orElse(new BsonDocument()))
+          .batchSize(BATCH_SIZE)
+          .cursor();
+
+      return getStream(cursor, (document) -> MongoUtils.toJsonNode(document, columnNames))
+          .onClose(() -> {
+            try {
+              cursor.close();
+            } catch (Exception e) {
+              throw new RuntimeException();
+            }
+          });
+
+    } catch (Exception e) {
+      LOGGER.error("Exception attempting to read data from collection: ", collectionName, e.getMessage());
+      throw new RuntimeException(e);
+    }
+  }
+
+  private Stream<JsonNode> getStream(MongoCursor<Document> cursor, CheckedFunction<Document, JsonNode, Exception> mapper) {
+    return StreamSupport.stream(new Spliterators.AbstractSpliterator<>(Long.MAX_VALUE, Spliterator.ORDERED) {
+
+      @Override
+      public boolean tryAdvance(Consumer<? super JsonNode> action) {
+        try {
+          Document document = cursor.tryNext();
+          if (document == null) {
+            return false;
+          }
+          action.accept(mapper.apply(document));
+          return true;
+        } catch (Exception e) {
+          throw new RuntimeException(e);
+        }
+      }
+
+    }, false);
+  }
+
+}