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

#pragma once

#include <memory>
#include <mutex>
#include <utility>

namespace yarpl {
namespace flowable {
namespace details {

template <typename T>
class EmiterBase : public virtual Refcounted {
 public:
  ~EmiterBase() = default;

  virtual std::tuple<int64_t, bool> emit(Reference<Subscriber<T>>, int64_t) = 0;
};

/**
 * Manager for a flowable subscription.
 *
 * This is synchronous: the emit calls are triggered within the context
 * of a request(n) call.
 */
template <typename T>
class EmiterSubscription : public Subscription, public Subscriber<T> {
  constexpr static auto kCanceled = credits::kCanceled;
  constexpr static auto kNoFlowControl = credits::kNoFlowControl;

 public:
  EmiterSubscription(
      Reference<EmiterBase<T>> emiter,
      Reference<Subscriber<T>> subscriber)
      : emiter_(std::move(emiter)), subscriber_(std::move(subscriber)) {
    subscriber_->onSubscribe(this->ref_from_this(this));
  }

  virtual ~EmiterSubscription() {
    subscriber_.reset();
  }

  void request(int64_t delta) override {
    if (delta <= 0) {
      auto message = "request(n): " + std::to_string(delta) + " <= 0";
      throw std::logic_error(message);
    }

    while (true) {
      auto current = requested_.load(std::memory_order_relaxed);

      if (current == kCanceled) {
        // this can happen because there could be an async barrier between the
        // subscriber and the subscription for instance while onComplete is
        // being delivered (on effectively cancelled subscription) the
        // subscriber can call request(n)
        return;<commit_msg>Replace useage of std::to_string with folly:to (#751)<commit_after>// Copyright 2004-present Facebook. All Rights Reserved.

#pragma once

#include <memory>
#include <mutex>
#include <utility>

#include <folly/Conv.h>

namespace yarpl {
namespace flowable {
namespace details {

template <typename T>
class EmiterBase : public virtual Refcounted {
 public:
  ~EmiterBase() = default;

  virtual std::tuple<int64_t, bool> emit(Reference<Subscriber<T>>, int64_t) = 0;
};

/**
 * Manager for a flowable subscription.
 *
 * This is synchronous: the emit calls are triggered within the context
 * of a request(n) call.
 */
template <typename T>
class EmiterSubscription : public Subscription, public Subscriber<T> {
  constexpr static auto kCanceled = credits::kCanceled;
  constexpr static auto kNoFlowControl = credits::kNoFlowControl;

 public:
  EmiterSubscription(
      Reference<EmiterBase<T>> emiter,
      Reference<Subscriber<T>> subscriber)
      : emiter_(std::move(emiter)), subscriber_(std::move(subscriber)) {
    subscriber_->onSubscribe(this->ref_from_this(this));
  }

  virtual ~EmiterSubscription() {
    subscriber_.reset();
  }

  void request(int64_t delta) override {
    if (delta <= 0) {
      auto message = "request(n): " + folly::to<std::string>(delta) + " <= 0";
      throw std::logic_error(message);
    }

    while (true) {
      auto current = requested_.load(std::memory_order_relaxed);

      if (current == kCanceled) {
        // this can happen because there could be an async barrier between the
        // subscriber and the subscription for instance while onComplete is
        // being delivered (on effectively cancelled subscription) the
        // subscriber can call request(n)
        return;
{'fraction_non_alphanumeric': 0.1109149749926449, 'fraction_numerical': 0.008531921153280376, 'mean_word_length': 3.871060171919771, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 35, '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': '5317257', 'n_tokens_mistral': 1023, 'n_tokens_neox': 993, 'n_words': 295}