/*
 * nghttp2 - HTTP/2 C Library
 *
 * Copyright (c) 2012 Tatsuhiro Tsujikawa
 *
 * Permission is hereby granted, free of charge, to any person obtaining
 * a copy of this software and associated documentation files (the
 * "Software"), to deal in the Software without restriction, including
 * without limitation the rights to use, copy, modify, merge, publish,
 * distribute, sublicense, and/or sell copies of the Software, and to
 * permit persons to whom the Software is furnished to do so, subject to
 * the following conditions:
 *
 * The above copyright notice and this permission notice shall be
 * included in all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
 * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
 * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
 * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */
#include "shrpx_http_downstream_connection.h"

#include "shrpx_client_handler.h"
#include "shrpx_upstream.h"
#include "shrpx_downstream.h"
#include "shrpx_config.h"
#include "shrpx_error.h"
#include "shrpx_http.h"
#include "shrpx_log_config.h"
#include "shrpx_connect_blocker.h"
#include "shrpx_downstream_connection_pool.h"
#include "shrpx_worker.h"
#include "shrpx_http2_session.h"
#include "shrpx_ssl.h"
#include "http2.h"
#include "util.h"

using namespace nghttp2;

namespace shrpx {

namespace {
void timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
  auto conn = static_cast<Connection *>(w->data);
  auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);

  if (LOG_ENABLED(INFO)) {
    DCLOG(INFO, dconn) << "Time out";
  }

  auto downstream = dconn->get_downstream();
  auto upstream = downstream->get_upstream();
  auto handler = upstream->get_client_handler();
  auto &resp = downstream->response();

  // Do this so that dconn is not pooled
  resp.connection_close = true;

  if (upstream->downstream_error(dconn, Downstream::EVENT_TIMEOUT) != 0) {
    delete handler;
  }
}
} // namespace

namespace {
void readcb(struct ev_loop *loop, ev_io *w, int revents) {
  auto conn = static_cast<Connection *>(w->data);
  auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
  auto downstream = dconn->get_downstream();
  auto upstream = downstream->get_upstream();
  auto handler = upstream->get_client_handler();

  if (upstream->downstream_read(dconn) != 0) {
    delete handler;
  }
}
} // namespace

namespace {
void writecb(struct ev_loop *loop, ev_io *w, int revents) {
  auto conn = static_cast<Connection *>(w->data);
  auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
  auto downstream = dconn->get_downstream();
  auto upstream = downstream->get_upstream();
  auto handler = upstream->get_client_handler();

  if (upstream->downstream_write(dconn) != 0) {
    delete handler;
  }
}
} // namespace

namespace {
void connectcb(struct ev_loop *loop, ev_io *w, int revents) {
  auto conn = static_cast<Connection *>(w->data);
  auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
  auto downstream = dconn->get_downstream();
  auto upstream = downstream->get_upstream();
  auto handler = upstream->get_client_handler();
  if (dconn->connected() != 0) {
    if (upstream->on_downstream_abort_request(downstream, 503) != 0) {
      delete handler;
    }
    return;
  }
  writecb(loop, w, revents);
}
} // namespace

HttpDownstreamConnection::HttpDownstreamConnection(
    DownstreamConnectionPool *dconn_pool, size_t group, struct ev_loop *loop,
    Worker *worker)
    : DownstreamConnection(dconn_pool),
      conn_(loop, -1, nullptr, worker->get_mcpool(),
            get_config()->conn.downstream.timeout.write,
            get_config()->conn.downstream.timeout.read, {}, {}, connectcb,
            readcb, timeoutcb, this, get_config()->tls.dyn_rec.warmup_threshold,
            get_config()->tls.dyn_rec.idle_timeout),
      do_read_(&HttpDownstreamConnection::noop),
      do_write_(&HttpDownstreamConnection::noop),
      worker_(worker),
      ssl_ctx_(worker->get_cl_ssl_ctx()),
      addr_(nullptr),
      ioctrl_(&conn_.rlimit),
      response_htp_{0},
      group_(group) {}

HttpDownstreamConnection::~HttpDownstreamConnection() {}

int HttpDownstreamConnection::attach_downstream(Downstream *downstream) {
  if (LOG_ENABLED(INFO)) {
    DCLOG(INFO, this) << "Attaching to DOWNSTREAM:" << downstream;
  }

  auto worker_blocker = worker_->get_connect_blocker();
  if (worker_blocker->blocked()) {
    if (LOG_ENABLED(INFO)) {
      DCLOG(INFO, this)
          << "Worker wide backend connection was blocked temporarily";
    }
    return SHRPX_ERR_NETWORK;
  }

  auto &downstreamconf = get_config()->conn.downstream;

  if (conn_.fd == -1) {
    if (ssl_ctx_) {
      auto ssl = ssl::create_ssl(ssl_ctx_);
      if (!ssl) {
        return -1;
      }

      conn_.set_ssl(ssl);
    }

    auto &next_downstream = worker_->get_dgrp(group_)->next;
    auto end = next_downstream;
    auto &groups = worker_->get_downstream_addr_groups();
    auto &addrs = groups[group_].addrs;
    for (;;) {
      auto &addr = addrs[next_downstream];

      if (++next_downstream >= addrs.size()) {
        next_downstream = 0;
      }

      auto &connect_blocker = addr.connect_blocker;

      if (connect_blocker->blocked()) {
        if (LOG_ENABLED(INFO)) {
          DCLOG(INFO, this) << "Backend server "
                            << util::to_numeric_addr(&addr.addr)
                            << " was not available temporarily";
        }

        if (end == next_downstream) {
          return SHRPX_ERR_NETWORK;
        }

        continue;
      }

      conn_.fd = util::create_nonblock_socket(addr.addr.su.storage.ss_family);

      if (conn_.fd == -1) {
        auto error = errno;
        DCLOG(WARN, this) << "socket() failed; addr="
                          << util::to_numeric_addr(&addr.addr)
                          << ", errno=" << error;

        worker_blocker->on_failure();

        return SHRPX_ERR_NETWORK;
      }

      worker_blocker->on_success();

      int rv;
      rv = connect(conn_.fd, &addr.addr.su.sa, addr.addr.len);
      if (rv != 0 && errno != EINPROGRESS) {
        auto error = errno;
        DCLOG(WARN, this) << "connect() failed; addr="
                          << util::to_numeric_addr(&addr.addr)
                          << ", errno=" << error;

        connect_blocker->on_failure();
        close(conn_.fd);
        conn_.fd = -1;

        if (end == next_downstream) {
          return SHRPX_ERR_NETWORK;
        }

        // Try again with the next downstream server
        continue;
      }

      if (LOG_ENABLED(INFO)) {
        DCLOG(INFO, this) << "Connecting to downstream server";
      }

      addr_ = &addr;

      if (ssl_ctx_) {
        auto sni_name = !get_config()->tls.backend_sni_name.empty()
                            ? StringRef(get_config()->tls.backend_sni_name)
                            : StringRef(addr_->host);
        if (!util::numeric_host(sni_name.c_str())) {
          SSL_set_tlsext_host_name(conn_.tls.ssl, sni_name.c_str());
        }

        auto session = ssl::reuse_tls_session(addr_);
        if (session) {
          SSL_set_session(conn_.tls.ssl, session);
          SSL_SESSION_free(session);
        }

        conn_.prepare_client_handshake();
      }

      ev_io_set(&conn_.wev, conn_.fd, EV_WRITE);
      ev_io_set(&conn_.rev, conn_.fd, EV_READ);

      conn_.wlimit.startw();

      break;
    }

    // TODO we should have timeout for connection establishment
    ev_timer_again(conn_.loop, &conn_.wt);
  } else {
    // we may set read timer cb to idle_timeoutcb.  Reset again.
    conn_.rt.repeat = downstreamconf.timeout.read;
    ev_set_cb(&conn_.rt, timeoutcb);
    ev_timer_again(conn_.loop, &conn_.rt);
    ev_set_cb(&conn_.rev, readcb);
  }

  downstream_ = downstream;

  http_parser_init(&response_htp_, HTTP_RESPONSE);
  response_htp_.data = downstream_;

  return 0;
}

