Commit 518aae52 authored by Dominik Charousset's avatar Dominik Charousset

Remove test utilities from public headers

parent 268e7806
...@@ -483,7 +483,7 @@ disposable observable<T>::subscribe(async::producer_resource<T> resource) { ...@@ -483,7 +483,7 @@ disposable observable<T>::subscribe(async::producer_resource<T> resource) {
template <class T> template <class T>
disposable observable<T>::subscribe(ignore_t) { disposable observable<T>::subscribe(ignore_t) {
return subscribe(observer<T>::ignore()); return for_each([](const T&) {});
} }
template <class T> template <class T>
......
...@@ -108,12 +108,6 @@ public: ...@@ -108,12 +108,6 @@ public:
return pimpl_.compare(other.pimpl_); return pimpl_.compare(other.pimpl_);
} }
/// Returns an observer that ignores any of its inputs.
static observer ignore();
/// Returns an observer that disposes its subscription immediately.
static observer cancel();
private: private:
intrusive_ptr<impl> pimpl_; intrusive_ptr<impl> pimpl_;
}; };
...@@ -143,55 +137,6 @@ public: ...@@ -143,55 +137,6 @@ public:
namespace caf::detail { namespace caf::detail {
template <class T>
class ignoring_observer : public flow::observer_impl_base<T> {
public:
void on_next(const T&) override {
if (sub_)
sub_.request(1);
}
void on_error(const error&) override {
sub_ = nullptr;
}
void on_complete() override {
sub_ = nullptr;
}
void on_subscribe(flow::subscription sub) override {
if (!sub_) {
sub_ = std::move(sub);
sub_.request(defaults::flow::buffer_size);
} else {
sub.dispose();
}
}
private:
flow::subscription sub_;
};
template <class T>
class canceling_observer : public flow::observer_impl_base<T> {
public:
void on_next(const T&) override {
// nop
}
void on_error(const error&) override {
// nop
}
void on_complete() override {
// nop
}
void on_subscribe(flow::subscription sub) override {
sub.dispose();
}
};
template <class OnNextSignature> template <class OnNextSignature>
struct on_next_trait; struct on_next_trait;
...@@ -279,16 +224,6 @@ private: ...@@ -279,16 +224,6 @@ private:
namespace caf::flow { namespace caf::flow {
template <class T>
observer<T> observer<T>::ignore() {
return observer<T>{make_counted<detail::ignoring_observer<T>>()};
}
template <class T>
observer<T> observer<T>::cancel() {
return observer<T>{make_counted<detail::canceling_observer<T>>()};
}
/// Creates an observer from given callbacks. /// Creates an observer from given callbacks.
/// @param on_next Callback for handling incoming elements. /// @param on_next Callback for handling incoming elements.
/// @param on_error Callback for handling an error. /// @param on_error Callback for handling an error.
......
...@@ -310,8 +310,8 @@ SCENARIO("asynchronous buffers can generate flow items") { ...@@ -310,8 +310,8 @@ SCENARIO("asynchronous buffers can generate flow items") {
THEN("the different pointer types manipulate the same ref count") { THEN("the different pointer types manipulate the same ref count") {
using buf_t = async::spsc_buffer<int>; using buf_t = async::spsc_buffer<int>;
using impl_t = flow::op::from_resource_sub<buf_t>; using impl_t = flow::op::from_resource_sub<buf_t>;
auto ptr = make_counted<impl_t>(ctx.get(), nullptr, auto obs = flow::make_auto_observer<int>();
flow::observer<int>::ignore()); auto ptr = make_counted<impl_t>(ctx.get(), nullptr, obs->as_observer());
CHECK_EQ(ptr->get_reference_count(), 1u); CHECK_EQ(ptr->get_reference_count(), 1u);
{ {
auto sub = flow::subscription{ptr.get()}; auto sub = flow::subscription{ptr.get()};
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment