#ifndef THIRD_PARTY_GBMCWEB_HTTP_HTTP_CLIENT_H_
#define THIRD_PARTY_GBMCWEB_HTTP_HTTP_CLIENT_H_

/*
// Copyright (c) 2020 Intel Corporation
//
// Licensed 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.
*/

#include <openssl/err.h>
#include <openssl/ssl.h>  // NOLINT

#include <chrono>  // NOLINT
#include <cstdint>
#include <cstdlib>
#include <functional>
#include <memory>
#include <optional>
#include <string>
#include <tuple>
#include <unordered_map>
#include <utility>
#include <vector>

#include "boost/asio/connect.hpp"  // NOLINT
#include "boost/asio/io_context.hpp"  // NOLINT
#include "boost/asio/ip/address.hpp"  // NOLINT
#include "boost/asio/ip/basic_endpoint.hpp"  // NOLINT
#include "boost/asio/ip/tcp.hpp"  // NOLINT
#include "boost/asio/ssl/context.hpp"  // NOLINT
#include "boost/asio/ssl/error.hpp"  // NOLINT
#include "boost/asio/steady_timer.hpp"  // NOLINT
#include "boost/beast/core/flat_buffer.hpp"  // NOLINT
#include "boost/beast/core/flat_static_buffer.hpp"  // NOLINT
#include "boost/beast/http/message.hpp"  // NOLINT
#include "boost/beast/http/parser.hpp"  // NOLINT
#include "boost/beast/http/read.hpp"  // NOLINT
#include "boost/beast/http/string_body.hpp"  // NOLINT
#include "boost/beast/http/write.hpp"  // NOLINT
#include "boost/beast/ssl/ssl_stream.hpp"  // NOLINT
#include "boost/beast/version.hpp"  // NOLINT
#include "boost/container/devector.hpp"  // NOLINT
#include "boost/system/error_code.hpp"  // NOLINT
#include "http_response.hpp"
#include "logging.hpp"
#include "async_resolve.hpp"
#include "ssl_key_handler.hpp"
#include "managed_store.hpp"

namespace crow {

// With Redfish Aggregation it is assumed we will connect to another instance
// of BMCWeb which can handle 100 simultaneous connections.
constexpr size_t maxPoolSize = 20;
constexpr size_t maxRequestQueueSize = 500;
constexpr unsigned int httpReadBodyLimit = 131072;
constexpr unsigned int httpReadBufferSize = 4096;

enum class ConnState : std::uint8_t {
  initialized,
  resolveInProgress,
  resolveFailed,
  connectInProgress,
  connectFailed,
  connected,
  handshakeInProgress,
  handshakeFailed,
  sendInProgress,
  sendFailed,
  recvInProgress,
  recvFailed,
  idle,
  closed,
  suspended,
  terminated,
  abortConnection,
  sslInitFailed,
  retry
};

static inline boost::system::error_code defaultRetryHandler(
    unsigned int respCode) {
  // As a default, assume 200X is alright
  BMCWEB_LOG_DEBUG << "Using default check for response code validity";
  if ((respCode < 200) || (respCode >= 300)) {
    return boost::system::errc::make_error_code(
        boost::system::errc::result_out_of_range);
  }

  // Return 0 if the response code is valid
  return boost::system::errc::make_error_code(boost::system::errc::success);
};

// We need to allow retry information to be set before a message has been sent
// and a connection pool has been created
struct ConnectionPolicy {
  uint32_t maxRetryAttempts = 5;  // NOLINT

  // the max size of requests in bytes.  0 for unlimited
  boost::optional<uint64_t> requestByteLimit = httpReadBodyLimit;  // NOLINT

  size_t maxConnections = 1;  // NOLINT

  std::string retryPolicyAction = "TerminateAfterRetries";  // NOLINT