int HttpDownstreamConnection::push_request_headers() {
  const auto &downstream_hostport = addr_->hostport;
  const auto &req = downstream_->request();

  auto connect_method = req.method == HTTP_CONNECT;

  auto &httpconf = get_config()->http;

  // For HTTP/1.0 request, there is no authority in request.  In that
  // case, we use backend server's host nonetheless.
  auto authority = StringRef(downstream_hostport);
  auto no_host_rewrite = httpconf.no_host_rewrite ||
                         get_config()->http2_proxy ||
                         get_config()->client_proxy || connect_method;

  if (no_host_rewrite && !req.authority.empty()) {
    authority = StringRef(req.authority);
  }

  downstream_->set_request_downstream_host(authority.str());

  auto buf = downstream_->get_request_buf();

  // Assume that method and request path do not contain \r\n.
  auto meth = http2::to_method_string(req.method);
  buf->append(meth, strlen(meth));
  buf->append(" ");

  if (connect_method) {
    buf->append(authority);
  } else if (get_config()->http2_proxy || get_config()->client_proxy) {
    // Construct absolute-form request target because we are going to
    // send a request to a HTTP/1 proxy.
    assert(!req.scheme.empty());
    buf->append(req.scheme);
    buf->append("://");
    buf->append(authority);
    buf->append(req.path);
  } else if (req.method == HTTP_OPTIONS && req.path.empty()) {
    // Server-wide OPTIONS
    buf->append("*");
  } else {
    buf->append(req.path);
  }
  buf->append(" HTTP/1.1\r\nHost: ");
  buf->append(authority);
  buf->append("\r\n");

  http2::build_http1_headers_from_headers(buf, req.fs.headers());

  auto cookie = downstream_->assemble_request_cookie();
  if (!cookie.empty()) {
    buf->append("Cookie: ");
    buf->append(cookie);
    buf->append("\r\n");
  }

  if (!connect_method && req.http2_expect_body &&
      !req.fs.header(http2::HD_CONTENT_LENGTH)) {

    downstream_->set_chunked_request(true);
    buf->append("Transfer-Encoding: chunked\r\n");
  }

  if (req.connection_close) {
    buf->append("Connection: close\r\n");
  }

  if (!connect_method && req.upgrade_request) {
    auto connection = req.fs.header(http2::HD_CONNECTION);
    if (connection) {
      buf->append("Connection: ");
      buf->append((*connection).value);
      buf->append("\r\n");
    }

    auto upgrade = req.fs.header(http2::HD_UPGRADE);
    if (upgrade) {
      buf->append("Upgrade: ");
      buf->append((*upgrade).value);
      buf->append("\r\n");
    }
  }

  auto upstream = downstream_->get_upstream();
  auto handler = upstream->get_client_handler();

  auto &fwdconf = httpconf.forwarded;

  auto fwd =
      fwdconf.strip_incoming ? nullptr : req.fs.header(http2::HD_FORWARDED);

  if (fwdconf.params) {
    auto params = fwdconf.params;

    if (get_config()->http2_proxy || get_config()->client_proxy ||
        connect_method) {
      params &= ~FORWARDED_PROTO;
    }

    auto value = http::create_forwarded(params, handler->get_forwarded_by(),
                                        handler->get_forwarded_for(),
                                        req.authority, req.scheme);
    if (fwd || !value.empty()) {
      buf->append("Forwarded: ");
      if (fwd) {
        buf->append(fwd->value);

        if (!value.empty()) {
          buf->append(", ");
        }
      }
      buf->append(value);
      buf->append("\r\n");
    }
  } else if (fwd) {
    buf->append("Forwarded: ");
    buf->append(fwd->value);
    buf->append("\r\n");
  }

  auto &xffconf = httpconf.xff;

  auto xff = xffconf.strip_incoming ? nullptr
                                    : req.fs.header(http2::HD_X_FORWARDED_FOR);

  if (xffconf.add) {
    buf->append("X-Forwarded-For: ");
    if (xff) {
      buf->append((*xff).value);
      buf->append(", ");
    }
    buf->append(client_handler_->get_ipaddr());
    buf->append("\r\n");
  } else if (xff) {
    buf->append("X-Forwarded-For: ");
    buf->append((*xff).value);
    buf->append("\r\n");
  }
  if (!get_config()->http2_proxy && !get_config()->client_proxy &&
      !connect_method) {
    buf->append("X-Forwarded-Proto: ");
    assert(!req.scheme.empty());
    buf->append(req.scheme);
    buf->append("\r\n");
  }
  auto via = req.fs.header(http2::HD_VIA);
  if (httpconf.no_via) {
    if (via) {
      buf->append("Via: ");
      buf->append((*via).value);
      buf->append("\r\n");
    }
  } else {
    buf->append("Via: ");
    if (via) {
      buf->append((*via).value);
      buf->append(", ");
    }
    buf->append(http::create_via_header_value(req.http_major, req.http_minor));
    buf->append("\r\n");
  }

  for (auto &p : httpconf.add_request_headers) {
    buf->append(p.name);
    buf->append(": ");
    buf->append(p.value);
    buf->append("\r\n");
  }

  buf->append("\r\n");

  if (LOG_ENABLED(INFO)) {
    std::string nhdrs;
    for (auto chunk = buf->head; chunk; chunk = chunk->next) {
      nhdrs.append(chunk->pos, chunk->last);
    }
    if (log_config()->errorlog_tty) {
      nhdrs = http::colorizeHeaders(nhdrs.c_str());
    }
    DCLOG(INFO, this) << "HTTP request headers. stream_id="
                      << downstream_->get_stream_id() << "\n" << nhdrs;
  }

  signal_write();

  return 0;
}

