<commit_before>// Copyright 2004-present Facebook. All Rights Reserved.

#include "rsocket/transports/tcp/TcpConnectionFactory.h"

#include <folly/io/async/AsyncSSLSocket.h>
#include <folly/io/async/AsyncSocket.h>
#include <folly/io/async/AsyncTransport.h>
#include <folly/io/async/EventBaseManager.h>
#include <glog/logging.h>

#include "rsocket/transports/tcp/TcpDuplexConnection.h"

using namespace rsocket;

namespace rsocket {

namespace {

class ConnectCallback : public folly::AsyncSocket::ConnectCallback {
 public:
  ConnectCallback(
      folly::SocketAddress address,
      const std::shared_ptr<folly::SSLContext>& sslContext,
      folly::Promise<ConnectionFactory::ConnectedDuplexConnection>
          connectPromise)
      : address_(address), connectPromise_(std::move(connectPromise)) {
    VLOG(2) << "Constructing ConnectCallback";

    // Set up by ScopedEventBaseThread.
    auto evb = folly::EventBaseManager::get()->getExistingEventBase();
    DCHECK(evb);

    if (sslContext) {
      VLOG(3) << "Starting SSL socket";
      sslContext->setAdvertisedNextProtocols({"rs"});
      socket_.reset(new folly::AsyncSSLSocket(sslContext, evb));
    } else {
      VLOG(3) << "Starting socket";
      socket_.reset(new folly::AsyncSocket(evb));
    }

    VLOG(3) << "Attempting connection to " << address_;

    socket_->connect(this, address_);
  }

  ~ConnectCallback() {
    VLOG(2) << "Destroying ConnectCallback";
  }

  void connectSuccess() noexcept override {
    std::unique_ptr<ConnectCallback> deleter(this);
    VLOG(4) << "connectSuccess() on " << address_;

    auto connection = TcpConnectionFactory::createDuplexConnectionFromSocket(
        std::move(socket_), RSocketStats::noop());
    auto evb = folly::EventBaseManager::get()->getExistingEventBase();
    CHECK(evb);
    connectPromise_.setValue(ConnectionFactory::ConnectedDuplexConnection{
        std::move(connection), *evb});
  }

  void connectErr(const folly::AsyncSocketException& ex) noexcept override {
    std::unique_ptr<ConnectCallback> deleter(this);
    VLOG(4) << "connectErr(" << ex.what() << ") on " << address_;
    connectPromise_.setException(ex);
  }