  std::chrono::seconds retryIntervalSecs = std::chrono::seconds(0);  // NOLINT
  std::function<boost::system::error_code(unsigned int respCode)>
      invalidResp =  // NOLINT
      defaultRetryHandler;
};

struct PendingRequest {
  boost::beast::http::request<boost::beast::http::string_body> req;
  std::function<void(bool, uint32_t, Response&)> callback;
  PendingRequest(
      boost::beast::http::request<boost::beast::http::string_body>&& reqIn,
      const std::function<void(bool, uint32_t, Response&)>& callbackIn)
      : req(std::move(reqIn)), callback(callbackIn) {}
};

class ConnectionInfo : public std::enable_shared_from_this<ConnectionInfo> {
 private:
  ConnState state = ConnState::initialized;      // NOLINT
  uint32_t retryCount = 0;                       // NOLINT
  std::string subId;                             // NOLINT
  std::shared_ptr<ConnectionPolicy> connPolicy;  // NOLINT
  std::string host;                              // NOLINT
  uint16_t port;                                 // NOLINT
  uint32_t connId;                               // NOLINT

  // Data buffers
  boost::beast::http::request<boost::beast::http::string_body> req;  // NOLINT
  std::optional<
      boost::beast::http::response_parser<boost::beast::http::string_body>>
      parser;                                                   // NOLINT
  boost::beast::flat_static_buffer<httpReadBufferSize> buffer;  // NOLINT
  Response res;                                                 // NOLINT

  // Ascync callables
  std::function<void(bool, uint32_t, Response&)> callback;  // NOLINT
  crow::async_resolve::Resolver resolver;                   // NOLINT
  boost::asio::io_context& ioc;                             // NOLINT
  bool useSsl;                                              // NOLINT
  boost::asio::ip::tcp::socket conn;                        // NOLINT
  std::optional<boost::beast::ssl_stream<boost::asio::ip::tcp::socket&>>
      sslConn;  // NOLINT

  boost::asio::steady_timer timer;  // NOLINT

  // Increments whenever the pending asynchronous work changes: an operation
  // deadline or retry wait is armed, or an operation completes. Timer
  // handlers capture the value at arm time and discard themselves if it has
  // moved on, because cancel() cannot revoke a handler the reactor has
  // already queued with its expiry code.
  uint64_t timerGeneration = 0;  // NOLINT

  friend class ConnectionPool;

  // Arms the per-operation deadline for the next asynchronous operation.
  void armOperationTimeout() {
    timer.expires_after(std::chrono::seconds(30));
    timer.async_wait(
        std::bind_front(onTimeout, weak_from_this(), ++timerGeneration));
  }

  // Marks the pending operation as complete: invalidates any timeout or
  // retry handler that is already queued and cancels the timer proper.
  void completeOperation() {
    ++timerGeneration;
    timer.cancel();
  }

  void doResolve() {
    state = ConnState::resolveInProgress;
    BMCWEB_LOG_DEBUG << "Trying to resolve: " << host << ":"
                     << std::to_string(port)
                     << ", id: " << std::to_string(connId);

    resolver.asyncResolve(host, port,
                          std::bind_front(&ConnectionInfo::afterResolve, this,
                                          shared_from_this()));
  }

  void afterResolve(
      const std::shared_ptr<ConnectionInfo>& /*self*/,
      const boost::beast::error_code ec,
      const std::vector<boost::asio::ip::tcp::endpoint>& endpointList) {
    if (ec || (endpointList.empty())) {
      BMCWEB_LOG_ERROR << "Resolve failed: " << ec.message();
      state = ConnState::resolveFailed;
      waitAndRetry();
      return;
    }
    BMCWEB_LOG_DEBUG << "Resolved " << host << ":" << std::to_string(port)
                     << ", id: " << std::to_string(connId);
    state = ConnState::connectInProgress;

    BMCWEB_LOG_DEBUG << "Trying to connect to: " << host << ":"
                     << std::to_string(port)
                     << ", id: " << std::to_string(connId);

    armOperationTimeout();

    boost::asio::async_connect(conn, endpointList,
                               std::bind_front(&ConnectionInfo::afterConnect,
                                               this, shared_from_this()));
  }