int HttpDownstreamConnection::push_upload_data_chunk(const uint8_t *data,
                                                     size_t datalen) {
  auto chunked = downstream_->get_chunked_request();
  auto output = downstream_->get_request_buf();

  if (chunked) {
    auto chunk_size_hex = util::utox(datalen);
    output->append(chunk_size_hex);
    output->append("\r\n");
  }

  output->append(data, datalen);

  if (chunked) {
    output->append("\r\n");
  }

  signal_write();

  return 0;
}

int HttpDownstreamConnection::end_upload_data() {
  if (!downstream_->get_chunked_request()) {
    return 0;
  }

  const auto &req = downstream_->request();

  auto output = downstream_->get_request_buf();
  const auto &trailers = req.fs.trailers();
  if (trailers.empty()) {
    output->append("0\r\n\r\n");
  } else {
    output->append("0\r\n");
    http2::build_http1_headers_from_headers(output, trailers);
    output->append("\r\n");
  }

  signal_write();

  return 0;
}

namespace {
void idle_readcb(struct ev_loop *loop, ev_io *w, int revents) {
  auto conn = static_cast<Connection *>(w->data);
  auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
  if (LOG_ENABLED(INFO)) {
    DCLOG(INFO, dconn) << "Idle connection EOF";
  }
  auto dconn_pool = dconn->get_dconn_pool();
  dconn_pool->remove_downstream_connection(dconn);
  // dconn was deleted
}
} // namespace

namespace {
void idle_timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
  auto conn = static_cast<Connection *>(w->data);
  auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
  if (LOG_ENABLED(INFO)) {
    DCLOG(INFO, dconn) << "Idle connection timeout";
  }
  auto dconn_pool = dconn->get_dconn_pool();
  dconn_pool->remove_downstream_connection(dconn);
  // dconn was deleted
}
} // namespace

