Commit 55801c50 authored by Dominik Charousset's avatar Dominik Charousset

Implement never and error flow operators

parent f2c9d7e3
......@@ -288,10 +288,13 @@ caf_add_component(
flow.broadcaster
flow.concat
flow.concat_map
flow.empty
flow.error
flow.flat_map
flow.for_each
flow.interval
flow.merge
flow.never
flow.observe_on
flow.prefix_and_tail
flow.single
......
......@@ -2137,6 +2137,64 @@ private:
coordinator* ctx_;
};
/// An observable that never calls any callbacks on its subscribers.
template <class T>
class mute_observable_impl : public ref_counted, public observable_impl<T> {
public:
// -- member types -----------------------------------------------------------
using output_type = T;
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(mute_observable_impl)
// -- constructors, destructors, and assignment operators --------------------
explicit mute_observable_impl(coordinator* ctx) : ctx_(ctx) {
// nop
}
// -- implementation of disposable::impl -------------------------------------
void dispose() override {
// nop
}
bool disposed() const noexcept override {
return true;
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override {
return ctx_;
}
void on_request(observer_impl<output_type>*, size_t) override {
// nop
}
void on_cancel(observer_impl<output_type>*) override {
// nop
}
disposable subscribe(observer<output_type> sink) override {
return this->do_subscribe(sink.ptr());
}
private:
coordinator* ctx_;
};
/// An observable with minimal internal logic. Useful for writing unit tests.
template <class T>
class passive_observable : public ref_counted, public observable_impl<T> {
......
......@@ -170,6 +170,19 @@ public:
return observable<T>{make_counted<empty_observable_impl<T>>(ctx_)};
}
/// Creates an @ref observable without any values that also never terminates.
template <class T>
[[nodiscard]] observable<T> never() {
return observable<T>{make_counted<mute_observable_impl<T>>(ctx_)};
}
/// Creates an @ref observable without any values that also never terminates.
template <class T>
[[nodiscard]] observable<T> error(caf::error what) {
auto ptr = make_counted<observable_error_impl<T>>(ctx_, std::move(what));
return observable<T>{std::move(ptr)};
}
private:
explicit observable_builder(coordinator* ctx) : ctx_(ctx) {
// nop
......
// 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.empty
#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("an empty observable terminates normally") {
GIVEN("an empty<int32>") {
WHEN("an observer subscribes") {
THEN("the observer receives on_complete") {
auto uut = ctx->make_observable().empty<int32_t>();
auto snk = flow::make_passive_observer<int32_t>();
uut.subscribe(snk->as_observer());
ctx->run();
if (CHECK(snk->sub)) {
snk->sub.request(42);
ctx->run();
CHECK_EQ(snk->state, flow::observer_state::completed);
CHECK(snk->buf.empty());
}
}
}
}
}
END_FIXTURE_SCOPE()
// 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.error
#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("an error observable immediately calls on_error on any subscriber") {
GIVEN("an error<int32>") {
WHEN("an observer subscribes") {
THEN("the observer receives on_error") {
auto uut = ctx->make_observable().error<int32_t>(sec::runtime_error);
auto snk = flow::make_passive_observer<int32_t>();
uut.subscribe(snk->as_observer());
ctx->run();
CHECK(!snk->sub);
CHECK_EQ(snk->state, flow::observer_state::aborted);
CHECK(snk->buf.empty());
}
}
}
}
END_FIXTURE_SCOPE()
// 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.never
#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("a mute observable never invokes any callbacks") {
GIVEN("an never<int32>") {
WHEN("an observer subscribes") {
THEN("the observer never observes any activity") {
auto uut = ctx->make_observable().never<int32_t>();
auto snk = flow::make_passive_observer<int32_t>();
uut.subscribe(snk->as_observer());
ctx->run();
if (CHECK(snk->sub)) {
snk->sub.request(42);
ctx->run();
CHECK_EQ(snk->state, flow::observer_state::subscribed);
CHECK(snk->buf.empty());
}
}
}
}
}
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