  void afterConnect(const std::shared_ptr<ConnectionInfo>& /*self*/,
                    boost::beast::error_code ec,
                    const boost::asio::ip::tcp::endpoint& endpoint) {
    // The operation already timed out.  We don't want do continue down
    // this branch
    if (ec && ec == boost::asio::error::operation_aborted) {
      return;
    }

    completeOperation();
    if (ec) {
      BMCWEB_LOG_ERROR << "Connect " << endpoint.address().to_string() << ":"
                       << std::to_string(endpoint.port())
                       << ", id: " << std::to_string(connId)
                       << " failed: " << ec.message();
      state = ConnState::connectFailed;
      waitAndRetry();
      return;
    }
    BMCWEB_LOG_DEBUG << "Connected to: " << endpoint.address().to_string()
                     << ":" << std::to_string(endpoint.port())
                     << ", id: " << std::to_string(connId);
    if (sslConn) {
      doSslHandshake();
      return;
    }
    state = ConnState::connected;
    sendMessage();
  }

  void doSslHandshake() {
    if (!sslConn) {
      return;
    }
    state = ConnState::handshakeInProgress;
    armOperationTimeout();
    sslConn->async_handshake(boost::asio::ssl::stream_base::client,
                             std::bind_front(&ConnectionInfo::afterSslHandshake,
                                             this, shared_from_this()));
  }

  void afterSslHandshake(const std::shared_ptr<ConnectionInfo>& /*self*/,
                         boost::beast::error_code ec) {
    // The operation already timed out.  We don't want do continue down
    // this branch
    if (ec && ec == boost::asio::error::operation_aborted) {
      return;
    }

    completeOperation();
    if (ec) {
      BMCWEB_LOG_ERROR << "SSL Handshake failed -"
                       << " id: " << std::to_string(connId)
                       << " error: " << ec.message();
      state = ConnState::handshakeFailed;
      waitAndRetry();
      return;
    }
    BMCWEB_LOG_DEBUG << "SSL Handshake successful -"
                     << " id: " << std::to_string(connId);
    state = ConnState::connected;
    sendMessage();
  }

  void sendMessage() {
    state = ConnState::sendInProgress;

    // Set a timeout on the operation
    armOperationTimeout();

    // Send the HTTP request to the remote host
    if (sslConn) {
      boost::beast::http::async_write(
          *sslConn, req,
          std::bind_front(&ConnectionInfo::afterWrite, this,
                          shared_from_this()));
    } else {
      boost::beast::http::async_write(
          conn, req,
          std::bind_front(&ConnectionInfo::afterWrite, this,
                          shared_from_this()));
    }
  }

  void afterWrite(const std::shared_ptr<ConnectionInfo>& /*self*/,
                  const boost::beast::error_code& ec, size_t bytesTransferred) {
    // The operation already timed out.  We don't want do continue down
    // this branch
    if (ec && ec == boost::asio::error::operation_aborted) {
      return;
    }

    completeOperation();
    if (ec) {
      BMCWEB_LOG_ERROR << "sendMessage() failed: " << ec.message();
      state = ConnState::sendFailed;
      waitAndRetry();
      return;
    }
    BMCWEB_LOG_DEBUG << "sendMessage() bytes transferred: " << bytesTransferred;

    recvMessage();
  }

  void recvMessage() {
    state = ConnState::recvInProgress;

    parser.emplace(std::piecewise_construct, std::make_tuple());

    parser->body_limit(connPolicy->requestByteLimit);

    armOperationTimeout();

    // Receive the HTTP response
    if (sslConn) {
      boost::beast::http::async_read(*sslConn, buffer, *parser,
                                     std::bind_front(&ConnectionInfo::afterRead,
                                                     this, shared_from_this()));
    } else {
      boost::beast::http::async_read(conn, buffer, *parser,
                                     std::bind_front(&ConnectionInfo::afterRead,
                                                     this, shared_from_this()));
    }
  }