void HttpDownstreamConnection::detach_downstream(Downstream *downstream) {
  if (LOG_ENABLED(INFO)) {
    DCLOG(INFO, this) << "Detaching from DOWNSTREAM:" << downstream;
  }
  downstream_ = nullptr;

  ev_set_cb(&conn_.rev, idle_readcb);
  ioctrl_.force_resume_read();

  conn_.rt.repeat = get_config()->conn.downstream.timeout.idle_read;
  ev_set_cb(&conn_.rt, idle_timeoutcb);
  ev_timer_again(conn_.loop, &conn_.rt);

  conn_.wlimit.stopw();
  ev_timer_stop(conn_.loop, &conn_.wt);
}

void HttpDownstreamConnection::pause_read(IOCtrlReason reason) {
  ioctrl_.pause_read(reason);
}

int HttpDownstreamConnection::resume_read(IOCtrlReason reason,
                                          size_t consumed) {
  if (downstream_->get_response_buf()->rleft() <=
      get_config()->conn.downstream.request_buffer_size / 2) {
    ioctrl_.resume_read(reason);
  }

  return 0;
}

void HttpDownstreamConnection::force_resume_read() {
  ioctrl_.force_resume_read();
}

namespace {
int htp_msg_begincb(http_parser *htp) {
  auto downstream = static_cast<Downstream *>(htp->data);

  if (downstream->get_response_state() != Downstream::INITIAL) {
    return -1;
  }

  return 0;
}
} // namespace

namespace {
int htp_hdrs_completecb(http_parser *htp) {
  auto downstream = static_cast<Downstream *>(htp->data);
  auto upstream = downstream->get_upstream();
  const auto &req = downstream->request();
  auto &resp = downstream->response();
  int rv;

  resp.http_status = htp->status_code;
  resp.http_major = htp->http_major;
  resp.http_minor = htp->http_minor;

  if (resp.fs.parse_content_length() != 0) {
    downstream->set_response_state(Downstream::MSG_BAD_HEADER);
    return -1;
  }

  // Check upgrade before processing non-final response, since if
  // upgrade succeeded, 101 response is treated as final in nghttpx.
  downstream->check_upgrade_fulfilled();

  if (downstream->get_non_final_response()) {
    // Reset content-length because we reuse same Downstream for the
    // next response.
    resp.fs.content_length = -1;
    // For non-final response code, we just call
    // on_downstream_header_complete() without changing response
    // state.
    rv = upstream->on_downstream_header_complete(downstream);

    if (rv != 0) {
      return -1;
    }

    // Ignore response body for non-final response.
    return 1;
  }

  resp.connection_close = !http_should_keep_alive(htp);
  downstream->set_response_state(Downstream::HEADER_COMPLETE);
  downstream->inspect_http1_response();
  if (downstream->get_upgraded()) {
    // content-length must be ignored for upgraded connection.
    resp.fs.content_length = -1;
    resp.connection_close = true;
    // transfer-encoding not applied to upgraded connection
    downstream->set_chunked_response(false);
  }
  if (upstream->on_downstream_header_complete(downstream) != 0) {
    return -1;
  }

  if (downstream->get_upgraded()) {
    // Upgrade complete, read until EOF in both ends
    if (upstream->resume_read(SHRPX_NO_BUFFER, downstream, 0) != 0) {
      return -1;
    }
    downstream->set_request_state(Downstream::HEADER_COMPLETE);
    if (LOG_ENABLED(INFO)) {
      LOG(INFO) << "HTTP upgrade success. stream_id="
                << downstream->get_stream_id();
    }
  }

  auto status = resp.http_status;
  // Ignore the response body. HEAD response may contain
  // Content-Length or Transfer-Encoding: chunked.  Some server send
  // 304 status code with nonzero Content-Length, but without response
  // body. See
  // https://tools.ietf.org/html/rfc7230#section-3.3

  // TODO It seems that the cases other than HEAD are handled by
  // http-parser.  Need test.
  return req.method == HTTP_HEAD || (100 <= status && status <= 199) ||
                 status == 204 || status == 304
             ? 1
             : 0;
}
} // namespace

