<commit_before>/******************************************************************************\

          This file is part of the C! library.  A.K.A the cbang library.

                Copyright (c) 2003-2019, Cauldron Development LLC
                   Copyright (c) 2003-2017, Stanford University
                               All rights reserved.

         The C! library is free software: you can redistribute it and/or
        modify it under the terms of the GNU Lesser General Public License
       as published by the Free Software Foundation, either version 2.1 of
               the License, or (at your option) any later version.

        The C! library is distributed in the hope that it will be useful,
          but WITHOUT ANY WARRANTY; without even the implied warranty of
        MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
                 Lesser General Public License for more details.

         You should have received a copy of the GNU Lesser General Public
                 License along with the C! library.  If not, see
                         <http://www.gnu.org/licenses/>.

        In addition, BSD licensing may be granted on a case by case basis
        by written permission from at least one of the copyright holders.
           You may request written permission by emailing the authors.

                  For information regarding this software email:
                                 <NAME>
                          <EMAIL>

\******************************************************************************/

#include "Connection.h"
#include "Base.h"
#include "DNSBase.h"
#include "Request.h"
#include "HTTP.h"
#include "Event.h"
#include "Websocket.h"

#include <cbang/config.h>
#include <cbang/Exception.h>
#include <cbang/Catch.h>

#include <cbang/net/IPAddress.h>

#include <cbang/time/Timer.h>
#include <cbang/log/Logger.h>
#include <cbang/socket/Socket.h>
#include <cbang/os/SysError.h>
#include <cbang/openssl/SSLContext.h>
#include <cbang/util/RateSet.h>

#ifdef _WIN32
#include <cbang/socket/Winsock.h>

#define ERRNO_CONNECT_REFUSED WSAECONNREFUSED
#else
#define ERRNO_CONNECT_REFUSED ECONNREFUSED
#endif

using namespace std;
using namespace cb;
using namespace cb::Event;


#undef CBANG_LOG_PREFIX
#define CBANG_LOG_PREFIX << "CON" << getID() << ':'


Connection::Connection(cb::Event::Base &base, bool incoming,
                       const IPAddress &peer,
                       const SmartPointer<Socket> &socket,
                       const SmartPointer<SSLContext> &sslCtx) :
  BufferEvent(base, incoming, socket, sslCtx), base(base),
  state(incoming ? STATE_READING_FIRSTLINE : STATE_DISCONNECTED),
  incoming(incoming), peer(peer), startTime(Timer::now()), sslCtx(sslCtx) {

  LOG_DEBUG(4, "created " << getStateString(state));
}


Connection::~Connection() {
  if (retryEvent.isSet()) {
    TRY_CATCH_ERROR(retryEvent->del());
    retryEvent.release();
  }

  LOG_DEBUG(4, "destroyed " << getStateString(state));
}


bool Connection::isConnected() const {
  return state != STATE_DISCONNECTED && state != STATE_CONNECTING;
}


const SmartPointer<Request> &Connection::getRequest() const {
  if (requests.empty()) THROW("No request");
  return requests.front();
}


void Connection::checkActiveRequest(Request &req) const {
  if (getRequest() != &req) THROW("Not the active request");
}


bool Connection::isWebsocket() const {
  return requests.size() && dynamic_cast<Websocket *>(requests.front().get());
}


Websocket &Connection::getWebsocket() const {
  return *getRequest().cast<Websocket>();
}


void Connection::setStats(const SmartPointer<RateSet> &stats) {
  this->stats = stats;
  if (stats.isSet()) stats->event(incoming ? "incoming" : "outgoing");
}


void Connection::sendServiceUnavailable() {
  LOG_DEBUG(4, __func__ << "()");

  if (requests.size()) THROW("Not a new Connection");
  push(new Request);
  getRequest()->sendError(HTTP_SERVICE_UNAVAILABLE);
}


void Connection::makeRequest(Request &req) {
  LOG_DEBUG(4, __func__ << "()");

  if (req.isReplying()) THROW("Request already replying");

  bool first = requests.empty();
  push(&req);

  // If this is the active request dispatch it
  if (first) dispatch();
}