  void afterRead(const std::shared_ptr<ConnectionInfo>& /*self*/,
                 const boost::beast::error_code& ec,
                 const std::size_t& bytesTransferred) {
    // The operation already timed out.  We don't want do continue down
    // this branch
    if (ec && ec == boost::asio::error::operation_aborted) {
      return;
    }

    completeOperation();
    if (ec && ec != boost::asio::ssl::error::stream_truncated) {
      BMCWEB_LOG_ERROR << "recvMessage() failed: " << ec.message();
      state = ConnState::recvFailed;
      waitAndRetry();
      return;
    }
    BMCWEB_LOG_DEBUG << "recvMessage() bytes transferred: " << bytesTransferred;
    BMCWEB_LOG_DEBUG << "recvMessage() data: " << parser->get().body();

    unsigned int resp_code = parser->get().result_int();
    BMCWEB_LOG_DEBUG << "recvMessage() Header Response Code: " << resp_code;

    // Make sure the received response code is valid as defined by
    // the associated retry policy
    if (connPolicy->invalidResp(resp_code)) {
      // The listener failed to receive the Sent-Event
      BMCWEB_LOG_ERROR << "recvMessage() Listener Failed to "
                          "receive Sent-Event. Header Response Code: "
                       << resp_code;
      state = ConnState::recvFailed;
      waitAndRetry();
      return;
    }

    // Send is successful
    // Reset the counter just in case this was after retrying
    retryCount = 0;

    // Keep the connection alive if server supports it
    // Else close the connection
    BMCWEB_LOG_DEBUG << "recvMessage() keepalive : " << parser->keep_alive();

    // Copy the response into a Response object so that it can be
    // processed by the callback function.
    res.stringResponse = parser->release();
    if (callback) {
      callback(parser->keep_alive(), connId, res);
    } else {
      // Unreachable while the generation checks hold; if this fires, a
      // stale completion survived them and the response has no owner.
      BMCWEB_LOG_CRITICAL << "Response completed but callback was empty, id: "
                          << std::to_string(connId);
    }
    res.clear();
  }

  static void onTimeout(const std::weak_ptr<ConnectionInfo>& weakSelf,
                        uint64_t generation,
                        const boost::system::error_code& ec) {
    if (ec == boost::asio::error::operation_aborted) {
      BMCWEB_LOG_DEBUG << "async_wait failed since the operation is aborted";
      return;
    }
    if (ec) {
      BMCWEB_LOG_ERROR << "async_wait failed: " << ec.message();
      // If the timer fails, we need to close the socket anyway, same as
      // if it expired.
    }
    std::shared_ptr<ConnectionInfo> self = weakSelf.lock();
    if (self == nullptr) {
      return;
    }
    if (generation != self->timerGeneration) {
      // The operation completed while this expiry was already queued;
      // cancel() cannot revoke a queued handler, so it discards itself.
      BMCWEB_LOG_DEBUG << "Ignoring stale operation timeout";
      return;
    }
    self->waitAndRetry();
  }

  void waitAndRetry() {
    // Stop the in-flight operation before deciding this connection's fate.
    // A completion the reactor has already queued still runs; the
    // generation checks in the timer handlers neutralize that case.
    boost::system::error_code cancelEc;
    conn.cancel(cancelEc);
    if ((retryCount >= connPolicy->maxRetryAttempts) ||
        (state == ConnState::sslInitFailed)) {
      BMCWEB_LOG_ERROR << "Maximum number of retries reached.";
      BMCWEB_LOG_DEBUG << "Retry policy: " << connPolicy->retryPolicyAction;

      if (connPolicy->retryPolicyAction == "TerminateAfterRetries") {
        // TODO: delete subscription
        state = ConnState::terminated;
      }
      if (connPolicy->retryPolicyAction == "SuspendRetries") {
        state = ConnState::suspended;
      }

      // We want to return a 502 to indicate there was an error with
      // the external server
      res.result(boost::beast::http::status::bad_gateway);
      if (callback) {
        callback(false, connId, res);
      } else {
        // Expected only when SSL init fails during construction, before a
        // request was ever assigned to this connection.
        BMCWEB_LOG_ERROR << "Connection terminated with no pending callback,"
                         << " id: " << std::to_string(connId);
      }
      res.clear();

      // Reset the retrycount to zero so that client can try connecting
      // again if needed
      retryCount = 0;
      return;
    }

    retryCount++;

    BMCWEB_LOG_DEBUG << "Attempt retry after "
                     << std::to_string(connPolicy->retryIntervalSecs.count())
                     << " seconds. RetryCount = " << retryCount;
    timer.expires_after(connPolicy->retryIntervalSecs);
    timer.async_wait(std::bind_front(&ConnectionInfo::onTimerDone, this,
                                     shared_from_this(), ++timerGeneration));
  }

