<commit_before>/*
 * Copyright 2018-present Facebook, Inc.
 *
 * 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.
 */

#pragma once

#include <thrift/lib/cpp2/async/Stream.h>
#include <yarpl/Flowable.h>

namespace apache {
namespace thrift {
namespace detail {

template <typename T>
class EagerSubscribeOnOperator
    : public yarpl::flowable::FlowableOperator<T, T> {
  using Super = yarpl::flowable::FlowableOperator<T, T>;

 public:
  EagerSubscribeOnOperator(
      std::shared_ptr<yarpl::flowable::Flowable<T>> upstream,
      folly::SequencedExecutor& executor)
      : subscription_(std::make_shared<Subscription>(executor)) {
    executor.add(
        [upstream = std::move(upstream), subscription = subscription_] {
          upstream->subscribe(subscription);
        });
  }

  void subscribe(
      std::shared_ptr<yarpl::flowable::Subscriber<T>> subscriber) override {
    auto subscription = std::move(subscription_);
    subscription->setSubscriber(std::move(subscriber));
  }

  ~EagerSubscribeOnOperator() {
    if (auto subscription = std::move(subscription_)) {
      class CancellingSubscriber : public yarpl::flowable::Subscriber<T> {
       public:
        void onSubscribe(std::shared_ptr<yarpl::flowable::Subscription>
                             subscription) override {
          subscription->cancel();
        }
        void onNext(T) override {}
        void onComplete() override {}
        void onError(folly::exception_wrapper) override {}
      };
      subscription->setSubscriber(std::make_shared<CancellingSubscriber>());
    }
  }

 private:
  using SuperSubscription = typename Super::Subscription;
  class Subscription : public SuperSubscription {
   public:
    explicit Subscription(folly::SequencedExecutor& executor)
        : executor_(&executor) {}

    void request(int64_t delta) override {
      auto rExecutor = executor_.rlock();
      if (auto executorPtr = *rExecutor) {
        executorPtr->add([delta, this, self = this->ref_from_this(this)] {
          this->callSuperRequest(delta);
        });
      }
    }

    void cancel() override {
      auto rExecutor = executor_.rlock();
      if (auto executorPtr = *rExecutor) {
        executorPtr->add([this, self = this->ref_from_this(this)] {
          this->callSuperCancel();
        });
      }
    }

    void onNextImpl(T value) override {
      DCHECK(state_ == State::HAS_SUBSCRIBER);
      SuperSubscription::subscriberOnNext(std::move(value));
    }

    void onCompleteImpl() override {
      *executor_.wlock() = nullptr;
      auto state = state_.load();
      switch (state) {
        case State::SUBSCRIBED:
          if (state_.compare_exchange_strong(
                  state,
                  State::COMPLETED,
                  std::memory_order_release,
                  std::memory_order_acquire)) {
            return;
          }
          DCHECK(state == State::HAS_SUBSCRIBER);
          FOLLY_FALLTHROUGH;<commit_msg>Relax racy DCHECK<commit_after>/*
 * Copyright 2018-present Facebook, Inc.
 *
 * 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.
 */

#pragma once

#include <folly/Portability.h>

#include <thrift/lib/cpp2/async/Stream.h>
#include <yarpl/Flowable.h>

namespace apache {
namespace thrift {
namespace detail {

template <typename T>
class EagerSubscribeOnOperator
    : public yarpl::flowable::FlowableOperator<T, T> {
  using Super = yarpl::flowable::FlowableOperator<T, T>;

 public:
  EagerSubscribeOnOperator(
      std::shared_ptr<yarpl::flowable::Flowable<T>> upstream,
      folly::SequencedExecutor& executor)
      : subscription_(std::make_shared<Subscription>(executor)) {
    executor.add(
        [upstream = std::move(upstream), subscription = subscription_] {
          upstream->subscribe(subscription);
        });
  }

  void subscribe(
      std::shared_ptr<yarpl::flowable::Subscriber<T>> subscriber) override {
    auto subscription = std::move(subscription_);
    subscription->setSubscriber(std::move(subscriber));
  }

  ~EagerSubscribeOnOperator() {
    if (auto subscription = std::move(subscription_)) {
      class CancellingSubscriber : public yarpl::flowable::Subscriber<T> {
       public:
        void onSubscribe(std::shared_ptr<yarpl::flowable::Subscription>
                             subscription) override {
          subscription->cancel();
        }
        void onNext(T) override {}
        void onComplete() override {}
        void onError(folly::exception_wrapper) override {}
      };
      subscription->setSubscriber(std::make_shared<CancellingSubscriber>());
    }
  }

 private:
  using SuperSubscription = typename Super::Subscription;
  class Subscription : public SuperSubscription {
   public:
    explicit Subscription(folly::SequencedExecutor& executor)
        : executor_(&executor) {}

    void request(int64_t delta) override {
      auto rExecutor = executor_.rlock();
      if (auto executorPtr = *rExecutor) {
        executorPtr->add([delta, this, self = this->ref_from_this(this)] {
          this->callSuperRequest(delta);
        });
      }
    }

    void cancel() override {
      auto rExecutor = executor_.rlock();
      if (auto executorPtr = *rExecutor) {
        executorPtr->add([this, self = this->ref_from_this(this)] {
          this->callSuperCancel();
        });
      }
    }

    void onNextImpl(T value) override {
      if (folly::kIsDebug) {
        const auto state = state_.load();
        // request(n) may race with the state transition from SUBSCRIBED to
        // HAS_SUBSCRIBER
        DCHECK(state == State::HAS_SUBSCRIBER || state == State::SUBSCRIBED);
      }
      SuperSubscription::subscriberOnNext(std::move(value));
    }

    void onCompleteImpl() override {
      *executor_.wlock() = nullptr;
      auto state = state_.load();
      switch (state) {
        case State::SUBSCRIBED:
          if (state_.compare_exchange_strong(
                  state,
                  State::COMPLETED,
                  std::memory_order_release,
                  std::memory_order_acquire)) {
            return;
          }
          DCHECK(state == State::HAS_SUBSCRIBER);
          FOLLY_FALLTHROUGH;
{'fraction_non_alphanumeric': 0.11101789709172259, 'fraction_numerical': 0.003076062639821029, 'mean_word_length': 2.9805230940456315, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 50, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '4148102', 'n_tokens_mistral': 2047, 'n_tokens_neox': 2029, 'n_words': 476}