void Connection::acceptRequest() {
  LOG_DEBUG(4, __func__ << "()");

  if (requests.size()) THROW("Not a new Connection");
  startRead();
}


void Connection::cancelRequest(Request &req) {
  LOG_DEBUG(4, __func__ << "()");

  if (getRequest() == &req) fail(CONN_ERR_REQUEST_CANCEL);
  else requests.remove(&req);
}


void Connection::write(Request &req, const cb::Event::Buffer &buf) {
  LOG_DEBUG(4, __func__ << "() bytes=" << buf.getLength());
  checkActiveRequest(req);

  getOutput().add(buf);

  detectClose = false; // Stop detect close

  // Don't change state or disable read if active Websocket
  if (!req.isWebsocket()) {
    setRead(false);
    setState(STATE_WRITING);
  }
}


const char *Connection::getStateString(state_t state) {
  switch (state) {
  case STATE_DISCONNECTED:      return "DISCONNECTED";
  case STATE_CONNECTING:        return "CONNECTING";
  case STATE_IDLE:              return "IDLE";
  case STATE_READING_FIRSTLINE: return "READING_FIRSTLINE";
  case STATE_READING_HEADERS:   return "READING_HEADERS";
  case STATE_READING_BODY:      return "READING_BODY";
  case STATE_READING_TRAILER:   return "READING_TRAILER";
  case STATE_WEBSOCK_HEADER:    return "WEBSOCK_HEADER";
  case STATE_WEBSOCK_BODY:      return "WEBSOCK_BODY";
  case STATE_WRITING:           return "WRITING";
  }

  return "INVALID";
}


