<commit_before>/* Copyright 2016-present Facebook, Inc.
 * Licensed under the Apache License, Version 2.0 */

#include "WatchmanConnection.h"

#include <folly/ExceptionWrapper.h>
#include <folly/SocketAddress.h>
#include <folly/Subprocess.h>
#include <folly/executors/InlineExecutor.h>
#include <folly/experimental/bser/Bser.h>

namespace watchman {

using namespace folly::bser;
using namespace folly;

// Ordered with the most likely kind first
static const std::vector<dynamic> kUnilateralLabels{"subscription", "log"};

static const dynamic kError("error");
static const dynamic kCapabilities("capabilities");

// We'll just dispatch bser decodes and callbacks inline unless they
// give us an alternative environment
static InlineExecutor inlineExecutor;

WatchmanConnection::WatchmanConnection(
    EventBase* eventBase,
    Optional<std::string>&& sockPath,
    Optional<WatchmanConnection::Callback>&& callback,
    Executor* cpuExecutor)
    : eventBase_(eventBase),
      sockPath_(std::move(sockPath)),
      callback_(std::move(callback)),
      cpuExecutor_(cpuExecutor ? cpuExecutor : &inlineExecutor),
      versionCmd_(nullptr),
      bufQ_(IOBufQueue::cacheChainLength()) {
  CHECK_NOTNULL(eventBase);
}

WatchmanConnection::~WatchmanConnection() {
  close();
}

folly::Future<std::string> WatchmanConnection::getSockPath() {
  // Take explicit configuration first
  if (sockPath_.hasValue()) {
    return makeFuture(sockPath_.value());
  }

  // Else use the environmental variable used by watchman to report
  // the active socket path
  auto var = getenv("WATCHMAN_SOCK");
  if (var && *var) {
    return makeFuture(std::string(var));
  }

  return via(cpuExecutor_, [] {
    // Else discover it from the CLI
    folly::Subprocess proc(
        {"watchman", "--output-encoding=bser", "get-sockname"},
        folly::Subprocess::Options().pipeStdout().pipeStderr().usePath());
    SCOPE_FAIL {
      // Always clean up to avoid Subprocess asserting on destruction
      proc.kill();
      proc.wait();
    };
    auto out_pair = proc.communicate();
    auto result = parseBser(out_pair.first);
    proc.waitChecked();
    return result["sockname"].asString();
  });
}

Future<dynamic> WatchmanConnection::connect(folly::dynamic versionArgs) {
  if (!versionArgs.isObject()) {
    throw WatchmanError("versionArgs must be object");
  }
  versionCmd_ = folly::dynamic::array("version", versionArgs);

  auto res = getSockPath().thenValue(
      [shared_this = shared_from_this()](std::string&& path) {
        shared_this->eventBase_->runInEventBaseThread([=] {
          folly::SocketAddress addr;
          addr.setFromPath(path);

          shared_this->sock_ =
              folly::AsyncSocket::newSocket(shared_this->eventBase_);
          shared_this->sock_->connect(shared_this.get(), addr);
        });

        return shared_this->connectPromise_.getFuture();
      });
  return res;
}

void WatchmanConnection::close() {
  if (closing_) {
    return;
  }
  closing_ = true;
  if (sock_) {
    eventBase_->runImmediatelyOrRunInEventBaseThreadAndWait([this] {
      // This implicitly closes the connection without flushing outstanding
      // writes. Should be fine as Watchman is mostly just providing info, so
      // an incomplete partial write isn't a problem. Doing a fully flushing
      // close here might be the cause a deadlock accessing the event base.
      sock_.reset();
    });
  }
  failQueuedCommands(
      make_exception_wrapper<WatchmanError>(
          "WatchmanConnection::close() was called"));
}

// The convention for Watchman responses is that they represent
// an error if they contain the "error" key.  We want to report
// those as exceptions, but it is easier to do that via a Try
Try<dynamic> WatchmanConnection::watchmanResponseToTry(dynamic&& value) {
  auto error = value.get_ptr(kError);
  if (error) {
    return Try<dynamic>(make_exception_wrapper<WatchmanResponseError>(value));
  }
  return Try<dynamic>(std::move(value));
}

void WatchmanConnection::connectSuccess() noexcept {
  try {
    sock_->setReadCB(this);
    sock_->setCloseOnExec();

    run(versionCmd_)
        .thenValue([shared_this = shared_from_this()](dynamic&& result) {
          // If there is no "capabilities" key then the version of
          // watchman is too old; treat this as an error
          if (!result.get_ptr(kCapabilities)) {
            result["error"] =
                "This watchman server has no support for capabilities, "
                "please upgrade to the current stable version of watchman";
            shared_this->connectPromise_.setTry(
                shared_this->watchmanResponseToTry(std::move(result)));
            return;
          }
          shared_this->connectPromise_.setValue(std::move(result));
        })
        .onError([shared_this =
                      shared_from_this()](const folly::exception_wrapper& e) {
          shared_this->connectPromise_.setException(e);
        });
  } catch(const std::exception& e) {
    connectPromise_.setException(
      folly::exception_wrapper(std::current_exception(), e));
  } catch(...) {<commit_msg>Replace Future::onError with Future::thenError<commit_after>/* Copyright 2016-present Facebook, Inc.
 * Licensed under the Apache License, Version 2.0 */

#include "WatchmanConnection.h"

#include <folly/ExceptionWrapper.h>
#include <folly/SocketAddress.h>
#include <folly/Subprocess.h>
#include <folly/executors/InlineExecutor.h>
#include <folly/experimental/bser/Bser.h>

namespace watchman {

using namespace folly::bser;
using namespace folly;

// Ordered with the most likely kind first
static const std::vector<dynamic> kUnilateralLabels{"subscription", "log"};

static const dynamic kError("error");
static const dynamic kCapabilities("capabilities");

// We'll just dispatch bser decodes and callbacks inline unless they
// give us an alternative environment
static InlineExecutor inlineExecutor;

WatchmanConnection::WatchmanConnection(
    EventBase* eventBase,
    Optional<std::string>&& sockPath,
    Optional<WatchmanConnection::Callback>&& callback,
    Executor* cpuExecutor)
    : eventBase_(eventBase),
      sockPath_(std::move(sockPath)),
      callback_(std::move(callback)),
      cpuExecutor_(cpuExecutor ? cpuExecutor : &inlineExecutor),
      versionCmd_(nullptr),
      bufQ_(IOBufQueue::cacheChainLength()) {
  CHECK_NOTNULL(eventBase);
}

WatchmanConnection::~WatchmanConnection() {
  close();
}

folly::Future<std::string> WatchmanConnection::getSockPath() {
  // Take explicit configuration first
  if (sockPath_.hasValue()) {
    return makeFuture(sockPath_.value());
  }

  // Else use the environmental variable used by watchman to report
  // the active socket path
  auto var = getenv("WATCHMAN_SOCK");
  if (var && *var) {
    return makeFuture(std::string(var));
  }

  return via(cpuExecutor_, [] {
    // Else discover it from the CLI
    folly::Subprocess proc(
        {"watchman", "--output-encoding=bser", "get-sockname"},
        folly::Subprocess::Options().pipeStdout().pipeStderr().usePath());
    SCOPE_FAIL {
      // Always clean up to avoid Subprocess asserting on destruction
      proc.kill();
      proc.wait();
    };
    auto out_pair = proc.communicate();
    auto result = parseBser(out_pair.first);
    proc.waitChecked();
    return result["sockname"].asString();
  });
}

Future<dynamic> WatchmanConnection::connect(folly::dynamic versionArgs) {
  if (!versionArgs.isObject()) {
    throw WatchmanError("versionArgs must be object");
  }
  versionCmd_ = folly::dynamic::array("version", versionArgs);

  auto res = getSockPath().thenValue(
      [shared_this = shared_from_this()](std::string&& path) {
        shared_this->eventBase_->runInEventBaseThread([=] {
          folly::SocketAddress addr;
          addr.setFromPath(path);

          shared_this->sock_ =
              folly::AsyncSocket::newSocket(shared_this->eventBase_);
          shared_this->sock_->connect(shared_this.get(), addr);
        });

        return shared_this->connectPromise_.getFuture();
      });
  return res;
}

void WatchmanConnection::close() {
  if (closing_) {
    return;
  }
  closing_ = true;
  if (sock_) {
    eventBase_->runImmediatelyOrRunInEventBaseThreadAndWait([this] {
      // This implicitly closes the connection without flushing outstanding
      // writes. Should be fine as Watchman is mostly just providing info, so
      // an incomplete partial write isn't a problem. Doing a fully flushing
      // close here might be the cause a deadlock accessing the event base.
      sock_.reset();
    });
  }
  failQueuedCommands(
      make_exception_wrapper<WatchmanError>(
          "WatchmanConnection::close() was called"));
}

// The convention for Watchman responses is that they represent
// an error if they contain the "error" key.  We want to report
// those as exceptions, but it is easier to do that via a Try
Try<dynamic> WatchmanConnection::watchmanResponseToTry(dynamic&& value) {
  auto error = value.get_ptr(kError);
  if (error) {
    return Try<dynamic>(make_exception_wrapper<WatchmanResponseError>(value));
  }
  return Try<dynamic>(std::move(value));
}

void WatchmanConnection::connectSuccess() noexcept {
  try {
    sock_->setReadCB(this);
    sock_->setCloseOnExec();

    run(versionCmd_)
        .thenValue([shared_this = shared_from_this()](dynamic&& result) {
          // If there is no "capabilities" key then the version of
          // watchman is too old; treat this as an error
          if (!result.get_ptr(kCapabilities)) {
            result["error"] =
                "This watchman server has no support for capabilities, "
                "please upgrade to the current stable version of watchman";
            shared_this->connectPromise_.setTry(
                shared_this->watchmanResponseToTry(std::move(result)));
            return;
          }
          shared_this->connectPromise_.setValue(std::move(result));
        })
        .thenError([shared_this =
                        shared_from_this()](const folly::exception_wrapper& e) {
          shared_this->connectPromise_.setException(e);
        });
  } catch(const std::exception& e) {
    connectPromise_.setException(
      folly::exception_wrapper(std::current_exception(), e));
  } catch(...) {
{'fraction_non_alphanumeric': 0.12061765562785277, 'fraction_numerical': 0.0011653879770807032, 'mean_word_length': 3.5606731620903456, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 61, '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': '4646918', 'n_tokens_mistral': 2924, 'n_tokens_neox': 2861, 'n_words': 719}