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.cassandra.sidecar.client;
  
  import java.io.ByteArrayOutputStream;
  import java.io.IOException;
  import java.io.InputStream;
  import java.nio.file.NoSuchFileException;
  import java.nio.file.Path;
  import java.nio.file.Paths;
  import java.util.AbstractMap;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.TimeUnit;
  import java.util.stream.Collectors;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import io.netty.handler.codec.http.HttpHeaderNames;
  import io.netty.handler.codec.http.HttpResponseStatus;
  import io.vertx.core.Future;
  import io.vertx.core.MultiMap;
  import io.vertx.core.Promise;
  import io.vertx.core.Vertx;
  import io.vertx.core.buffer.Buffer;
  import io.vertx.core.file.AsyncFile;
  import io.vertx.core.file.FileSystem;
  import io.vertx.core.file.OpenOptions;
  import io.vertx.core.http.HttpMethod;
  import io.vertx.core.net.KeyCertOptions;
  import io.vertx.core.net.KeyStoreOptions;
  import io.vertx.core.net.OpenSSLEngineOptions;
  import io.vertx.core.net.TrustOptions;
  import io.vertx.ext.web.client.HttpRequest;
  import io.vertx.ext.web.client.WebClient;
  import io.vertx.ext.web.client.WebClientOptions;
  import io.vertx.ext.web.client.predicate.ResponsePredicateResult;
  import io.vertx.ext.web.codec.BodyCodec;
  import org.apache.cassandra.sidecar.common.request.DownloadableRequest;
  import org.apache.cassandra.sidecar.common.request.Request;
  import org.apache.cassandra.sidecar.common.request.UploadableRequest;
  
  import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.AUTH_ROLE;
  import static org.apache.cassandra.sidecar.common.utils.StringUtils.isNullOrEmpty;
  
  /**
   * An {@link HttpClient} implementation that uses vertx's WebClient internally
   */
  public class VertxHttpClient implements HttpClient
  {
      private static final Logger LOGGER = LoggerFactory.getLogger(VertxHttpClient.class);
  
      protected final Vertx vertx;
      protected final WebClient webClient;
      protected final HttpClientConfig config;
  
      public VertxHttpClient(Vertx vertx, HttpClientConfig config)
      {
          WebClientOptions options = new WebClientOptions()
                                     .setMaxPoolSize(config.maxPoolSize())
                                     .setIdleTimeout(config.idleTimeoutMillis())
                                     .setIdleTimeoutUnit(TimeUnit.MILLISECONDS)
                                     .setMaxChunkSize(config.maxChunkSize())
                                     .setReceiveBufferSize(config.receiveBufferSize())
                                     .setConnectTimeout((int) config.timeoutMillis())
                                     .setUserAgent(config.userAgent());
  
          options = applySSLOptions(options, config);
  
          this.vertx = vertx;
          this.webClient = WebClient.create(vertx, options);
          this.config = config;
      }
  
      public VertxHttpClient(Vertx vertx, WebClient webClient, HttpClientConfig config)
      {
          this.vertx = vertx;
          this.webClient = webClient;
          this.config = config;
      }
  
      Vertx vertx()
      {
          return vertx;
      }
  
      /**
       * {@inheritDoc}
       */
      @Override
      public HttpClientConfig config()
      {
          return config;
      }
  
      /**
       * {@inheritDoc}
       */
      @Override
      public CompletableFuture<HttpResponse> execute(SidecarInstance sidecarInstance, RequestContext context)
      {
          if (context.request() instanceof UploadableRequest)
          {
              HttpRequest<Buffer> vertxRequest = vertxRequest(sidecarInstance, context);
              UploadableRequest uploadableRequest = (UploadableRequest) context.request();
              LOGGER.debug("Uploading file={}, for request={}, instance={}",
                           uploadableRequest.filename(), context.request(), sidecarInstance);
              return executeUploadFileInternal(sidecarInstance, vertxRequest, uploadableRequest.filename());
          }
          else
          {
              LOGGER.debug("Executing request={}, on instance={}", context.request(), sidecarInstance);
              return executeInternal(sidecarInstance, context);
          }
      }
  
      protected CompletableFuture<HttpResponse> executeInternal(SidecarInstance sidecarInstance, RequestContext context)
      {
          return vertxRequestFuture(sidecarInstance, context)
                 .compose(vertxRequest -> {
                     Request request = context.request();
                     if (request.requestBody() != null)
                     {
                         return vertxRequest.sendJson(request.requestBody());
                     }
                     return vertxRequest.send();
                 })
                 .map(rawResponse -> {
                     io.vertx.ext.web.client.HttpResponse response = (io.vertx.ext.web.client.HttpResponse) rawResponse;
                     byte[] raw = null;
                     if (response.body() instanceof Buffer)
                     {
                         raw = ((Buffer) response.body()).getBytes();
                     }
  
                     if (response.statusCode() != HttpResponseStatus.OK.code()
                         && context.request() instanceof DownloadableRequest)
                     {
                         DownloadableRequest downloadableRequest = (DownloadableRequest) context.request();
                         // cleanup the destinationPath
                         vertx.fileSystem().deleteBlocking(downloadableRequest.destinationPath());
                     }
  
                     return new HttpResponseImpl(response.statusCode(),
                                                 response.statusMessage(),
                                                 raw,
                                                 mapHeaders(response.headers()),
                                                 sidecarInstance);
                 })
                 .toCompletionStage().toCompletableFuture();
      }
  
      protected Future<HttpRequest> vertxRequestFuture(SidecarInstance sidecarInstance, RequestContext context)
      {
          if (context.request() instanceof DownloadableRequest)
          {
              Promise<HttpRequest> promise = Promise.promise();
              DownloadableRequest downloadableRequest = (DownloadableRequest) context.request();
              String destinationPath = downloadableRequest.destinationPath();
              createTargetDirectory(destinationPath)
              .compose(Void -> openTargetFile(destinationPath))
              .onSuccess(targetFile -> promise.complete(vertxRequest(sidecarInstance, context)
                                                        .as(BodyCodec.pipe(targetFile))))
              .onFailure(promise::fail);
              return promise.future();
          }
          else
          {
              return Future.succeededFuture(vertxRequest(sidecarInstance, context)
                                            .ssl(config.ssl())
                                            .timeout(config.timeoutMillis()));
          }
      }
  
      protected Future<Void> createTargetDirectory(String targetPath)
      {
          Path parent = Paths.get(targetPath).getParent();
          if (null == parent)
          {
              return Future.failedFuture(targetPath + " does not have a parent directory");
          }
  
          String targetDirectory = parent.toString();
          return vertx.fileSystem().mkdirs(targetDirectory)
                      .onFailure(ex -> LOGGER.error("Unable to create folder {}", targetDirectory, ex));
      }
  
      protected Future<AsyncFile> openTargetFile(String targetPath)
      {
          return vertx.fileSystem().open(targetPath, new OpenOptions().setTruncateExisting(true)
                                                                      .setCreate(true));
      }
  
      protected CompletableFuture<HttpResponse> executeUploadFileInternal(SidecarInstance sidecarInstance,
                                                                          HttpRequest<Buffer> vertxRequest,
                                                                          String filename)
      {
          Promise<HttpResponse> promise = Promise.promise();
          // open the local file
          openFileForRead(vertx.fileSystem(), filename)
          .compose(pair -> vertxRequest.ssl(config.ssl())
                                       .putHeader(HttpHeaderNames.CONTENT_LENGTH.toString(),
                                                  String.valueOf(pair.getKey()))
                                       .sendStream(pair.getValue()
                                                       .setReadBufferSize(config.sendReadBufferSize())))
          .onFailure(promise::fail)
          .onSuccess(response -> {
              byte[] raw = response.body() != null ? response.body().getBytes() : null;
              promise.complete(new HttpResponseImpl(response.statusCode(),
                                                    response.statusMessage(),
                                                    raw,
                                                    mapHeaders(response.headers()),
                                                    sidecarInstance
              ));
          });
  
          return promise.future().toCompletionStage().toCompletableFuture();
      }
  
      /**
       * {@inheritDoc}
       */
      @Override
      public CompletableFuture<HttpResponse> stream(SidecarInstance sidecarInstance,
                                                    RequestContext context,
                                                    StreamConsumer streamConsumer)
      {
          Objects.requireNonNull(streamConsumer, "The streamConsumer must be set");
          HttpRequest<Buffer> vertxRequest = vertxRequest(sidecarInstance, context);
  
          LOGGER.debug("Streaming request={}, from instance={}", context.request(), sidecarInstance);
  
          Promise<HttpResponse> promise = Promise.promise();
          vertxRequest.ssl(config.ssl())
                      .timeout(config.timeoutMillis())
                      .expect(response -> {
  
                          // fulfill the promise with the response
                          promise.complete(new HttpResponseImpl(response.statusCode(),
                                                                response.statusMessage(),
                                                                mapHeaders(response.headers()),
                                                                sidecarInstance));
  
                          if (response.statusCode() == HttpResponseStatus.OK.code() ||
                              response.statusCode() == HttpResponseStatus.PARTIAL_CONTENT.code())
                          {
                              return ResponsePredicateResult.success();
                          }
                          else
                          {
                              LOGGER.warn("Unexpected status code received statusCode={}, statusMessage={}",
                                          response.statusCode(), response.statusMessage());
                              return ResponsePredicateResult.failure("Unexpected status code: " +
                                                                     response.statusCode());
                          }
                      })
                      .as(BodyCodec.pipe(new StreamConsumerWriteStream(streamConsumer)))
                      .send()
                      .onFailure(throwable -> {
                          if (!promise.tryFail(throwable))
                          {
                              // the stream has already started, we need to signal the consumer that the
                              // there was a failure mid-stream. This is a non-retryable case
                              streamConsumer.onError(throwable);
                          }
                      });
          return promise.future().toCompletionStage().toCompletableFuture();
      }
  
      /**
       * {@inheritDoc}
       */
      @Override
      public void close()
      {
          webClient.close();
      }
  
      protected HttpRequest<Buffer> vertxRequest(SidecarInstance sidecarInstance, RequestContext context)
      {
          Request request = context.request();
          HttpMethod method = HttpMethod.valueOf(request.method().name());
          HttpRequest<Buffer> vertxRequest = webClient.request(method,
                                                               sidecarInstance.port(),
                                                               sidecarInstance.hostname(),
                                                               request.requestURI());
  
          vertxRequest = applyHeaders(vertxRequest, request.headers());
  
          Map<String, String> customHeaders = context.customHeaders();
          if (customHeaders != null && !customHeaders.isEmpty())
          {
              vertxRequest = applyHeaders(vertxRequest, customHeaders);
          }
  
          return vertxRequest;
      }
  
      protected HttpRequest<Buffer> applyHeaders(HttpRequest<Buffer> vertxRequest, Map<String, String> headers)
      {
          vertxRequest = applyAuthHeader(vertxRequest);
  
          if (headers == null || headers.isEmpty())
              return vertxRequest;
  
          for (Map.Entry<String, String> header : headers.entrySet())
          {
              vertxRequest = vertxRequest.putHeader(header.getKey(), header.getValue());
          }
          return vertxRequest;
      }
  
      protected HttpRequest<Buffer> applyAuthHeader(HttpRequest<Buffer> vertxRequest)
      {
          String cassandraRole = config().cassandraRole();
          if (isNullOrEmpty(cassandraRole))
          {
              return vertxRequest;
          }
  
          return vertxRequest.putHeader(AUTH_ROLE, cassandraRole);
      }
  
      protected Map<String, List<String>> mapHeaders(MultiMap headers)
      {
          if (headers == null)
          {
              return Collections.emptyMap();
          }
          return headers.entries().stream()
                        .filter(entry -> entry.getKey() != null && entry.getValue() != null)
                        .collect(Collectors.toMap(Map.Entry::getKey,
                                                  entry -> Collections.singletonList(entry.getValue())));
      }
  
      protected static WebClientOptions applySSLOptions(WebClientOptions options, HttpClientConfig config)
      {
          if (!config.ssl())
          {
              return options;
          }
  
          options = options.setSsl(true);
  
          if (config.trustStoreInputStream() != null && config.trustStorePassword() != null)
          {
              TrustOptions trustOptions = buildKeyCertOptions(config.trustStoreInputStream(),
                                                              config.trustStorePassword(),
                                                              config.trustStoreType());
              options = options.setTrustOptions(trustOptions);
          }
  
          if (config.keyStoreInputStream() != null && config.keyStorePassword() != null)
          {
              KeyCertOptions keyCertOptions = buildKeyCertOptions(config.keyStoreInputStream(),
                                                                  config.keyStorePassword(),
                                                                  config.keyStoreType());
              options = options.setKeyCertOptions(keyCertOptions);
          }
  
          if (OpenSSLEngineOptions.isAvailable())
          {
              LOGGER.info("Building Sidecar vertx client with OpenSSL");
              options = options.setOpenSslEngineOptions(new OpenSSLEngineOptions());
          }
          else
          {
              LOGGER.warn("OpenSSL not available when building Sidecar vertx client");
          }
  
          return options;
      }
  
      protected static KeyStoreOptions buildKeyCertOptions(InputStream storeStream, String storePass, String storeType)
      {
          try (InputStream inputStream = storeStream)
          {
              byte[] trustBytes = readStore(inputStream);
              return new KeyStoreOptions().setType(storeType)
                                          .setPassword(storePass)
                                          .setValue(Buffer.buffer(trustBytes));
          }
          catch (IOException e)
          {
              throw new RuntimeException("Failed to load default truststore.", e);
          }
      }
  
      protected static byte[] readStore(InputStream storeStream) throws IOException
      {
          byte[] buffer = new byte[1024];
          int temp;
          ByteArrayOutputStream bos = new ByteArrayOutputStream();
          while ((temp = storeStream.read(buffer)) != -1)
          {
              bos.write(buffer, 0, temp);
          }
          return bos.toByteArray();
      }
  
      protected Future<AbstractMap.SimpleEntry<Long, AsyncFile>> openFileForRead(FileSystem fs, String filename)
      {
          Promise<AbstractMap.SimpleEntry<Long, AsyncFile>> promise = Promise.promise();
          fs.exists(filename)
            .compose(exists -> {
                if (!exists)
                {
                    String errMsg = "File '" + filename + "' does not exist";
                    return Future.failedFuture(new NoSuchFileException(errMsg));
                }
                return fs.props(filename);
            })
            .onSuccess(props -> fs.open(filename, new OpenOptions().setWrite(false).setCreate(false).setRead(true))
                                  .onFailure(promise::tryFail)
                                  .onSuccess(asyncFile -> promise.complete(new AbstractMap.SimpleEntry<>(props.size(),
                                                                                                         asyncFile))))
            .onFailure(promise::tryFail);
          return promise.future();
      }
  }
  