void Connection::setState(state_t state) {
  if (state == this->state) return;
  LOG_DEBUG(4, getStateString(this->state) << " -> " << getStateString(state));<commit_msg>Set writing content length<commit_after>/******************************************************************************\

          This file is part of the C! library.  A.K.A the cbang library.

                Copyright (c) 2003-2019, Cauldron Development LLC
                   Copyright (c) 2003-2017, Stanford University
                               All rights reserved.

         The C! library is free software: you can redistribute it and/or
        modify it under the terms of the GNU Lesser General Public License
       as published by the Free Software Foundation, either version 2.1 of
               the License, or (at your option) any later version.

        The C! library is distributed in the hope that it will be useful,
          but WITHOUT ANY WARRANTY; without even the implied warranty of
        MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
                 Lesser General Public License for more details.

         You should have received a copy of the GNU Lesser General Public
                 License along with the C! library.  If not, see
                         <http://www.gnu.org/licenses/>.

        In addition, BSD licensing may be granted on a case by case basis
        by written permission from at least one of the copyright holders.
           You may request written permission by emailing the authors.

                  For information regarding this software email:
                                 <NAME>
                          <EMAIL>

\******************************************************************************/

#include "Connection.h"
#include "Base.h"
#include "DNSBase.h"
#include "Request.h"
#include "HTTP.h"
#include "Event.h"
#include "Websocket.h"

#include <cbang/config.h>
#include <cbang/Exception.h>
#include <cbang/Catch.h>

#include <cbang/net/IPAddress.h>

#include <cbang/time/Timer.h>
#include <cbang/log/Logger.h>
#include <cbang/socket/Socket.h>
#include <cbang/os/SysError.h>
#include <cbang/openssl/SSLContext.h>
#include <cbang/util/RateSet.h>

#ifdef _WIN32
#include <cbang/socket/Winsock.h>

#define ERRNO_CONNECT_REFUSED WSAECONNREFUSED
#else
#define ERRNO_CONNECT_REFUSED ECONNREFUSED
#endif

using namespace std;
using namespace cb;
using namespace cb::Event;


#undef CBANG_LOG_PREFIX
#define CBANG_LOG_PREFIX << "CON" << getID() << ':'


Connection::Connection(cb::Event::Base &base, bool incoming,
                       const IPAddress &peer,
                       const SmartPointer<Socket> &socket,
                       const SmartPointer<SSLContext> &sslCtx) :
  BufferEvent(base, incoming, socket, sslCtx), base(base),
  state(incoming ? STATE_READING_FIRSTLINE : STATE_DISCONNECTED),
  incoming(incoming), peer(peer), startTime(Timer::now()), sslCtx(sslCtx) {

  LOG_DEBUG(4, "created " << getStateString(state));
}


Connection::~Connection() {
  if (retryEvent.isSet()) {
    TRY_CATCH_ERROR(retryEvent->del());
    retryEvent.release();
  }

  LOG_DEBUG(4, "destroyed " << getStateString(state));
}


bool Connection::isConnected() const {
  return state != STATE_DISCONNECTED && state != STATE_CONNECTING;
}


const SmartPointer<Request> &Connection::getRequest() const {
  if (requests.empty()) THROW("No request");
  return requests.front();
}


void Connection::checkActiveRequest(Request &req) const {
  if (getRequest() != &req) THROW("Not the active request");
}


bool Connection::isWebsocket() const {
  return requests.size() && dynamic_cast<Websocket *>(requests.front().get());
}


Websocket &Connection::getWebsocket() const {
  return *getRequest().cast<Websocket>();
}


void Connection::setStats(const SmartPointer<RateSet> &stats) {
  this->stats = stats;
  if (stats.isSet()) stats->event(incoming ? "incoming" : "outgoing");
}


void Connection::sendServiceUnavailable() {
  LOG_DEBUG(4, __func__ << "()");

  if (requests.size()) THROW("Not a new Connection");
  push(new Request);
  getRequest()->sendError(HTTP_SERVICE_UNAVAILABLE);
}


void Connection::makeRequest(Request &req) {
  LOG_DEBUG(4, __func__ << "()");

  if (req.isReplying()) THROW("Request already replying");

  bool first = requests.empty();
  push(&req);

  // If this is the active request dispatch it
  if (first) dispatch();
}


void Connection::acceptRequest() {
  LOG_DEBUG(4, __func__ << "()");

  if (requests.size()) THROW("Not a new Connection");
  startRead();
}


void Connection::cancelRequest(Request &req) {
  LOG_DEBUG(4, __func__ << "()");

  if (getRequest() == &req) fail(CONN_ERR_REQUEST_CANCEL);
  else requests.remove(&req);
}


void Connection::write(Request &req, const cb::Event::Buffer &buf) {
  LOG_DEBUG(4, __func__ << "() bytes=" << buf.getLength());
  checkActiveRequest(req);

  getOutput().add(buf);

  detectClose = false; // Stop detect close

  // Don't change state or disable read if active Websocket
  if (!req.isWebsocket()) {
    setRead(false);
    setState(STATE_WRITING);
    contentLength = getOutput().getLength();
  }
}


const char *Connection::getStateString(state_t state) {
  switch (state) {
  case STATE_DISCONNECTED:      return "DISCONNECTED";
  case STATE_CONNECTING:        return "CONNECTING";
  case STATE_IDLE:              return "IDLE";
  case STATE_READING_FIRSTLINE: return "READING_FIRSTLINE";
  case STATE_READING_HEADERS:   return "READING_HEADERS";
  case STATE_READING_BODY:      return "READING_BODY";
  case STATE_READING_TRAILER:   return "READING_TRAILER";
  case STATE_WEBSOCK_HEADER:    return "WEBSOCK_HEADER";
  case STATE_WEBSOCK_BODY:      return "WEBSOCK_BODY";
  case STATE_WRITING:           return "WRITING";
  }

  return "INVALID";
}


void Connection::setState(state_t state) {
  if (state == this->state) return;
  LOG_DEBUG(4, getStateString(this->state) << " -> " << getStateString(state));
{'fraction_non_alphanumeric': 0.14475281873373808, 'fraction_numerical': 0.004856895056374675, 'mean_word_length': 3.5613132911392404, 'pattern_counts': {'":': 0, '<': 95, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4488037', 'n_tokens_mistral': 3545, 'n_tokens_neox': 3414, 'n_words': 944}