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.eip.it;
  
  import java.util.Arrays;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Set;
  import java.util.concurrent.TimeUnit;
  
  import io.quarkus.test.junit.QuarkusTest;
  import io.restassured.RestAssured;
  import io.restassured.http.ContentType;
  import org.assertj.core.api.Assertions;
  import org.hamcrest.Matchers;
  import org.jboss.logging.Logger;
  import org.junit.jupiter.api.Test;
  
  import static org.awaitility.Awaitility.await;
  
  @QuarkusTest
  class EipTest {
  
      private static final Logger LOG = Logger.getLogger(EipTest.class);
  
      @Test
      public void claimCheckByHeader() {
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .body("Secret")
                  .queryParam("claimCheckId", "foo")
                  .post("/eip/route/claimCheckByHeader")
                  .then()
                  .statusCode(200);
  
          RestAssured.get("/eip/mock/claimCheckByHeader/4/10000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("Bye World,Secret,Hi World,Secret"));
  
      }
  
      @Test
      public void customLoadBalancer() {
          final List<String> messages = Arrays.asList("a", "b", "c", "d");
          for (String msg : messages) {
              RestAssured.given()
                      .contentType(ContentType.TEXT)
                      .body(msg)
                      .post("/eip/route/customLoadBalancer")
                      .then()
                      .statusCode(200);
          }
  
          RestAssured.get("/eip/mock/customLoadBalancer1/2/10000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("a,c"));
  
          RestAssured.get("/eip/mock/customLoadBalancer2/2/10000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("b,d"));
  
      }
  
      @Test
      public void roundRobinLoadBalancer() {
          final List<String> messages = Arrays.asList("a", "b", "c", "d");
          for (String msg : messages) {
              RestAssured.given()
                      .contentType(ContentType.TEXT)
                      .body(msg)
                      .post("/eip/route/roundRobinLoadBalancer")
                      .then()
                      .statusCode(200);
          }
  
          RestAssured.get("/eip/mock/roundRobinLoadBalancer1/2/10000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("a,c"));
  
          RestAssured.get("/eip/mock/roundRobinLoadBalancer2/2/10000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("b,d"));
  
      }
  
      @Test
      public void stickyLoadBalancer() {
          final List<String> messages = Arrays.asList("a", "b", "c", "d");
          int i = 0;
          for (String msg : messages) {
              RestAssured.given()
                      .contentType(ContentType.TEXT)
                      .queryParam("stickyKey", String.valueOf(1 + (i++ % 2)))
                      .body(msg)
                      .post("/eip/route/stickyLoadBalancer")
                      .then()
                      .statusCode(200);
          }
  
          RestAssured.get("/eip/mock/stickyLoadBalancer1/2/10000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("a,c"));
  
          RestAssured.get("/eip/mock/stickyLoadBalancer2/2/10000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("b,d"));
  
      }
  
      @Test
      public void weightedLoadBalancer() {
          final int msgCount = 60;
          LOG.infof("Sending %d messages", msgCount);
          for (int i = 0; i < msgCount; i++) {
              RestAssured.given()
                      .contentType(ContentType.TEXT)
                      .body("hello " + i)
                      .post("/eip/route/weightedLoadBalancer")
                      .then()
                      .statusCode(200);
          }
  
          final int tolerance = 2;
  
          final int expectedCount1 = (msgCount / (EipRoutes.WEIGHTED_1 + EipRoutes.WEIGHTED_2)) * EipRoutes.WEIGHTED_1;
          final int expectedCount2 = msgCount - expectedCount1;
  
          int actualCount1 = RestAssured.get("/eip/mock/weightedLoadBalancer1/" + (expectedCount1 - tolerance) + "+/10000/body")
                  .then()
                  .statusCode(200)
                  .extract().body().asString().split(",").length;
  
          int actualCount2 = RestAssured.get("/eip/mock/weightedLoadBalancer2/" + (expectedCount2 - tolerance) + "+/10000/body")
                  .then()
                  .statusCode(200)
                  .extract().body().asString().split(",").length;
  
          LOG.infof("Expected message distribution %d:%d, got %d:%d", expectedCount1, expectedCount2, actualCount1, actualCount2);
  
          Assertions.assertThat(actualCount1).isBetween(expectedCount1 - tolerance, expectedCount1 + tolerance);
          Assertions.assertThat(actualCount2).isBetween(expectedCount2 - tolerance, expectedCount2 + tolerance);
  
      }
  
      @Test
      public void enrich() {
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .body("Franz")
                  .post("/eip/route/enrich")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("Hello Franz"));
  
      }
  
      @Test
      public void failover() {
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .body("Arthur")
                  .post("/eip/route/failover")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("Hello from failover2 Arthur"));
  
      }
  
      @Test
      public void loop() {
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .body("foo")
                  .post("/eip/route/loop")
                  .then()
                  .statusCode(200);
  
          RestAssured.get("/eip/mock/loop/3/5000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("foo,foo,foo"));
  
      }
  
      @Test
      public void multicast() {
          final List<String> messages = Arrays.asList("a", "b", "c", "d");
          for (String msg : messages) {
              RestAssured.given()
                      .contentType(ContentType.TEXT)
                      .body(msg)
                      .post("/eip/route/multicast")
                      .then()
                      .statusCode(200);
          }
  
          RestAssured.get("/eip/mock/multicast1/4/5000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("a,b,c,d"));
  
          RestAssured.get("/eip/mock/multicast2/4/5000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("a,b,c,d"));
  
          RestAssured.get("/eip/mock/multicast3/4/5000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("a,b,c,d"));
  
      }
  
      @Test
      public void recipientList() {
          final List<String> messages = Arrays.asList("a", "b", "c", "d");
          for (String msg : messages) {
              RestAssured.given()
                      .contentType(ContentType.TEXT)
                      .body(msg)
                      .post("/eip/route/recipientList")
                      .then()
                      .statusCode(200);
          }
  
          RestAssured.get("/eip/mock/recipientList1/4/5000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("a,b,c,d"));
  
          RestAssured.get("/eip/mock/recipientList2/4/5000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("a,b,c,d"));
  
          RestAssured.get("/eip/mock/recipientList3/4/5000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("a,b,c,d"));
  
      }
  
      @Test
      public void removeHeader() {
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .body("baz")
                  .queryParam("headerToKeep", "foo")
                  .queryParam("headerToRemove", "bar")
                  .post("/eip/route/removeHeader")
                  .then()
                  .statusCode(200);
  
          RestAssured.get("/eip/mock/removeHeader/1/5000/header")
                  .then()
                  .statusCode(200)
                  .body(
                          Matchers.allOf(
                                  Matchers.containsString("headerToKeep=foo"),
                                  Matchers.not(Matchers.containsString("headerToRemove"))));
  
      }
  
      @Test
      public void removeHeaders() {
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .body("baz")
                  .queryParam("headerToKeep", "keepFoo")
                  .queryParam("headerToRemove1", "bar1")
                  .queryParam("headerToRemove2", "bar2")
                  .post("/eip/route/removeHeaders")
                  .then()
                  .statusCode(200);
  
          RestAssured.get("/eip/mock/removeHeaders/1/5000/header")
                  .then()
                  .statusCode(200)
                  .body(
                          Matchers.allOf(
                                  Matchers.containsString("headerToKeep=keepFoo"),
                                  Matchers.not(Matchers.containsString("headerToRemove1")),
                                  Matchers.not(Matchers.containsString("headerToRemove2"))));
  
      }
  
      @Test
      public void removeProperty() {
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .body("baz")
                  .queryParam("propertyToKeep", "keep")
                  .queryParam("propertyToRemove", "bar")
                  .post("/eip/route/removeProperty")
                  .then()
                  .statusCode(200);
  
          RestAssured.get("/eip/mock/removeProperty/1/5000/property")
                  .then()
                  .statusCode(200)
                  .body(
                          Matchers.allOf(
                                  Matchers.containsString("propertyToKeep=keep"),
                                  Matchers.not(Matchers.containsString("propertyToRemove"))));
  
      }
  
      @Test
      public void removeProperties() {
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .body("baz")
                  .queryParam("propertyToKeep", "keepProp")
                  .queryParam("propertyToRemove1", "bar1")
                  .queryParam("propertyToRemove2", "bar2")
                  .post("/eip/route/removeProperties")
                  .then()
                  .statusCode(200);
  
          RestAssured.get("/eip/mock/removeProperties/1/5000/property")
                  .then()
                  .statusCode(200)
                  .body(
                          Matchers.allOf(
                                  Matchers.containsString("propertyToKeep=keepProp"),
                                  Matchers.not(Matchers.containsString("propertyToRemove1")),
                                  Matchers.not(Matchers.containsString("propertyToRemove2"))));
  
      }
  
      @Test
      public void routingSlip() {
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .body("message-1")
                  .queryParam("routingSlipHeader", "mock:routingSlip1,mock:routingSlip2")
                  .post("/eip/route/routingSlip")
                  .then()
                  .statusCode(200);
  
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .body("message-2")
                  .queryParam("routingSlipHeader", "mock:routingSlip2,mock:routingSlip3")
                  .post("/eip/route/routingSlip")
                  .then()
                  .statusCode(200);
  
          RestAssured.get("/eip/mock/routingSlip1/1/5000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("message-1"));
  
          RestAssured.get("/eip/mock/routingSlip2/2/5000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("message-1,message-2"));
  
          RestAssured.get("/eip/mock/routingSlip3/1/5000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("message-2"));
  
      }
  
      @Test
      public void sample() {
          final int durationSec = 2;
          LOG.infof("About to sent messages for %d seconds", durationSec);
          final long deadline = System.currentTimeMillis() + (durationSec * 1000); // two seconds ahead
          int i = 0;
          while (System.currentTimeMillis() < deadline) {
              /* Send messages for 2 seconds */
              RestAssured.given()
                      .contentType(ContentType.TEXT)
                      .body("message-" + i++)
                      .post("/eip/route/sample")
                      .then()
                      .statusCode(200);
          }
          LOG.infof("Sent %d messages", i);
          /*
           * We should normally get just 2 samples in 2 seconds using the default sample rate of 1 message per second
           * But timing is hard in programming, let's allow one more
           */
          int overratedSampleUpperBound = durationSec + 1;
          Assertions.assertThat(i).isGreaterThan(overratedSampleUpperBound);
          String[] samples = RestAssured.get("/eip/mock/sample/1+/5000/body")
                  .then()
                  .statusCode(200)
                  .extract()
                  .body().asString().split(",");
          LOG.infof("Got %d samples", samples.length);
          Assertions.assertThat(samples.length).isBetween(1, overratedSampleUpperBound);
      }
  
      @Test
      public void step() {
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .body("Monty")
                  .post("/eip/route/step")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("Hello Monty from step!"));
  
      }
  
      @Test
      public void resequenceStream() {
          final List<String> messages = Arrays.asList("a:2", "b:1", "c:4", "d:3");
          for (String input : messages) {
              String[] message = input.split(":");
              RestAssured.given()
                      .contentType(ContentType.TEXT)
                      .queryParam("seqno", message[1])
                      .body(message[0])
                      .post("/eip/route/resequenceStream")
                      .then()
                      .statusCode(200);
          }
  
          RestAssured.get("/eip/mock/resequenceStream/4/10000/body")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("b,a,d,c"));
  
      }
  
      @Test
      public void threads() {
          final Set<String> threadNames = new HashSet<>();
          final int period = 10000;
          final long deadline = System.currentTimeMillis() + period;
          final int expectedThreadCount = 2;
          do {
              if (System.currentTimeMillis() >= deadline) {
                  Assertions.fail("Have not seen " + expectedThreadCount + " distict thread names within " + period
                          + " ms; thread names seen so far: "
                          + threadNames);
              }
              final String threadName = RestAssured.given()
                      .contentType(ContentType.TEXT)
                      .body("foo")
                      .post("/eip/route/threads")
                      .then()
                      .statusCode(200)
                      .extract().body().asString();
              threadNames.add(threadName);
          } while (threadNames.size() < expectedThreadCount);
  
      }
  
      @Test
      public void throttle() {
          LOG.infof("About to sent 6 messages");
          for (int i = 0; i < 6; i++) {
              RestAssured.given()
                      .contentType(ContentType.TEXT)
                      .body("message-" + i)
                      .post("/eip/routeAsync/throttle")
                      .then()
                      .extract().statusCode();
          }
  
          //wait for two first throttled messages
          await().atMost(EipRoutes.THROTTLE_TIMEOUT, TimeUnit.MILLISECONDS)
                  .pollInterval(500, TimeUnit.MILLISECONDS).until(() -> {
                      String throttleBodies = RestAssured.get("/eip/mock/throttle/body")
                              .then()
                              .statusCode(200)
                              .extract()
                              .body().asString();
                      LOG.infof("Mock throttle received: %s", throttleBodies);
                      return throttleBodies.split(",").length == EipRoutes.THROTTLE_MAXIMUM_REQUEST_COUNT;
                  });
          //wait for remaining messages
          //THROTTLE_TIMEOUT-time taken after read 1/2 messages + THROTTLE_TIMEOUT for 3/4 messages + time for camel to throttle the last 5/6 messages
          long startTime = System.currentTimeMillis();
          await().atMost(3 * EipRoutes.THROTTLE_TIMEOUT, TimeUnit.MILLISECONDS)
                  .pollInterval(1, TimeUnit.SECONDS).until(() -> {
                      String throttleBodies = RestAssured.get("/eip/mock/throttle/body")
                              .then()
                              .statusCode(200)
                              .extract()
                              .body().asString();
                      LOG.infof("Mock throttle received: %s", throttleBodies);
                      return throttleBodies.split(",").length == 3 * EipRoutes.THROTTLE_MAXIMUM_REQUEST_COUNT;
                  });
          long endTime = System.currentTimeMillis();
          //the time of waiting has to be bigger then throttle_Period (assert that it is > throttle_period + 500)
          Assertions.assertThat(endTime - startTime).isGreaterThan(EipRoutes.THROTTLE_TIMEOUT + 500);
      }
  
      @Test
      public void throttleRejectExecution() throws InterruptedException {
          LOG.infof("About to sent 6 messages");
          for (int i = 0; i < 6; i++) {
              RestAssured.given()
                      .contentType(ContentType.TEXT)
                      .body("message-" + i)
                      .post("/eip/routeAsync/throttleRejection")
                      .then()
                      .extract().statusCode();
          }
  
          await().atMost(EipRoutes.THROTTLE_TIMEOUT + 5000, TimeUnit.MILLISECONDS)
                  .pollInterval(1, TimeUnit.SECONDS).until(() -> {
                      String throttleBodies = RestAssured.get("/eip/mock/throttleRejection/body")
                              .then()
                              .statusCode(200)
                              .extract()
                              .body().asString();
                      LOG.infof("Mock throttleRejection received: %s", throttleBodies);
                      return throttleBodies.split(",").length == EipRoutes.THROTTLE_MAXIMUM_REQUEST_COUNT;
                  });
  
          //wait for another 4 messages in error mock
          await().atMost(EipRoutes.THROTTLE_TIMEOUT + 5000, TimeUnit.MILLISECONDS)
                  .pollInterval(1, TimeUnit.SECONDS).until(() -> {
                      String throttleBodies = RestAssured.get("/eip/mock/throttleRejectionError/body")
                              .then()
                              .statusCode(200)
                              .extract()
                              .body().asString();
                      LOG.infof("Mock throttleRejectionError received: %s", throttleBodies);
                      return throttleBodies.split(",").length == 4;
                  });
      }
  
      @Test
      public void tryCatchFinally() {
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .body("foo")
                  .post("/eip/route/tryCatchFinally")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("Handled by finally: Hello foo"));
  
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .body("throw")
                  .post("/eip/route/tryCatchFinally")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("Handled by finally: Caught throw"));
  
      }
  
      @Test
      public void expression() {
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .queryParam("expressionHeader", "true")
                  .body("foo")
                  .post("/eip/route/expression")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("expressionHeader was true"));
  
          RestAssured.given()
                  .contentType(ContentType.TEXT)
                  .queryParam("expressionHeader", "false")
                  .body("foo")
                  .post("/eip/route/expression")
                  .then()
                  .statusCode(200)
                  .body(Matchers.is("expressionHeader was false"));
  
      }
  
  }
  