Add MongoDB Connection helper class

Signed-off-by: Dinu John <[email protected]>

data-prepper-plugins/mongodb/src/main/java/org/opensearch/dataprepper/plugins/mongo/client/MongoDBConnection.java

@@ -0,0 +1,48 @@
+package org.opensearch.dataprepper.plugins.mongo.client;
+
+import com.mongodb.ConnectionString;
+import com.mongodb.MongoClientSettings;
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoClients;
+import org.opensearch.dataprepper.plugins.mongo.configuration.MongoDBSourceConfig;
+import org.opensearch.dataprepper.plugins.truststore.TrustStoreProvider;
+
+import java.io.File;
+import java.util.Objects;
+
+public class MongoDBConnection {
+    private static final String MONGO_CONNECTION_STRING_TEMPLATE = "mongodb://%s:%s@%s:%s/?replicaSet=rs0&directConnection=true&readpreference=%s&ssl=%s&tlsAllowInvalidHostnames=%s&directConnection=%s";
+
+    public static MongoClient getMongoClient(final MongoDBSourceConfig sourceConfig) {
+
+        final String connectionString = getConnectionString(sourceConfig);
+
+        final MongoClientSettings.Builder settingBuilder = MongoClientSettings.builder()
+                .applyConnectionString(new ConnectionString(connectionString));
+
+        if (Objects.nonNull(sourceConfig.getTrustStoreFilePath())) {
+            final File truststoreFilePath = new File(sourceConfig.getTrustStoreFilePath());
+            settingBuilder.applyToSslSettings(builder -> {
+                builder.enabled(sourceConfig.getInsecure());
+                builder.invalidHostNameAllowed(sourceConfig.getSslInsecureDisableVerification());
+                builder.context(TrustStoreProvider.createSSLContext(truststoreFilePath.toPath(),
+                        sourceConfig.getTrustStorePassword()));
+            });
+        }
+
+        return MongoClients.create(settingBuilder.build());
+    }
+
+    private static String getConnectionString(final MongoDBSourceConfig sourceConfig) {
+        final String username = sourceConfig.getCredentialsConfig().getUsername();
+        final String password = sourceConfig.getCredentialsConfig().getPassword();
+        final String hostname = sourceConfig.getHostname();
+        final int port = sourceConfig.getPort();
+        final String ssl = sourceConfig.getInsecure().toString();
+        final String invalidHostAllowed = sourceConfig.getSslInsecureDisableVerification().toString();
+        final String readPreference = sourceConfig.getReadPreference();
+        final String directionConnection = sourceConfig.getDirectConnection().toString();
+        return String.format(MONGO_CONNECTION_STRING_TEMPLATE, username, password, hostname, port,
+                readPreference, ssl, invalidHostAllowed, directionConnection);
+    }
+}

data-prepper-plugins/mongodb/src/test/java/org/opensearch/dataprepper/plugins/mongo/client/MongoDBConnectionTest.java

@@ -0,0 +1,75 @@
+package org.opensearch.dataprepper.plugins.mongo.client;
+
+import com.mongodb.client.MongoClient;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.opensearch.dataprepper.plugins.mongo.configuration.MongoDBSourceConfig;
+import org.opensearch.dataprepper.plugins.truststore.TrustStoreProvider;
+
+import javax.net.ssl.SSLContext;
+import java.nio.file.Path;
+import java.util.Random;
+import java.util.UUID;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+public class MongoDBConnectionTest {
+    @Mock
+    private MongoDBSourceConfig mongoDBSourceConfig;
+
+    @Mock
+    private MongoDBSourceConfig.CredentialsConfig credentialsConfig;
+
+    private final Random random = new Random();
+
+    @BeforeEach
+    void setUp() {
+        when(mongoDBSourceConfig.getCredentialsConfig()).thenReturn(credentialsConfig);
+        when(credentialsConfig.getUsername()).thenReturn(UUID.randomUUID().toString());
+        when(credentialsConfig.getPassword()).thenReturn(UUID.randomUUID().toString());
+        when(mongoDBSourceConfig.getHostname()).thenReturn(UUID.randomUUID().toString());
+        when(mongoDBSourceConfig.getPort()).thenReturn(getRandomInteger());
+        when(mongoDBSourceConfig.getInsecure()).thenReturn(getRandomBoolean());
+        when(mongoDBSourceConfig.getSslInsecureDisableVerification()).thenReturn(getRandomBoolean());
+        when(mongoDBSourceConfig.getReadPreference()).thenReturn("secondaryPreferred");
+    }
+
+    @Test
+    public void getMongoClient() {
+        final MongoClient mongoClient = MongoDBConnection.getMongoClient(mongoDBSourceConfig);
+        assertThat(mongoClient, is(notNullValue()));
+    }
+
+    @Test
+    public void getMongoClientWithTLS() {
+        when(mongoDBSourceConfig.getTrustStoreFilePath()).thenReturn(UUID.randomUUID().toString());
+        when(mongoDBSourceConfig.getTrustStorePassword()).thenReturn(UUID.randomUUID().toString());
+        final Path path = mock(Path.class);
+        final SSLContext sslContext = mock(SSLContext.class);
+        try (MockedStatic<TrustStoreProvider> trustStoreProviderMockedStatic = mockStatic(TrustStoreProvider.class)) {
+            trustStoreProviderMockedStatic.when(() -> TrustStoreProvider.createSSLContext(path,
+                            UUID.randomUUID().toString()))
+                    .thenReturn(sslContext);
+            final MongoClient mongoClient = MongoDBConnection.getMongoClient(mongoDBSourceConfig);
+            assertThat(mongoClient, is(notNullValue()));
+        }
+    }
+
+    private Boolean getRandomBoolean() {
+        return Math.random() < 0.5;
+    }
+
+    private int getRandomInteger() {
+        return random.nextInt(10000);
+    }
+}