  void onTimerDone(const std::shared_ptr<ConnectionInfo>& /*self*/,
                   uint64_t generation, const boost::system::error_code& ec) {
    if (ec == boost::asio::error::operation_aborted) {
      BMCWEB_LOG_DEBUG << "Retry wait was cancelled";
      return;
    }
    if (ec) {
      BMCWEB_LOG_ERROR << "async_wait failed: " << ec.message();
      // A timer error still warrants the retry the policy asked for.
    }
    if (generation != timerGeneration) {
      // The response arrived and was delivered while this retry was
      // pending; there is nothing left to retry.
      BMCWEB_LOG_DEBUG << "Ignoring stale retry wait";
      return;
    }

    // Close the connection and restart from resolve on a fresh transport.
    // The peer is unresponsive or broken, so skip the graceful SSL
    // shutdown; its completion may never arrive and would stall the retry.
    shutdownConn(true);
  }

  void shutdownConn(bool retry) {
    boost::beast::error_code ec;
    // NOLINTNEXTLINE(bugprone-unused-return-value)
    conn.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
    conn.close();

    // not_connected happens sometimes so don't bother reporting it.
    if (ec && ec != boost::beast::errc::not_connected) {
      BMCWEB_LOG_ERROR << host << ":" << std::to_string(port)
                       << ", id: " << std::to_string(connId)
                       << " shutdown failed: " << ec.message();
    } else {
      BMCWEB_LOG_DEBUG << host << ":" << std::to_string(port)
                       << ", id: " << std::to_string(connId)
                       << " closed gracefully";
    }

    if (retry) {
      // Now let's try to resend the data
      state = ConnState::retry;
      restartConnection();
    } else {
      state = ConnState::closed;
    }
  }

  void doClose(bool retry = false) {
    if (!sslConn) {
      shutdownConn(retry);
      return;
    }

    sslConn->async_shutdown(std::bind_front(&ConnectionInfo::afterSslShutdown,
                                            this, shared_from_this(), retry));
  }

  void afterSslShutdown(const std::shared_ptr<ConnectionInfo>& /*self*/,
                        bool retry, const boost::system::error_code& ec) {
    if (ec) {
      BMCWEB_LOG_ERROR << host << ":" << std::to_string(port)
                       << ", id: " << std::to_string(connId)
                       << " shutdown failed: " << ec.message();
    } else {
      BMCWEB_LOG_DEBUG << host << ":" << std::to_string(port)
                       << ", id: " << std::to_string(connId)
                       << " closed gracefully";
    }
    shutdownConn(retry);
  }

  bool setCipherSuiteTLSext() {
    if (!sslConn) {
      return false;
    }

#ifdef GOOGLE3_GBMCWEB_BUILD
    if (SSL_set_tlsext_host_name(sslConn->native_handle(), host.c_str()) == 0)
#else
    // NOTE: The SSL_set_tlsext_host_name is defined in tlsv1.h header
    // file but its having old style casting (name is cast to void*).
    // Since bmcweb compiler treats all old-style-cast as error, its
    // causing the build failure. So replaced the same macro inline and
    // did corrected the code by doing static_cast to viod*. This has to
    // be fixed in openssl library in long run. Set SNI Hostname (many
    // hosts need this to handshake successfully)
    if (SSL_ctrl(sslConn->native_handle(), SSL_CTRL_SET_TLSEXT_HOSTNAME,
                 TLSEXT_NAMETYPE_host_name,
                 static_cast<void*>(&host.front())) == 0)
#endif
    {
      boost::beast::error_code ec{static_cast<int>(::ERR_get_error()),
                                  boost::asio::error::get_ssl_category()};

      BMCWEB_LOG_ERROR << "SSL_set_tlsext_host_name " << host << ":" << port
                       << ", id: " << std::to_string(connId)
                       << " failed: " << ec.message();
      // Set state as sslInit failed so that we close the connection
      // and take appropriate action as per retry configuration.
      state = ConnState::sslInitFailed;
      waitAndRetry();
      return false;
    }
    return true;
  }