namespace {
int ensure_header_field_buffer(const Downstream *downstream,
                               const HttpConfig &httpconf, size_t len) {
  auto &resp = downstream->response();

  if (resp.fs.buffer_size() + len > httpconf.response_header_field_buffer) {
    if (LOG_ENABLED(INFO)) {
      DLOG(INFO, downstream) << "Too large header header field size="
                             << resp.fs.buffer_size() + len;
    }
    return -1;
  }

  return 0;
}
} // namespace

namespace {
int ensure_max_header_fields(const Downstream *downstream,
                             const HttpConfig &httpconf) {
  auto &resp = downstream->response();

  if (resp.fs.num_fields() >= httpconf.max_response_header_fields) {
    if (LOG_ENABLED(INFO)) {
      DLOG(INFO, downstream)
          << "Too many header field num=" << resp.fs.num_fields() + 1;
    }
    return -1;
  }

  return 0;
}
} // namespace

namespace {
int htp_hdr_keycb(http_parser *htp, const char *data, size_t len) {
  auto downstream = static_cast<Downstream *>(htp->data);
  auto &resp = downstream->response();
  auto &httpconf = get_config()->http;

  if (ensure_header_field_buffer(downstream, httpconf, len) != 0) {
    return -1;
  }

  if (downstream->get_response_state() == Downstream::INITIAL) {
    if (resp.fs.header_key_prev()) {
      resp.fs.append_last_header_key(data, len);
    } else {
      if (ensure_max_header_fields(downstream, httpconf) != 0) {
        return -1;
      }
      resp.fs.add_header_lower(StringRef{data, len}, StringRef{}, false);
    }
  } else {
    // trailer part
    if (resp.fs.trailer_key_prev()) {
      resp.fs.append_last_trailer_key(data, len);
    } else {
      if (ensure_max_header_fields(downstream, httpconf) != 0) {
        // Could not ignore this trailer field easily, since we may
        // get its value in htp_hdr_valcb, and it will be added to
        // wrong place or crash if trailer fields are currently empty.
        return -1;
      }
      resp.fs.add_trailer_lower(StringRef(data, len), StringRef{}, false);
    }
  }
  return 0;
}
} // namespace

namespace {
int htp_hdr_valcb(http_parser *htp, const char *data, size_t len) {
  auto downstream = static_cast<Downstream *>(htp->data);
  auto &resp = downstream->response();
  auto &httpconf = get_config()->http;

  if (ensure_header_field_buffer(downstream, httpconf, len) != 0) {
    return -1;
  }

  if (downstream->get_response_state() == Downstream::INITIAL) {
    resp.fs.append_last_header_value(data, len);
  } else {
    resp.fs.append_last_trailer_value(data, len);
  }
  return 0;
}
} // namespace

namespace {
int htp_bodycb(http_parser *htp, const char *data, size_t len) {
  auto downstream = static_cast<Downstream *>(htp->data);
  auto &resp = downstream->response();

  resp.recv_body_length += len;

  return downstream->get_upstream()->on_downstream_body(
      downstream, reinterpret_cast<const uint8_t *>(data), len, true);
}
} // namespace

namespace {
int htp_msg_completecb(http_parser *htp) {
  auto downstream = static_cast<Downstream *>(htp->data);

  // http-parser does not treat "200 connection established" response
  // against CONNECT request, and in that case, this function is not
  // called.  But if HTTP Upgrade is made (e.g., WebSocket), this
  // function is called, and http_parser_execute() returns just after
  // that.
  if (downstream->get_upgraded()) {
    return 0;
  }

  if (downstream->get_non_final_response()) {
    downstream->reset_response();

    return 0;
  }

  downstream->set_response_state(Downstream::MSG_COMPLETE);
  // Block reading another response message from (broken?)
  // server. This callback is not called if the connection is
  // tunneled.
  downstream->pause_read(SHRPX_MSG_BLOCK);
  return downstream->get_upstream()->on_downstream_body_complete(downstream);
}
} // namespace

