Commit 2db16d67 authored by Dominik Charousset's avatar Dominik Charousset

Implement defer and iota operators

parent bf5e8554
...@@ -288,6 +288,7 @@ caf_add_component( ...@@ -288,6 +288,7 @@ caf_add_component(
flow.broadcaster flow.broadcaster
flow.concat flow.concat
flow.concat_map flow.concat_map
flow.defer
flow.empty flow.empty
flow.error flow.error
flow.flat_map flow.flat_map
......
...@@ -337,13 +337,22 @@ protected: ...@@ -337,13 +337,22 @@ protected:
/// Convenience function for creating a sub-type of @ref observable_impl without /// Convenience function for creating a sub-type of @ref observable_impl without
/// exposing the derived type. /// exposing the derived type.
template <class T, class... Ts> template <class Impl, class... Ts>
intrusive_ptr<observable_impl<typename T::output_type>> intrusive_ptr<observable_impl<typename Impl::output_type>>
make_observable_impl(coordinator* ctx, Ts&&... xs) { make_observable_impl(coordinator* ctx, Ts&&... xs) {
using out_t = typename T::output_type; using out_t = typename Impl::output_type;
using res_t = intrusive_ptr<observable_impl<out_t>>; using res_t = intrusive_ptr<observable_impl<out_t>>;
auto ptr = new T(ctx, std::forward<Ts>(xs)...); observable_impl<out_t>* ptr = new Impl(ctx, std::forward<Ts>(xs)...);
return res_t{static_cast<observable_impl<out_t>*>(ptr), false}; return res_t{ptr, false};
}
/// Convenience function for creating an @ref observable from a concrete
/// implementation type.
template <class Impl, class... Ts>
observable<typename Impl::output_type>
make_observable(coordinator* ctx, Ts&&... xs) {
using res_t = observable<typename Impl::output_type>;
return res_t{make_observable_impl<Impl>(ctx, std::forward<Ts>(xs)...)};
} }
/// Base type for classes that represent a definition of an `observable` which /// Base type for classes that represent a definition of an `observable` which
...@@ -565,10 +574,6 @@ protected: ...@@ -565,10 +574,6 @@ protected:
template <class T> template <class T>
class observable_error_impl : public ref_counted, public observable_impl<T> { class observable_error_impl : public ref_counted, public observable_impl<T> {
public: public:
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(observable_error_impl)
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
observable_error_impl(coordinator* ctx, error what) observable_error_impl(coordinator* ctx, error what)
...@@ -1258,8 +1263,8 @@ public: ...@@ -1258,8 +1263,8 @@ public:
if (sub_) { if (sub_) {
sub_ = nullptr; sub_ = nullptr;
merger_->shutdown_on_last_complete(true); merger_->shutdown_on_last_complete(true);
auto obs = make_counted<observable_error_impl<inner_type>>(ctx_, what); auto obs = make_observable<observable_error_impl<inner_type>>(ctx_, what);
merger_->add(obs->as_observable()); merger_->add(std::move(obs));
merger_ = nullptr; merger_ = nullptr;
} }
} }
...@@ -1556,8 +1561,8 @@ public: ...@@ -1556,8 +1561,8 @@ public:
if (sub_) { if (sub_) {
sub_ = nullptr; sub_ = nullptr;
concat_->shutdown_on_last_complete(true); concat_->shutdown_on_last_complete(true);
auto obs = make_counted<observable_error_impl<inner_type>>(ctx_, what); auto obs = make_observable<observable_error_impl<inner_type>>(ctx_, what);
concat_->add(obs->as_observable()); concat_->add(std::move(obs));
concat_ = nullptr; concat_ = nullptr;
} }
} }
...@@ -2041,10 +2046,6 @@ public: ...@@ -2041,10 +2046,6 @@ public:
using output_type = T; using output_type = T;
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(empty_observable_impl)
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit empty_observable_impl(coordinator* ctx) : super(ctx) { explicit empty_observable_impl(coordinator* ctx) : super(ctx) {
...@@ -2089,10 +2090,6 @@ public: ...@@ -2089,10 +2090,6 @@ public:
using output_type = T; using output_type = T;
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(mute_observable_impl)
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit mute_observable_impl(coordinator* ctx) : super(ctx) { explicit mute_observable_impl(coordinator* ctx) : super(ctx) {
...@@ -2260,4 +2257,53 @@ intrusive_ptr<passive_observable<T>> make_passive_observable(coordinator* ctx) { ...@@ -2260,4 +2257,53 @@ intrusive_ptr<passive_observable<T>> make_passive_observable(coordinator* ctx) {
return make_counted<passive_observable<T>>(ctx); return make_counted<passive_observable<T>>(ctx);
} }
/// Implementation of the `defer` operator.
template <class T, class Factory>
class defer_observable_impl : public ref_counted, public observable_impl<T> {
public:
// -- constructors, destructors, and assignment operators --------------------
defer_observable_impl(coordinator* ctx, Factory factory)
: ctx_(ctx), factory_(std::move(factory)) {
// nop
}
// -- implementation of disposable_impl --------------------------------------
void ref_disposable() const noexcept final {
this->ref();
}
void deref_disposable() const noexcept final {
this->deref();
}
void dispose() override {
factory_ = std::nullopt;
}
bool disposed() const noexcept override {
return factory_.has_value();
}
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept final {
return ctx_;
}
disposable subscribe(observer<T> what) override {
if (factory_) {
return (*factory_)().subscribe(what);
} else {
what.on_error(make_error(sec::invalid_observable));
return disposable{};
}
}
private:
coordinator* ctx_;
std::optional<Factory> factory_;
};
} // namespace caf::flow } // namespace caf::flow
...@@ -108,6 +108,12 @@ public: ...@@ -108,6 +108,12 @@ public:
template <class F> template <class F>
[[nodiscard]] generation<callable_source<F>> from_callable(F fn) const; [[nodiscard]] generation<callable_source<F>> from_callable(F fn) const;
/// Creates a @ref generation that emits ascending values.
template <class T>
[[nodiscard]] auto iota(T init) const {
return from_callable([x = std::move(init)]() mutable { return x++; });
}
/// Creates a @ref generation that emits values by repeatedly calling /// Creates a @ref generation that emits values by repeatedly calling
/// `pullable.pull(...)`. For example implementations of the `Pullable` /// `pullable.pull(...)`. For example implementations of the `Pullable`
/// concept, see @ref container_source, @ref repeater_source and /// concept, see @ref container_source, @ref repeater_source and
...@@ -157,21 +163,30 @@ public: ...@@ -157,21 +163,30 @@ public:
/// after subscribing to it. /// after subscribing to it.
template <class T> template <class T>
[[nodiscard]] observable<T> empty() { [[nodiscard]] observable<T> empty() {
return observable<T>{make_counted<empty_observable_impl<T>>(ctx_)}; return make_observable<empty_observable_impl<T>>(ctx_);
} }
/// Creates an @ref observable without any values that also never terminates. /// Creates an @ref observable without any values that also never terminates.
template <class T> template <class T>
[[nodiscard]] observable<T> never() { [[nodiscard]] observable<T> never() {
return observable<T>{make_counted<mute_observable_impl<T>>(ctx_)}; return make_observable<mute_observable_impl<T>>(ctx_);
} }
/// Creates an @ref observable without any values that calls `on_error` after /// Creates an @ref observable without any values that calls `on_error` after
/// subscribing to it. /// subscribing to it.
template <class T> template <class T>
[[nodiscard]] observable<T> error(caf::error what) { [[nodiscard]] observable<T> error(caf::error what) {
auto ptr = make_counted<observable_error_impl<T>>(ctx_, std::move(what)); return make_observable<observable_error_impl<T>>(ctx_, std::move(what));
return observable<T>{std::move(ptr)}; }
/// Create a fresh @ref observable for each @ref observer using the factory.
template <class Factory>
[[nodiscard]] auto defer(Factory factory) {
using res_t = decltype(factory());
static_assert(is_observable_v<res_t>);
using out_t = typename res_t::output_type;
using impl_t = defer_observable_impl<out_t, Factory>;
return make_observable<impl_t>(ctx_, std::move(factory));
} }
private: private:
...@@ -485,7 +500,7 @@ observable_builder::from_resource(async::consumer_resource<T> hdl) const { ...@@ -485,7 +500,7 @@ observable_builder::from_resource(async::consumer_resource<T> hdl) const {
} else { } else {
auto err = make_error(sec::invalid_observable, auto err = make_error(sec::invalid_observable,
"from_resource: failed to open the resource"); "from_resource: failed to open the resource");
return res_t{make_counted<observable_error_impl<T>>(ctx_, std::move(err))}; return make_observable<observable_error_impl<T>>(ctx_, std::move(err));
} }
} }
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.defer
#include "caf/flow/observable_builder.hpp"
#include "core-test.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
namespace {
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("the defer operator produces a fresh observable for each observer") {
GIVEN("a deferred observable") {
WHEN("two observers subscribes") {
THEN("each observer subscribes to a fresh observable") {
using ivec = std::vector<int>;
size_t factory_calls = 0;
auto factory = [this, &factory_calls] {
++factory_calls;
return ctx->make_observable().iota(1).take(5);
};
auto uut = ctx->make_observable().defer(factory);
auto snk1 = flow::make_passive_observer<int>();
auto snk2 = flow::make_passive_observer<int>();
uut.subscribe(snk1->as_observer());
CHECK_EQ(factory_calls, 1u);
REQUIRE(snk1->sub);
uut.subscribe(snk2->as_observer());
CHECK_EQ(factory_calls, 2u);
REQUIRE(snk2->sub);
snk1->sub.request(27);
snk2->sub.request(3);
ctx->run();
CHECK_EQ(snk1->state, flow::observer_state::completed);
CHECK_EQ(snk1->buf, ivec({1, 2, 3, 4, 5}));
CHECK_EQ(snk2->state, flow::observer_state::subscribed);
CHECK_EQ(snk2->buf, ivec({1, 2, 3}));
snk2->sub.request(2);
ctx->run();
CHECK_EQ(snk2->state, flow::observer_state::completed);
CHECK_EQ(snk2->buf, ivec({1, 2, 3, 4, 5}));
}
}
}
}
END_FIXTURE_SCOPE()
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