  bool initializeSslConn() {
    std::optional<boost::asio::ssl::context> ssl_ctx =
        ensuressl::getSSLClientContext();

    if (!ssl_ctx) {
      BMCWEB_LOG_ERROR << "prepareSSLContext failed - " << host << ":" << port
                       << ", id: " << std::to_string(connId);
      // Don't retry if failure occurs while preparing SSL context
      // such as certificate is invalid or set cipher failure or set
      // host name failure etc... Setting conn state to sslInitFailed
      // and connection state will be transitioned to next state
      // depending on retry policy set by subscription.
      state = ConnState::sslInitFailed;
      waitAndRetry();
      return false;
    }
    sslConn.emplace(conn, *ssl_ctx);
    return setCipherSuiteTLSext();
  }

  // Retries and reconnects must not reuse the old transport: the socket has
  // been shut down, and an SSL session that already ran a handshake cannot
  // run another one. Recreate both, then resolve again.
  void restartConnection() {
    sslConn.reset();
    conn = boost::asio::ip::tcp::socket(ioc);
    if (useSsl) {
      if (!initializeSslConn()) {
        return;
      }
    }
    doResolve();
  }

 public:
  explicit ConnectionInfo(boost::asio::io_context& iocIn,
                          const std::string& idIn,
                          const std::shared_ptr<ConnectionPolicy>& connPolicyIn,
                          const std::string& destIPIn, uint16_t destPortIn,
                          bool useSSL, unsigned int connIdIn)
      : subId(idIn),
        connPolicy(connPolicyIn),
        host(destIPIn),
        port(destPortIn),
        connId(connIdIn),
        ioc(iocIn),
        useSsl(useSSL),
        conn(iocIn),
        timer(iocIn) {
    if (useSsl) {
      initializeSslConn();
    }
  }
};

class ConnectionPool : public std::enable_shared_from_this<ConnectionPool> {
 private:
  boost::asio::io_context& ioc;                              // NOLINT
  std::string id;                                            // NOLINT
  std::shared_ptr<ConnectionPolicy> connPolicy;              // NOLINT
  std::string destIP;                                        // NOLINT
  uint16_t destPort;                                         // NOLINT
  bool useSSL;                                               // NOLINT
  std::vector<std::shared_ptr<ConnectionInfo>> connections;  // NOLINT
  boost::container::devector<PendingRequest> requestQueue;   // NOLINT

  friend class HttpClient;

  // Configure a connections's request, callback, and retry info in
  // preparation to begin sending the request
  void setConnProps(ConnectionInfo& conn) {
    if (requestQueue.empty()) {
      BMCWEB_LOG_ERROR << "setConnProps() should not have been called when "
                          "requestQueue is empty";
      return;
    }

    auto next_req = requestQueue.front();
    conn.req = std::move(next_req.req);
    conn.callback = std::move(next_req.callback);

    BMCWEB_LOG_DEBUG << "Setting properties for connection " << conn.host << ":"
                     << std::to_string(conn.port)
                     << ", id: " << std::to_string(conn.connId);

    // We can remove the request from the queue at this point
    requestQueue.pop_front();
  }