 private:
  const folly::SocketAddress address_;
  folly::AsyncSocket::UniquePtr socket_;
  folly::Promise<ConnectionFactory::ConnectedDuplexConnection> connectPromise_;
};

} // namespace

TcpConnectionFactory::TcpConnectionFactory(
    folly::EventBase& eventBase,
    folly::SocketAddress address,
    std::shared_ptr<folly::SSLContext> sslContext)
    : eventBase_(&eventBase),
      address_(std::move(address)),
      sslContext_(std::move(sslContext)) {
  VLOG(1) << "Constructing TcpConnectionFactory";
}

TcpConnectionFactory::~TcpConnectionFactory() {
  VLOG(1) << "Destroying TcpConnectionFactory";
}

folly::Future<ConnectionFactory::ConnectedDuplexConnection>
TcpConnectionFactory::connect() {
  folly::Promise<ConnectionFactory::ConnectedDuplexConnection> connectPromise;
  auto connectFuture = connectPromise.getFuture();

  eventBase_->runInEventBaseThread(
      [this, promise = std::move(connectPromise)]() mutable {
        new ConnectCallback(address_, sslContext_, std::move(promise));
      });
  return connectFuture;
}

std::unique_ptr<DuplexConnection>
TcpConnectionFactory::createDuplexConnectionFromSocket(
    folly::AsyncTransportWrapper::UniquePtr socket,<commit_msg>Remove spammy log<commit_after>// Copyright 2004-present Facebook. All Rights Reserved.

#include "rsocket/transports/tcp/TcpConnectionFactory.h"

#include <folly/io/async/AsyncSSLSocket.h>
#include <folly/io/async/AsyncSocket.h>
#include <folly/io/async/AsyncTransport.h>
#include <folly/io/async/EventBaseManager.h>
#include <glog/logging.h>

#include "rsocket/transports/tcp/TcpDuplexConnection.h"

using namespace rsocket;

namespace rsocket {

namespace {

class ConnectCallback : public folly::AsyncSocket::ConnectCallback {
 public:
  ConnectCallback(
      folly::SocketAddress address,
      const std::shared_ptr<folly::SSLContext>& sslContext,
      folly::Promise<ConnectionFactory::ConnectedDuplexConnection>
          connectPromise)
      : address_(address), connectPromise_(std::move(connectPromise)) {
    VLOG(2) << "Constructing ConnectCallback";

    // Set up by ScopedEventBaseThread.
    auto evb = folly::EventBaseManager::get()->getExistingEventBase();
    DCHECK(evb);

    if (sslContext) {
      VLOG(3) << "Starting SSL socket";
      sslContext->setAdvertisedNextProtocols({"rs"});
      socket_.reset(new folly::AsyncSSLSocket(sslContext, evb));
    } else {
      VLOG(3) << "Starting socket";
      socket_.reset(new folly::AsyncSocket(evb));
    }

    VLOG(3) << "Attempting connection to " << address_;

    socket_->connect(this, address_);
  }

  ~ConnectCallback() {
    VLOG(2) << "Destroying ConnectCallback";
  }

  void connectSuccess() noexcept override {
    std::unique_ptr<ConnectCallback> deleter(this);
    VLOG(4) << "connectSuccess() on " << address_;

    auto connection = TcpConnectionFactory::createDuplexConnectionFromSocket(
        std::move(socket_), RSocketStats::noop());
    auto evb = folly::EventBaseManager::get()->getExistingEventBase();
    CHECK(evb);
    connectPromise_.setValue(ConnectionFactory::ConnectedDuplexConnection{
        std::move(connection), *evb});
  }

  void connectErr(const folly::AsyncSocketException& ex) noexcept override {
    std::unique_ptr<ConnectCallback> deleter(this);
    VLOG(4) << "connectErr(" << ex.what() << ") on " << address_;
    connectPromise_.setException(ex);
  }

 private:
  const folly::SocketAddress address_;
  folly::AsyncSocket::UniquePtr socket_;
  folly::Promise<ConnectionFactory::ConnectedDuplexConnection> connectPromise_;
};

} // namespace

TcpConnectionFactory::TcpConnectionFactory(
    folly::EventBase& eventBase,
    folly::SocketAddress address,
    std::shared_ptr<folly::SSLContext> sslContext)
    : eventBase_(&eventBase),
      address_(std::move(address)),
      sslContext_(std::move(sslContext)) {
}

TcpConnectionFactory::~TcpConnectionFactory() = default;

folly::Future<ConnectionFactory::ConnectedDuplexConnection>
TcpConnectionFactory::connect() {
  folly::Promise<ConnectionFactory::ConnectedDuplexConnection> connectPromise;
  auto connectFuture = connectPromise.getFuture();

  eventBase_->runInEventBaseThread(
      [this, promise = std::move(connectPromise)]() mutable {
        new ConnectCallback(address_, sslContext_, std::move(promise));
      });
  return connectFuture;
}

std::unique_ptr<DuplexConnection>
TcpConnectionFactory::createDuplexConnectionFromSocket(
    folly::AsyncTransportWrapper::UniquePtr socket,
{'fraction_non_alphanumeric': 0.13055932458917532, 'fraction_numerical': 0.003618272274988693, 'mean_word_length': 5.324118207816968, 'pattern_counts': {'":': 0, '<': 83, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6880812', 'n_tokens_mistral': 2024, 'n_tokens_neox': 1978, 'n_words': 324}