<commit_before>// Copyright 2017 <NAME>. All Rights Reserved.
//
// 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 <memory>

#include <rs/element_count.h>
#include <rs/publisher.h>
#include <rs/subscriber.h>
#include <rs/subscription.h>

namespace shk {
namespace detail {

template <typename InnerSubscriberType, typename Callback>
class CatchSubscription : public SubscriberBase, public SubscriptionBase {
 public:
  CatchSubscription(
      InnerSubscriberType &&inner_subscriber, const Callback &callback)
      : inner_subscriber_(std::move(inner_subscriber)),
        callback_(callback) {}

  template <typename Publisher>
  void Subscribe(
      std::shared_ptr<CatchSubscription> me,
      Publisher &&publisher) {
    me_ = me;
    auto sub = Subscription(publisher.Subscribe(MakeSubscriber(me)));
    if (!has_failed_) {
    	// It is possible that Subscribe causes OnError to be called before it
    	// even returns. In that case, inner_subscription_ will have been set to
    	// the catch subscription before Subscribe returns, and then we must not
    	// overwrite inner_subscription_
    	inner_subscription_ = std::move(sub);
    }
  }

  template <typename T, class = IsRvalue<T>>
  void OnNext(T &&t) {
  	--requested_;
  	if (requested_ < 0) {
  		cancelled_ = true;
  		inner_subscriber_.OnError(std::make_exception_ptr(
  				std::logic_error("Got value that was not Request-ed")));
  	} else {
  		inner_subscriber_.OnNext(std::forward<T>(t));
  	}
  }

  void OnError(std::exception_ptr &&error) {
  	if (cancelled_) {
  		// Do nothing
  	} else if (has_failed_) {
  		inner_subscriber_.OnError(std::move(error));
  	} else {
  		has_failed_ = true;
  		auto catch_publisher = callback_(std::move(error));
  		static_assert(
  				IsPublisher<decltype(catch_publisher)>,
  				"Catch callback must return a Publisher");
  		inner_subscription_ = Subscription(catch_publisher.Subscribe(
  				MakeSubscriber(me_.lock())));
  		printf("ld: %lld\n", requested_.Get());
  		inner_subscription_.Request(requested_);
  	}
  }

  void OnComplete() {
  	if (!cancelled_) {
  		inner_subscriber_.OnComplete();
  	}
  }

  void Request(ElementCount count) {
  	requested_ += count;<commit_msg>Remove stray debug output from Catch test<commit_after>// Copyright 2017 <NAME>. All Rights Reserved.
//
// 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 <memory>

#include <rs/element_count.h>
#include <rs/publisher.h>
#include <rs/subscriber.h>
#include <rs/subscription.h>

namespace shk {
namespace detail {

template <typename InnerSubscriberType, typename Callback>
class CatchSubscription : public SubscriberBase, public SubscriptionBase {
 public:
  CatchSubscription(
      InnerSubscriberType &&inner_subscriber, const Callback &callback)
      : inner_subscriber_(std::move(inner_subscriber)),
        callback_(callback) {}

  template <typename Publisher>
  void Subscribe(
      std::shared_ptr<CatchSubscription> me,
      Publisher &&publisher) {
    me_ = me;
    auto sub = Subscription(publisher.Subscribe(MakeSubscriber(me)));
    if (!has_failed_) {
    	// It is possible that Subscribe causes OnError to be called before it
    	// even returns. In that case, inner_subscription_ will have been set to
    	// the catch subscription before Subscribe returns, and then we must not
    	// overwrite inner_subscription_
    	inner_subscription_ = std::move(sub);
    }
  }

  template <typename T, class = IsRvalue<T>>
  void OnNext(T &&t) {
  	--requested_;
  	if (requested_ < 0) {
  		cancelled_ = true;
  		inner_subscriber_.OnError(std::make_exception_ptr(
  				std::logic_error("Got value that was not Request-ed")));
  	} else {
  		inner_subscriber_.OnNext(std::forward<T>(t));
  	}
  }

  void OnError(std::exception_ptr &&error) {
  	if (cancelled_) {
  		// Do nothing
  	} else if (has_failed_) {
  		inner_subscriber_.OnError(std::move(error));
  	} else {
  		has_failed_ = true;
  		auto catch_publisher = callback_(std::move(error));
  		static_assert(
  				IsPublisher<decltype(catch_publisher)>,
  				"Catch callback must return a Publisher");
  		inner_subscription_ = Subscription(catch_publisher.Subscribe(
  				MakeSubscriber(me_.lock())));
  		inner_subscription_.Request(requested_);
  	}
  }

  void OnComplete() {
  	if (!cancelled_) {
  		inner_subscriber_.OnComplete();
  	}
  }

  void Request(ElementCount count) {
  	requested_ += count;
{'fraction_non_alphanumeric': 0.10692054192603442, 'fraction_numerical': 0.0032954961552544857, 'mean_word_length': 3.6852487135506005, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7189146', 'n_tokens_mistral': 1731, 'n_tokens_neox': 1688, 'n_words': 482}