  // Gets called as part of callback after request is sent
  // Reuses the connection if there are any requests waiting to be sent
  // Otherwise closes the connection if it is not a keep-alive
  void sendNext(bool keepAlive, uint32_t connId) {
    auto conn = connections[connId];

    // Allow the connection's handler to be deleted
    // This is needed because of Redfish Aggregation passing an
    // AsyncResponse shared_ptr to this callback
    conn->callback = nullptr;

    // Reuse the connection to send the next request in the queue
    if (!requestQueue.empty()) {
      BMCWEB_LOG_DEBUG << std::to_string(requestQueue.size())
                       << " requests remaining in queue for " << destIP << ":"
                       << std::to_string(destPort) << ", reusing connnection "
                       << std::to_string(connId);

      setConnProps(*conn);

      if (keepAlive) {
        conn->sendMessage();
      } else {
        // Server is not keep-alive enabled so we need to close the
        // connection and then start over from resolve. doClose(true)
        // restarts on a fresh transport once the shutdown completes,
        // instead of racing a resolve against an asynchronous SSL
        // shutdown on the same socket.
        conn->doClose(true);
      }
      return;
    }

    // No more messages to send so close the connection if necessary
    if (keepAlive) {
      conn->state = ConnState::idle;
    } else {
      // Abort the connection since server is not keep-alive enabled
      conn->state = ConnState::abortConnection;
      conn->doClose();
    }
  }

  void sendData(std::string& data, const std::string& destUri,
                const boost::beast::http::fields& httpHeader,
                const boost::beast::http::verb verb,
                const std::function<void(Response&)>& resHandler) {
    // Construct the request to be sent
    boost::beast::http::request<boost::beast::http::string_body> this_req(
        verb, destUri, 11, "", httpHeader);
    this_req.set(boost::beast::http::field::host, destIP);
    this_req.keep_alive(true);
    this_req.body() = std::move(data);
    this_req.prepare_payload();
    auto cb = std::bind_front(&ConnectionPool::afterSendData, weak_from_this(),
                              resHandler);
    // Reuse an existing connection if one is available
    for (unsigned int i = 0; i < connections.size(); i++) {
      auto conn = connections[i];
      if ((conn->state == ConnState::idle) ||
          (conn->state == ConnState::initialized) ||
          (conn->state == ConnState::closed)) {
        conn->req = std::move(this_req);
        conn->callback = std::move(cb);
        std::string common_msg = std::to_string(i) + " from pool " + destIP +
                                 ":" + std::to_string(destPort);

        if (conn->state == ConnState::idle) {
          BMCWEB_LOG_DEBUG << "Grabbing idle connection " << common_msg;
          conn->sendMessage();
        } else {
          BMCWEB_LOG_DEBUG << "Reusing existing connection " << common_msg;
          conn->restartConnection();
        }
        return;
      }
    }

    // All connections in use so create a new connection or add request to
    // the queue
    if (connections.size() < connPolicy->maxConnections) {
      BMCWEB_LOG_DEBUG << "Adding new connection to pool " << destIP << ":"
                       << std::to_string(destPort);
      auto conn = addConnection();
      conn->req = std::move(this_req);
      conn->callback = std::move(cb);
      conn->doResolve();
    } else if (requestQueue.size() < maxRequestQueueSize) {
      BMCWEB_LOG_ERROR << "Max pool size reached. Adding data to queue.";
      requestQueue.emplace_back(std::move(this_req), std::move(cb));
    } else {
      // If we can't buffer the request then we should let the callback
      // handle a 429 Too Many Requests dummy response
      BMCWEB_LOG_ERROR << destIP << ":" << std::to_string(destPort)
                       << " request queue full.  Dropping request.";
      Response dummy_res;
      dummy_res.result(boost::beast::http::status::too_many_requests);
      resHandler(dummy_res);
    }
  }

  // Callback to be called once the request has been sent
  static void afterSendData(const std::weak_ptr<ConnectionPool>& weakSelf,
                            const std::function<void(Response&)>& resHandler,
                            bool keepAlive, uint32_t connId, Response& res) {
    // Allow provided callback to perform additional processing of the
    // request
    resHandler(res);

    // If requests remain in the queue then we want to reuse this
    // connection to send the next request
    std::shared_ptr<ConnectionPool> self = weakSelf.lock();
    if (!self) {
      BMCWEB_LOG_CRITICAL << self << " Failed to capture connection";
      return;
    }

    self->sendNext(keepAlive, connId);
  }

