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.aws2.kinesis.it;
  
  import java.nio.charset.StandardCharsets;
  import java.util.List;
  import java.util.concurrent.TimeUnit;
  
  import io.quarkus.test.common.QuarkusTestResource;
  import io.quarkus.test.junit.QuarkusTest;
  import io.restassured.RestAssured;
  import io.restassured.http.ContentType;
  import org.apache.camel.quarkus.test.support.aws2.Aws2Client;
  import org.apache.camel.quarkus.test.support.aws2.Aws2TestResource;
  import org.apache.camel.quarkus.test.support.aws2.BaseAWs2TestSupport;
  import org.apache.commons.lang3.RandomStringUtils;
  import org.awaitility.Awaitility;
  import org.eclipse.microprofile.config.Config;
  import org.eclipse.microprofile.config.ConfigProvider;
  import org.jboss.logging.Logger;
  import org.junit.jupiter.api.Test;
  import org.testcontainers.containers.localstack.LocalStackContainer.Service;
  import software.amazon.awssdk.core.ResponseInputStream;
  import software.amazon.awssdk.services.s3.S3Client;
  import software.amazon.awssdk.services.s3.model.GetObjectRequest;
  import software.amazon.awssdk.services.s3.model.GetObjectResponse;
  import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
  import software.amazon.awssdk.services.s3.model.ListObjectsResponse;
  import software.amazon.awssdk.services.s3.model.S3Object;
  
  @QuarkusTest
  @QuarkusTestResource(Aws2TestResource.class)
  class Aws2KinesisFirehoseTest extends BaseAWs2TestSupport {
  
      private static final Logger LOG = Logger.getLogger(Aws2KinesisFirehoseTest.class);
  
      @Aws2Client(Service.S3)
      S3Client client;
  
      public Aws2KinesisFirehoseTest() {
          super("/aws2-kinesis-firehose");
      }
  
      @Test
      public void firehose() {
          LOG.info("Testing firehose");
          final String msg = RandomStringUtils.secure().nextAlphanumeric(32 * 1024);
          final String msgPrefix = msg.substring(0, 32);
          final long maxDataBytes = Aws2KinesisTestEnvCustomizer.BUFFERING_SIZE_MB * 1024 * 1024;
          long bytesSent = 0;
          LOG.info("Sending " + Aws2KinesisTestEnvCustomizer.BUFFERING_SIZE_MB + " MB of data to firehose using chunk "
                  + msgPrefix + "...");
          final long deadline = System.currentTimeMillis() + (Aws2KinesisTestEnvCustomizer.BUFFERING_TIME_SEC * 1000);
          while (bytesSent < maxDataBytes && System.currentTimeMillis() < deadline) {
              /*
               * Send at least 1MB of data but do not spend more than a minute by doing it.
               * This is to overpass minimum buffering limits we have set via BufferingHints in the EnvCustomizer
               */
              RestAssured.given() //
                      .contentType(ContentType.TEXT)
                      .body(msg)
                      .post("/aws2-kinesis-firehose/send") //
                      .then()
                      .statusCode(201);
              bytesSent += msg.length();
              LOG.info("Sent " + bytesSent + "/" + maxDataBytes + " bytes of data");
          }
          LOG.info("Sent " + Aws2KinesisTestEnvCustomizer.BUFFERING_SIZE_MB + " MB of data to firehose");
  
          final Config config = ConfigProvider.getConfig();
  
          final String bucketName = config.getValue("aws-kinesis.s3-bucket-name", String.class);
          LOG.infof("Bucket '%s' should contain objects.", bucketName);
          Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
                  () -> {
                      LOG.infof("Reading objects from bucket '%s'", bucketName);
                      final ListObjectsResponse objects = client
                              .listObjects(ListObjectsRequest.builder().bucket(bucketName).build());
                      final List<S3Object> objs = objects.contents();
                      LOG.info("There are  " + objs.size() + " objects in bucket " + bucketName);
                      for (S3Object obj : objs) {
                          LOG.info("Checking object " + obj.key() + " of size " + obj.size());
                          try (ResponseInputStream<GetObjectResponse> o = client
                                  .getObject(GetObjectRequest.builder().bucket(bucketName).key(obj.key()).build())) {
                              final StringBuilder sb = new StringBuilder(msg.length());
                              final byte[] buf = new byte[1024];
                              int len;
                              while ((len = o.read(buf)) >= 0 && sb.length() < msgPrefix.length()) {
                                  sb.append(new String(buf, 0, len, StandardCharsets.UTF_8));
                              }
                              final String foundContent = sb.toString();
                              if (foundContent.startsWith(msgPrefix)) {
                                  /* Yes, this is what we have sent */
                                  LOG.info("Found the expected content in object " + obj.key());
                                  return true;
                              }
                          }
                      }
                      return false;
                  });
  
      }
  
      @Override
      public void testMethodForDefaultCredentialsProvider() {
          LOG.info("Testing testMethodForDefaultCredentialsProvider");
          RestAssured.given() //
                  .contentType(ContentType.TEXT)
                  .body("test")
                  .post("/aws2-kinesis-firehose/send") //
                  .then()
                  .statusCode(201);
          ;
      }
  }
  