text: // Copyright (c) Facebook, Inc. and its affiliates.
  //
  // 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 "yarpl/observable/ObservableOperator.h"
  
  namespace yarpl {
  namespace observable {
  namespace details {
  
  template <
      typename U,
      typename OnSubscribeFunc,
      typename OnNextFunc,
      typename OnErrorFunc,
      typename OnCompleteFunc,
      typename OnCancelFunc>
  class DoOperator : public ObservableOperator<U, U> {
    using Super = ObservableOperator<U, U>;
    static_assert(
        std::is_same<std::decay_t<OnSubscribeFunc>, OnSubscribeFunc>::value,
        "undecayed");
    static_assert(
        std::is_same<std::decay_t<OnNextFunc>, OnNextFunc>::value,
        "undecayed");
    static_assert(
        std::is_same<std::decay_t<OnErrorFunc>, OnErrorFunc>::value,
        "undecayed");
    static_assert(
        std::is_same<std::decay_t<OnCompleteFunc>, OnCompleteFunc>::value,
        "undecayed");
    static_assert(
        std::is_same<std::decay_t<OnCancelFunc>, OnCancelFunc>::value,
        "undecayed");
  
   public:
    template <
        typename FSubscribe,
        typename FNext,
        typename FError,
        typename FComplete,
        typename FCancel>
    DoOperator(
        std::shared_ptr<Observable<U>> upstream,
        FSubscribe&& onSubscribeFunc,
        FNext&& onNextFunc,
        FError&& onErrorFunc,
        FComplete&& onCompleteFunc,
        FCancel&& onCancelFunc)
        : upstream_(std::move(upstream)),
          onSubscribeFunc_(std::forward<FSubscribe>(onSubscribeFunc)),
          onNextFunc_(std::forward<FNext>(onNextFunc)),
          onErrorFunc_(std::forward<FError>(onErrorFunc)),
          onCompleteFunc_(std::forward<FComplete>(onCompleteFunc)),
          onCancelFunc_(std::forward<FCancel>(onCancelFunc)) {}
  
    std::shared_ptr<Subscription> subscribe(
        std::shared_ptr<Observer<U>> observer) override {
      auto subscription = std::make_shared<DoSubscription>(
          this->ref_from_this(this), std::move(observer));
      upstream_->subscribe(
          // Note: implicit cast to a reference to a observer.
          subscription);
      return subscription;
    }
  
   private:
    class DoSubscription : public Super::OperatorSubscription {
      using SuperSub = typename Super::OperatorSubscription;
  
     public:
      DoSubscription(
          std::shared_ptr<DoOperator> observable,
          std::shared_ptr<Observer<U>> observer)
          : SuperSub(std::move(observer)), observable_(std::move(observable)) {}
  
      void onSubscribe(std::shared_ptr<yarpl::observable::Subscription>
                           subscription) override {
        observable_->onSubscribeFunc_();
        SuperSub::onSubscribe(std::move(subscription));
      }
  
      void onNext(U value) override {
        const auto& valueRef = value;
        observable_->onNextFunc_(valueRef);
        SuperSub::observerOnNext(std::move(value));
      }
  
      void onError(folly::exception_wrapper ex) override {
        const auto& exRef = ex;
        observable_->onErrorFunc_(exRef);
        SuperSub::onError(std::move(ex));
      }
  
      void onComplete() override {
        observable_->onCompleteFunc_();
        SuperSub::onComplete();
      }
  
      void cancel() override {
        observable_->onCancelFunc_();
        SuperSub::cancel();
      }
  
     private:
      std::shared_ptr<DoOperator> observable_;
    };
  
    std::shared_ptr<Observable<U>> upstream_;
    OnSubscribeFunc onSubscribeFunc_;
    OnNextFunc onNextFunc_;
    OnErrorFunc onErrorFunc_;
    OnCompleteFunc onCompleteFunc_;
    OnCancelFunc onCancelFunc_;
  };
  
  template <
      typename U,
      typename OnSubscribeFunc,
      typename OnNextFunc,
      typename OnErrorFunc,
      typename OnCompleteFunc,
      typename OnCancelFunc>
  inline auto createDoOperator(
      std::shared_ptr<Observable<U>> upstream,
      OnSubscribeFunc&& onSubscribeFunc,
      OnNextFunc&& onNextFunc,
      OnErrorFunc&& onErrorFunc,
      OnCompleteFunc&& onCompleteFunc,
      OnCancelFunc&& onCancelFunc) {
    return std::make_shared<DoOperator<
        U,
        std::decay_t<OnSubscribeFunc>,
        std::decay_t<OnNextFunc>,
        std::decay_t<OnErrorFunc>,
        std::decay_t<OnCompleteFunc>,
        std::decay_t<OnCancelFunc>>>(
        std::move(upstream),
        std::forward<OnSubscribeFunc>(onSubscribeFunc),
        std::forward<OnNextFunc>(onNextFunc),
        std::forward<OnErrorFunc>(onErrorFunc),
        std::forward<OnCompleteFunc>(onCompleteFunc),
        std::forward<OnCancelFunc>(onCancelFunc));
  }
  } // namespace details
  } // namespace observable
  } // namespace yarpl
  