namespace {
http_parser_settings htp_hooks = {
    htp_msg_begincb,     // http_cb on_message_begin;
    nullptr,             // http_data_cb on_url;
    nullptr,             // http_data_cb on_status;
    htp_hdr_keycb,       // http_data_cb on_header_field;
    htp_hdr_valcb,       // http_data_cb on_header_value;
    htp_hdrs_completecb, // http_cb      on_headers_complete;
    htp_bodycb,          // http_data_cb on_body;
    htp_msg_completecb   // http_cb      on_message_complete;
};
} // namespace

int HttpDownstreamConnection::read_clear() {
  ev_timer_again(conn_.loop, &conn_.rt);
  std::array<uint8_t, 16_k> buf;
  int rv;

  for (;;) {
    auto nread = conn_.read_clear(buf.data(), buf.size());
    if (nread == 0) {
      return 0;
    }

    if (nread < 0) {
      return nread;
    }

    rv = process_input(buf.data(), nread);
    if (rv != 0) {
      return rv;
    }

    if (!ev_is_active(&conn_.rev)) {
      return 0;
    }
  }
}

int HttpDownstreamConnection::write_clear() {
  ev_timer_again(conn_.loop, &conn_.rt);

  auto upstream = downstream_->get_upstream();
  auto input = downstream_->get_request_buf();

  std::array<struct iovec, MAX_WR_IOVCNT> iov;

  while (input->rleft() > 0) {
    auto iovcnt = input->riovec(iov.data(), iov.size());

    auto nwrite = conn_.writev_clear(iov.data(), iovcnt);

    if (nwrite == 0) {
      return 0;
    }

    if (nwrite < 0) {
      return nwrite;
    }

    input->drain(nwrite);
  }

  conn_.wlimit.stopw();
  ev_timer_stop(conn_.loop, &conn_.wt);

  if (input->rleft() == 0) {
    auto &req = downstream_->request();

    upstream->resume_read(SHRPX_NO_BUFFER, downstream_,
                          req.unconsumed_body_length);
  }

  return 0;
}

int HttpDownstreamConnection::tls_handshake() {
  ERR_clear_error();

  ev_timer_again(conn_.loop, &conn_.rt);

  auto rv = conn_.tls_handshake();
  if (rv == SHRPX_ERR_INPROGRESS) {
    return 0;
  }

  if (rv < 0) {
    return rv;
  }

  if (LOG_ENABLED(INFO)) {
    DCLOG(INFO, this) << "SSL/TLS handshake completed";
  }

  if (!get_config()->tls.insecure &&
      ssl::check_cert(conn_.tls.ssl, addr_) != 0) {
    return -1;
  }

  if (!SSL_session_reused(conn_.tls.ssl)) {
    auto session = SSL_get0_session(conn_.tls.ssl);
    if (session) {
      ssl::try_cache_tls_session(addr_, session, ev_now(conn_.loop));
    }
  }

  do_read_ = &HttpDownstreamConnection::read_tls;
  do_write_ = &HttpDownstreamConnection::write_tls;

  // TODO Check negotiated ALPN

  return on_write();
}

int HttpDownstreamConnection::read_tls() {
  ERR_clear_error();

  ev_timer_again(conn_.loop, &conn_.rt);
  std::array<uint8_t, 16_k> buf;
  int rv;

  for (;;) {
    auto nread = conn_.read_tls(buf.data(), buf.size());
    if (nread == 0) {
      return 0;
    }

    if (nread < 0) {
      return nread;
    }

    rv = process_input(buf.data(), nread);
    if (rv != 0) {
      return rv;
    }

    if (!ev_is_active(&conn_.rev)) {
      return 0;
    }
  }
}

