<commit_before>////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2016-2018 ArangoDB GmbH, Cologne, Germany
///
/// 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.
///
/// Copyright holder is ArangoDB GmbH, Cologne, Germany
///
/// @author <NAME>
/// @author <NAME>
/// @author <NAME>
////////////////////////////////////////////////////////////////////////////////

#include "VstConnection.h"

#include <fuerte/helper.h>
#include <fuerte/loop.h>
#include <fuerte/message.h>
#include <fuerte/types.h>
#include <velocypack/velocypack-aliases.h>

#include "debugging.h"

namespace arangodb { namespace fuerte { inline namespace v1 { namespace vst {
namespace fu = arangodb::fuerte::v1;
using arangodb::fuerte::v1::SocketType;

template <SocketType ST>
VstConnection<ST>::VstConnection(
    EventLoopService& loop, fu::detail::ConnectionConfiguration const& config)
    : fuerte::GeneralConnection<ST, vst::RequestItem>(loop, config),
      _numMessages(0),
      _vstVersion(config._vstVersion) {}

template <SocketType ST>
VstConnection<ST>::~VstConnection() try {
  abortOngoingRequests(Error::Canceled);
} catch (...) {
}

template <SocketType ST>
std::size_t VstConnection<ST>::requestsLeft() const {
  uint32_t qd = this->_numQueued.load(std::memory_order_relaxed);
  qd += _numMessages.load(std::memory_order_relaxed);
  FUERTE_LOG_VSTTRACE << "requestsLeft = " << qd << "\n";
  return qd;
}

// -----------------------------------------------------------------------------
// --SECTION--                                                   private methods
// -----------------------------------------------------------------------------

// socket connection is up (with optional SSL), now initiate the VST protocol.
template <SocketType ST>
void VstConnection<ST>::finishConnect() {
  if (this->state() != Connection::State::Connecting) {
    return;
  }
  
  FUERTE_LOG_VSTTRACE << "finishInitialization (vst)\n";
  const char* vstHeader;
  switch (_vstVersion) {
    case VST1_0:
      vstHeader = "VST/1.0\r\n\r\n";
      break;
    case VST1_1:
      vstHeader = "VST/1.1\r\n\r\n";
      break;
    default:
      throw std::logic_error("Unknown VST version");
  }

  auto self = Connection::shared_from_this();
  asio_ns::async_write(
      this->_proto.socket, asio_ns::buffer(vstHeader, strlen(vstHeader)),
      [self](asio_ns::error_code const& ec, std::size_t nsend) {
        auto* me = static_cast<VstConnection<ST>*>(self.get());
        if (ec || !me->_active.load()) {
          FUERTE_LOG_ERROR << ec.message() << "\n";
          me->shutdownConnection(Error::CouldNotConnect,
                                 "unable to connect: " + ec.message());
          return;
        }
        FUERTE_LOG_VSTTRACE << "VST connection established\n";
        if (me->_config._authenticationType != AuthenticationType::None) {
          // send the auth, then set _state == connected
          me->sendAuthenticationRequest();
        } else {
          me->_state.store(Connection::State::Connected);
          me->asyncWriteNextRequest();
        }
      });
}

// Send out the authentication message on this connection
template <SocketType ST>
void VstConnection<ST>::sendAuthenticationRequest() {
  FUERTE_ASSERT(this->_config._authenticationType != AuthenticationType::None);
  FUERTE_ASSERT(this->_active.load());

  // Part 1: Build ArangoDB VST auth message (1000)
  auto self = Connection::shared_from_this();
  auto item = std::make_unique<RequestItem>(nullptr,
                                            [self](Error error, std::unique_ptr<Request>,
                                                   std::unique_ptr<Response> resp) {
    auto& me = static_cast<VstConnection<ST>&>(*self);
    if (error != Error::NoError || resp->statusCode() != StatusOK) {
      me.shutdownConnection(Error::VstUnauthorized, "could not authenticate");
    } else {
      me._state.store(Connection::State::Connected);
      if (!me._writing) {
        me.asyncWriteNextRequest();
      }
    }
  });
  item->expires = std::chrono::steady_clock::now() + Request::defaultTimeout;
  
  if (this->_config._authenticationType == AuthenticationType::Basic) {
     vst::message::authBasic(this->_config._user, this->_config._password,
                             item->_buffer);
   } else if (this->_config._authenticationType == AuthenticationType::Jwt) {
     vst::message::authJWT(this->_config._jwtToken, item->_buffer);
   }
   FUERTE_ASSERT(item->_buffer.size() < defaultMaxChunkSize);

   // actually send auth request
   std::vector<asio_ns::const_buffer> buffers;
   vst::message::prepareForNetwork(_vstVersion, item->_messageID, item->_buffer,
                                   /*payload*/ asio_ns::const_buffer(), buffers);

  RequestItem* ptr = item.get();
  _messages.emplace(item->_messageID, std::move(item));  // add message to store
  _numMessages.fetch_add(1, std::memory_order_relaxed);

  this->_writing = true;
  this->setIOTimeout(); // has to be after setting _writing
  asio_ns::async_write(this->_proto.socket, buffers,
                       [self](auto const& ec, std::size_t nsend) {
    auto& me = static_cast<VstConnection<ST>&>(*self);
    me._writing = false;
    if (ec) {
      me.shutdownConnection(Error::CouldNotConnect, "authorization failed");
    } else {
      me.asyncReadSome(); // read the response
    }
  });
}

// ------------------------------------
// Writing data
// ------------------------------------

// writes data from task queue to network using asio_ns::async_write
template <SocketType ST>
void VstConnection<ST>::asyncWriteNextRequest() {
  FUERTE_LOG_VSTTRACE << "asyncWrite: preparing to send next\n";
  
  auto state = this->_state.load();
  FUERTE_ASSERT(state == Connection::State::Connected ||<commit_msg>fix compile warning in fuerte compilation<commit_after>////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2016-2018 ArangoDB GmbH, Cologne, Germany
///
/// 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.
///
/// Copyright holder is ArangoDB GmbH, Cologne, Germany
///
/// @author <NAME>
/// @author <NAME>
/// @author <NAME>
////////////////////////////////////////////////////////////////////////////////

#include "VstConnection.h"

#include <fuerte/helper.h>
#include <fuerte/loop.h>
#include <fuerte/message.h>
#include <fuerte/types.h>
#include <velocypack/velocypack-aliases.h>

#include "debugging.h"

namespace arangodb { namespace fuerte { inline namespace v1 { namespace vst {
namespace fu = arangodb::fuerte::v1;
using arangodb::fuerte::v1::SocketType;

template <SocketType ST>
VstConnection<ST>::VstConnection(
    EventLoopService& loop, fu::detail::ConnectionConfiguration const& config)
    : fuerte::GeneralConnection<ST, vst::RequestItem>(loop, config),
      _numMessages(0),
      _vstVersion(config._vstVersion) {}

template <SocketType ST>
VstConnection<ST>::~VstConnection() try {
  abortOngoingRequests(Error::Canceled);
} catch (...) {
}

template <SocketType ST>
std::size_t VstConnection<ST>::requestsLeft() const {
  uint32_t qd = this->_numQueued.load(std::memory_order_relaxed);
  qd += _numMessages.load(std::memory_order_relaxed);
  FUERTE_LOG_VSTTRACE << "requestsLeft = " << qd << "\n";
  return qd;
}

// -----------------------------------------------------------------------------
// --SECTION--                                                   private methods
// -----------------------------------------------------------------------------

// socket connection is up (with optional SSL), now initiate the VST protocol.
template <SocketType ST>
void VstConnection<ST>::finishConnect() {
  if (this->state() != Connection::State::Connecting) {
    return;
  }
  
  FUERTE_LOG_VSTTRACE << "finishInitialization (vst)\n";
  const char* vstHeader;
  switch (_vstVersion) {
    case VST1_0:
      vstHeader = "VST/1.0\r\n\r\n";
      break;
    case VST1_1:
      vstHeader = "VST/1.1\r\n\r\n";
      break;
    default:
      throw std::logic_error("Unknown VST version");
  }

  auto self = Connection::shared_from_this();
  asio_ns::async_write(
      this->_proto.socket, asio_ns::buffer(vstHeader, strlen(vstHeader)),
      [self](asio_ns::error_code const& ec, std::size_t nsend) {
        auto* me = static_cast<VstConnection<ST>*>(self.get());
        if (ec || !me->_active.load()) {
          FUERTE_LOG_ERROR << ec.message() << "\n";
          me->shutdownConnection(Error::CouldNotConnect,
                                 "unable to connect: " + ec.message());
          return;
        }
        FUERTE_LOG_VSTTRACE << "VST connection established\n";
        if (me->_config._authenticationType != AuthenticationType::None) {
          // send the auth, then set _state == connected
          me->sendAuthenticationRequest();
        } else {
          me->_state.store(Connection::State::Connected);
          me->asyncWriteNextRequest();
        }
      });
}

// Send out the authentication message on this connection
template <SocketType ST>
void VstConnection<ST>::sendAuthenticationRequest() {
  FUERTE_ASSERT(this->_config._authenticationType != AuthenticationType::None);
  FUERTE_ASSERT(this->_active.load());

  // Part 1: Build ArangoDB VST auth message (1000)
  auto self = Connection::shared_from_this();
  auto item = std::make_unique<RequestItem>(nullptr,
                                            [self](Error error, std::unique_ptr<Request>,
                                                   std::unique_ptr<Response> resp) {
    auto& me = static_cast<VstConnection<ST>&>(*self);
    if (error != Error::NoError || resp->statusCode() != StatusOK) {
      me.shutdownConnection(Error::VstUnauthorized, "could not authenticate");
    } else {
      me._state.store(Connection::State::Connected);
      if (!me._writing) {
        me.asyncWriteNextRequest();
      }
    }
  });
  item->expires = std::chrono::steady_clock::now() + Request::defaultTimeout;
  
  if (this->_config._authenticationType == AuthenticationType::Basic) {
     vst::message::authBasic(this->_config._user, this->_config._password,
                             item->_buffer);
   } else if (this->_config._authenticationType == AuthenticationType::Jwt) {
     vst::message::authJWT(this->_config._jwtToken, item->_buffer);
   }
   FUERTE_ASSERT(item->_buffer.size() < defaultMaxChunkSize);

   // actually send auth request
   std::vector<asio_ns::const_buffer> buffers;
   vst::message::prepareForNetwork(_vstVersion, item->_messageID, item->_buffer,
                                   /*payload*/ asio_ns::const_buffer(), buffers);

  _messages.emplace(item->_messageID, std::move(item));  // add message to store
  _numMessages.fetch_add(1, std::memory_order_relaxed);

  this->_writing = true;
  this->setIOTimeout(); // has to be after setting _writing
  asio_ns::async_write(this->_proto.socket, buffers,
                       [self](auto const& ec, std::size_t nsend) {
    auto& me = static_cast<VstConnection<ST>&>(*self);
    me._writing = false;
    if (ec) {
      me.shutdownConnection(Error::CouldNotConnect, "authorization failed");
    } else {
      me.asyncReadSome(); // read the response
    }
  });
}

// ------------------------------------
// Writing data
// ------------------------------------

// writes data from task queue to network using asio_ns::async_write
template <SocketType ST>
void VstConnection<ST>::asyncWriteNextRequest() {
  FUERTE_LOG_VSTTRACE << "asyncWrite: preparing to send next\n";
  
  auto state = this->_state.load();
  FUERTE_ASSERT(state == Connection::State::Connected ||
{'fraction_non_alphanumeric': 0.1925470731899472, 'fraction_numerical': 0.0050421492161033645, 'mean_word_length': 2.5192680898253395, 'pattern_counts': {'":': 0, '<': 99, '<?xml version=': 0, '>': 121, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 17, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '958157', 'n_tokens_mistral': 3674, 'n_tokens_neox': 3599, 'n_words': 816}