  std::shared_ptr<ConnectionInfo>& addConnection() {
    unsigned int new_id = static_cast<unsigned int>(connections.size());

    auto& ret = connections.emplace_back(std::make_shared<ConnectionInfo>(
        ioc, id, connPolicy, destIP, destPort, useSSL, new_id));

    BMCWEB_LOG_DEBUG << "Added connection "
                     << std::to_string(connections.size() - 1) << " to pool "
                     << destIP << ":" << std::to_string(destPort);

    return ret;
  }

 public:
  explicit ConnectionPool(boost::asio::io_context& iocIn,
                          const std::string& idIn,
                          const std::shared_ptr<ConnectionPolicy>& connPolicyIn,
                          const std::string& destIPIn, uint16_t destPortIn,
                          bool useSSLIn)
      : ioc(iocIn),
        id(idIn),
        connPolicy(connPolicyIn),
        destIP(destIPIn),
        destPort(destPortIn),
        useSSL(useSSLIn) {
    BMCWEB_LOG_DEBUG << "Initializing connection pool for " << destIP << ":"
                     << std::to_string(destPort);

    // Initialize the pool with a single connection
    addConnection();
  }
};

class HttpClient {
 private:
  std::unordered_map<std::string, std::shared_ptr<ConnectionPool>>
      connectionPools;            // NOLINT
  boost::asio::io_context& ioc =  // NOLINT
      managedStore::GetManagedObjectStore()->GetIoContext();
  std::shared_ptr<ConnectionPolicy> connPolicy;  // NOLINT

  // Used as a dummy callback by sendData() in order to call
  // sendDataWithCallback()
  static void genericResHandler(const Response& res) {
    BMCWEB_LOG_DEBUG << "Response handled with return code: "
                     << std::to_string(res.resultInt());
  }

 public:
  HttpClient() = delete;
  explicit HttpClient(const std::shared_ptr<ConnectionPolicy>& connPolicyIn)
      : connPolicy(connPolicyIn) {}
  HttpClient(const HttpClient&) = delete;
  HttpClient& operator=(const HttpClient&) = delete;
  HttpClient(HttpClient&&) = delete;
  HttpClient& operator=(HttpClient&&) = delete;
  ~HttpClient() = default;

  // Send a request to destIP:destPort where additional processing of the
  // result is not required
  void sendData(std::string& data, const std::string& destIP, uint16_t destPort,
                const std::string& destUri, bool useSSL,
                const boost::beast::http::fields& httpHeader,
                const boost::beast::http::verb verb) {
    const std::function<void(Response&)> cb = genericResHandler;
    sendDataWithCallback(data, destIP, destPort, destUri, useSSL, httpHeader,
                         verb, cb);
  }

  // Send request to destIP:destPort and use the provided callback to
  // handle the response
  void sendDataWithCallback(std::string& data, const std::string& destIP,
                            uint16_t destPort, const std::string& destUri,
                            bool useSSL,
                            const boost::beast::http::fields& httpHeader,
                            const boost::beast::http::verb verb,
                            const std::function<void(Response&)>& resHandler) {
    std::string client_key = useSSL ? "https" : "http";
    client_key += destIP;
    client_key += ":";
    client_key += std::to_string(destPort);
    auto pool = connectionPools.try_emplace(client_key);
    if (pool.first->second == nullptr) {
      pool.first->second = std::make_shared<ConnectionPool>(
          ioc, client_key, connPolicy, destIP, destPort, useSSL);
    }
    // Send the data using either the existing connection pool or the newly
    // created connection pool
    pool.first->second->sendData(data, destUri, httpHeader, verb, resHandler);
  }
};
}  // namespace crow

#endif  // THIRD_PARTY_GBMCWEB_HTTP_HTTP_CLIENT_H_