int HttpDownstreamConnection::write_tls() {
  ERR_clear_error();

  ev_timer_again(conn_.loop, &conn_.rt);

  auto upstream = downstream_->get_upstream();
  auto input = downstream_->get_request_buf();

  struct iovec iov;

  while (input->rleft() > 0) {
    auto iovcnt = input->riovec(&iov, 1);
    assert(iovcnt == 1);
    auto nwrite = conn_.write_tls(iov.iov_base, iov.iov_len);

    if (nwrite == 0) {
      return 0;
    }

    if (nwrite < 0) {
      return nwrite;
    }

    input->drain(nwrite);
  }

  conn_.wlimit.stopw();
  ev_timer_stop(conn_.loop, &conn_.wt);

  if (input->rleft() == 0) {
    auto &req = downstream_->request();

    upstream->resume_read(SHRPX_NO_BUFFER, downstream_,
                          req.unconsumed_body_length);
  }

  return 0;
}

int HttpDownstreamConnection::process_input(const uint8_t *data,
                                            size_t datalen) {
  int rv;

  if (downstream_->get_upgraded()) {
    // For upgraded connection, just pass data to the upstream.
    rv = downstream_->get_upstream()->on_downstream_body(downstream_, data,
                                                         datalen, true);
    if (rv != 0) {
      return rv;
    }

    if (downstream_->response_buf_full()) {
      downstream_->pause_read(SHRPX_NO_BUFFER);
      return 0;
    }

    return 0;
  }

  auto nproc =
      http_parser_execute(&response_htp_, &htp_hooks,
                          reinterpret_cast<const char *>(data), datalen);

  auto htperr = HTTP_PARSER_ERRNO(&response_htp_);

  if (htperr != HPE_OK) {
    // Handling early return (in other words, response was hijacked by
    // mruby scripting).
    if (downstream_->get_response_state() == Downstream::MSG_COMPLETE) {
      return SHRPX_ERR_DCONN_CANCELED;
    }

    if (LOG_ENABLED(INFO)) {
      DCLOG(INFO, this) << "HTTP parser failure: "
                        << "(" << http_errno_name(htperr) << ") "
                        << http_errno_description(htperr);
    }

    return -1;
  }

  if (downstream_->get_upgraded()) {
    if (nproc < datalen) {
      // Data from data + nproc are for upgraded protocol.
      rv = downstream_->get_upstream()->on_downstream_body(
          downstream_, data + nproc, datalen - nproc, true);
      if (rv != 0) {
        return rv;
      }

      if (downstream_->response_buf_full()) {
        downstream_->pause_read(SHRPX_NO_BUFFER);
        return 0;
      }
    }
    return 0;
  }

  if (downstream_->response_buf_full()) {
    downstream_->pause_read(SHRPX_NO_BUFFER);
    return 0;
  }

  return 0;
}

int HttpDownstreamConnection::connected() {
  auto connect_blocker = addr_->connect_blocker;

  if (!util::check_socket_connected(conn_.fd)) {
    conn_.wlimit.stopw();

    if (LOG_ENABLED(INFO)) {
      DCLOG(INFO, this) << "Backend connect failed; addr="
                        << util::to_numeric_addr(&addr_->addr);
    }

    connect_blocker->on_failure();

    downstream_->set_request_state(Downstream::CONNECT_FAIL);

    return -1;
  }

  if (LOG_ENABLED(INFO)) {
    DCLOG(INFO, this) << "Connected to downstream host";
  }

  connect_blocker->on_success();

  conn_.rlimit.startw();

  ev_set_cb(&conn_.wev, writecb);

  if (conn_.tls.ssl) {
    do_read_ = &HttpDownstreamConnection::tls_handshake;
    do_write_ = &HttpDownstreamConnection::tls_handshake;

    return 0;
  }

  do_read_ = &HttpDownstreamConnection::read_clear;
  do_write_ = &HttpDownstreamConnection::write_clear;

  return 0;
}

int HttpDownstreamConnection::on_read() { return do_read_(*this); }

int HttpDownstreamConnection::on_write() { return do_write_(*this); }

void HttpDownstreamConnection::on_upstream_change(Upstream *upstream) {}

void HttpDownstreamConnection::signal_write() {
  ev_feed_event(conn_.loop, &conn_.wev, EV_WRITE);
}

size_t HttpDownstreamConnection::get_group() const { return group_; }

int HttpDownstreamConnection::noop() { return 0; }

} // namespace shrpx