content: /*
   * 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.quarkus.component.mongodb.it;
  
  import java.sql.SQLException;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.TimeUnit;
  
  import com.mongodb.client.MongoClient;
  import com.mongodb.client.MongoClients;
  import com.mongodb.client.MongoCollection;
  import com.mongodb.client.MongoDatabase;
  import io.quarkus.test.common.QuarkusTestResource;
  import io.quarkus.test.junit.QuarkusTest;
  import io.restassured.RestAssured;
  import io.restassured.http.ContentType;
  import io.restassured.path.json.JsonPath;
  import org.apache.camel.ServiceStatus;
  import org.apache.camel.quarkus.test.support.mongodb.MongoDbConstants;
  import org.apache.camel.quarkus.test.support.mongodb.MongoDbTestResource;
  import org.apache.camel.util.CollectionHelper;
  import org.bson.Document;
  import org.eclipse.microprofile.config.ConfigProvider;
  import org.jboss.logging.Logger;
  import org.junit.jupiter.api.AfterAll;
  import org.junit.jupiter.api.BeforeAll;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.params.ParameterizedTest;
  import org.junit.jupiter.params.provider.ValueSource;
  
  import static org.awaitility.Awaitility.await;
  import static org.hamcrest.Matchers.is;
  import static org.hamcrest.Matchers.not;
  import static org.hamcrest.Matchers.nullValue;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  @QuarkusTest
  @QuarkusTestResource(MongoDbTestResource.class)
  class MongoDbTest {
      private static final Logger LOG = Logger.getLogger(MongoDbTest.class);
  
      public static final String MSG = "Hello Camel Quarkus Mongo DB";
  
      private static final String COLLECTION_OUTPUT_TYPE_DOCUMENT_LIST = "outputTypeDocumentList";
      private static final String COLLECTION_OUTPUT_TYPE_DOCUMENT = "outputTypeDocument";
  
      private static MongoClient mongoClient;
  
      private static MongoDatabase db;
  
      @BeforeAll
      public static void setUp() throws SQLException {
          final String mongoUrl = "mongodb://" + ConfigProvider.getConfig().getValue("quarkus.mongodb.hosts", String.class);
  
          mongoClient = MongoClients.create(mongoUrl);
          db = mongoClient.getDatabase("test");
      }
  
      @AfterAll
      public static void cleanUp() throws SQLException {
          if (mongoClient != null) {
              mongoClient.close();
          }
      }
  
      @ParameterizedTest
      @ValueSource(strings = { MongoDbConstants.DEFAULT_MONGO_CLIENT_NAME, MongoDbConstants.NAMED_MONGO_CLIENT_NAME })
      public void testMongoDbComponent(String namedClient) {
          // As we will create a route for each client, we use a different collection for each route
          String collectionName = String.format("%sCamelTest", namedClient);
  
          // Write to collection
          RestAssured.given()
                  .contentType(ContentType.JSON)
                  .body("{message:\"" + MSG + "\"}")
                  .header("mongoClientName", namedClient)
                  .post("/mongodb/collection/" + collectionName)
                  .then()
                  .statusCode(201);
  
          // Retrieve from collection
          JsonPath jsonPath = RestAssured
                  .given()
                  .header("mongoClientName", namedClient)
                  .get("/mongodb/collection/" + collectionName)
                  .then()
                  .contentType(ContentType.JSON)
                  .statusCode(200)
                  .extract()
                  .body()
                  .jsonPath();
  
          List<Map<String, String>> documents = jsonPath.get();
          assertEquals(1, documents.size());
  
          Map<String, String> document = documents.get(0);
          assertEquals(MSG, document.get("message"));
      }
  
      @Test
      public void testDynamicOperation() {
          String collectionName = "dynamicCamelTest";
  
          // Write to collection
          RestAssured.given()
                  .contentType(ContentType.JSON)
                  .body("{message:\"" + MSG + "\"}")
                  .header("mongoClientName", MongoDbConstants.DEFAULT_MONGO_CLIENT_NAME)
                  .post("/mongodb/collection/dynamic/" + collectionName)
                  .then()
                  .statusCode(200)
                  .body("message", is(MSG));
  
          //count results with dynamic operation
          RestAssured.given()
                  .contentType(ContentType.JSON)
                  .body("{message:\"" + MSG + "\"}")
                  .header("mongoClientName", MongoDbConstants.DEFAULT_MONGO_CLIENT_NAME)
                  .header("dynamicOperation", "count")
                  .post("/mongodb/collection/dynamic/" + collectionName)
                  .then()
                  .statusCode(200)
                  .body(is("1"));
  
      }
  
      @Test
      public void testTailingConsumer() throws Exception {
          MongoCollection<Document> collection = db.getCollection(MongoDbConstants.COLLECTION_TAILING, Document.class);
  
          for (int i = 1; i <= (10 * MongoDbConstants.CAP_NUMBER); i++) {
              collection.insertOne(new Document("increasing", i).append("string", "value" + i));
  
              //verify continuously
              if (i % MongoDbConstants.CAP_NUMBER == 0) {
                  waitAndResetTailingResults(MongoDbConstants.CAP_NUMBER, "value" + i, MongoDbConstants.COLLECTION_TAILING);
              }
          }
      }
  
      @Test
      public void testPersistentTailingConsumer() throws Exception {
          MongoCollection<Document> collection = db.getCollection(MongoDbConstants.COLLECTION_PERSISTENT_TAILING, Document.class);
  
          for (int i = 1; i <= (3 * MongoDbConstants.CAP_NUMBER); i++) {
              collection.insertOne(new Document("increasing", i).append("string", "value" + i));
  
              //verify continuously
              if (i % MongoDbConstants.CAP_NUMBER == 0) {
                  waitAndResetTailingResults(MongoDbConstants.CAP_NUMBER, "value" + i,
                          MongoDbConstants.COLLECTION_PERSISTENT_TAILING);
              }
          }
  
          //restart route
          routeController(MongoDbConstants.COLLECTION_PERSISTENT_TAILING, "stop", null);
          routeController(MongoDbConstants.COLLECTION_PERSISTENT_TAILING, "status", ServiceStatus.Stopped.name());
          routeController(MongoDbConstants.COLLECTION_PERSISTENT_TAILING, "start", null);
          routeController(MongoDbConstants.COLLECTION_PERSISTENT_TAILING, "status", ServiceStatus.Started.name());
  
          for (int i = (3 * MongoDbConstants.CAP_NUMBER + 1); i <= (6 * MongoDbConstants.CAP_NUMBER); i++) {
              collection.insertOne(new Document("increasing", i).append("string", "value" + i));
  
              //verify continuously
              if (i % MongoDbConstants.CAP_NUMBER == 0) {
                  waitAndResetTailingResults(MongoDbConstants.CAP_NUMBER, "value" + i,
                          MongoDbConstants.COLLECTION_PERSISTENT_TAILING);
              }
          }
      }
  
      private String routeController(String routeId, String operation, String expectedResult) {
          if (expectedResult == null) {
              RestAssured.given()
                      .get("/mongodb/route/" + routeId + "/" + operation)
                      .then().statusCode(204);
          } else {
              await().atMost(5, TimeUnit.SECONDS).until(
                      () -> RestAssured
                              .get("/mongodb/route/" + routeId + "/" + operation)
                              .then()
                              .statusCode(200)
                              .extract().asString(),
                      is(expectedResult));
          }
  
          return null;
      }
  
      @Test
      public void testStreamConsumerWithFilter() throws Exception {
          MongoCollection<Document> collection = db.getCollection(MongoDbConstants.COLLECTION_STREAM_CHANGES, Document.class);
  
          for (int i = 1; i <= 10; i++) {
              collection.insertOne(new Document("increasing", i).append("string", "value" + i));
          }
  
          waitAndResetTailingResults(1, "value2", MongoDbConstants.COLLECTION_STREAM_CHANGES);
      }
  
      @Test
      public void testConvertMapToDocument() {
          RestAssured.given()
                  .contentType(ContentType.JSON)
                  .body(CollectionHelper.mapOf("key1", "val1", "key2", "val2"))
                  .post("/mongodb/convertMapToDocument")
                  .then()
                  .statusCode(200)
                  .body("clazz", is(Document.class.getName()), "key1", is("val1"), "key2", is("val2"));
      }
  
      @Test
      public void testConvertAnyObjectToDocument() {
          RestAssured.given()
                  .body("Hello!")
                  .post("/mongodb/convertAnyObjectToDocument")
                  .then()
                  .statusCode(200)
                  .body("clazz", is(Document.class.getName()), "value", is("Hello!"));
      }
  
      @Test
      public void testOutputTypeDocumentList() throws Exception {
          MongoCollection<Document> collection = db.getCollection(COLLECTION_OUTPUT_TYPE_DOCUMENT_LIST, Document.class);
  
          collection.insertOne(new Document("name", "Sheldon"));
          collection.insertOne(new Document("name", "Irma"));
          collection.insertOne(new Document("name", "Leonard"));
  
          List results = RestAssured
                  .given()
                  .header("mongoClientName", MongoDbConstants.DEFAULT_MONGO_CLIENT_NAME)
                  .get("/mongodb/collectionAsList/" + COLLECTION_OUTPUT_TYPE_DOCUMENT_LIST)
                  .then()
                  .contentType(ContentType.JSON)
                  .statusCode(200)
                  .body("size()", is(3))
                  .extract().as(List.class);
  
          assertTrue(results.contains("Sheldon"));
          assertTrue(results.contains("Irma"));
          assertTrue(results.contains("Leonard"));
  
      }
  
      @Test
      public void testOutputTypeDocument() throws Exception {
          MongoCollection<Document> collection = db.getCollection(COLLECTION_OUTPUT_TYPE_DOCUMENT, Document.class);
  
          collection.insertOne(new Document("name", "Sheldon"));
  
          RestAssured
                  .given()
                  .header("mongoClientName", MongoDbConstants.DEFAULT_MONGO_CLIENT_NAME)
                  .get("/mongodb/searchByNameAsDocument/" + COLLECTION_OUTPUT_TYPE_DOCUMENT + "/Sheldon")
                  .then()
                  .contentType(ContentType.JSON)
                  .statusCode(200)
                  .body("name", is("Sheldon"), "_id", not(nullValue()));
  
      }
  
      private void waitAndResetTailingResults(int expectedSize, String laststring, String resultId) {
          LOG.info("Awaiting results with size " + expectedSize + " and last string " + laststring);
          await().atMost(30, TimeUnit.SECONDS).until(
                  () -> {
                      @SuppressWarnings("unchecked")
                      Map<String, ?> result = (Map<String, ?>) RestAssured
                              .given().contentType(ContentType.JSON)
                              .get("/mongodb/results/" + resultId)
                              .then()
                              .statusCode(200)
                              .extract().as(Map.class);
                      LOG.info("Received results" + result);
                      return result;
                  },
                  m -> ((int) m.get("size") == expectedSize && laststring.equals(((Map) m.get("last")).get("string"))));
  
          RestAssured
                  .given().contentType(ContentType.JSON)
                  .get("/mongodb/resultsReset/" + resultId)
                  .then()
                  .statusCode(204);
      }
  }
  