Commit c2758a4b authored by Dominik Charousset's avatar Dominik Charousset

Fix typos and inconsistencies

parent 04e02412
......@@ -70,7 +70,7 @@ public:
// nop
}
void cancel() {
void dispose() {
if (buf_) {
buf_->cancel();
buf_ = nullptr;
......
......@@ -66,7 +66,7 @@ public:
}
}
bool cancelled() const {
bool canceled() const {
std::unique_lock<std::mutex> guard{mtx_};
return demand_ == -1;
}
......@@ -160,9 +160,9 @@ public:
}
}
/// Checks whether the consumer cancelled its subscription.
bool cancelled() const {
return impl_->cancelled();
/// Checks whether the consumer canceled its subscription.
bool canceled() const {
return impl_->canceled();
}
private:
......
......@@ -57,7 +57,6 @@ public:
return *this;
}
// -- factories --------------------------------------------------------------
/// Combines multiple disposables into a single disposable. The new disposable
......
......@@ -195,11 +195,6 @@ public:
return std::move(*this).transform(map_step<F>{std::move(f)});
}
template <class F>
auto flat_map_optional(F f) && {
return std::move(*this).transform(flat_map_optional_step<F>{std::move(f)});
}
template <class F>
auto do_on_next(F f) && {
return std::move(*this) //
......@@ -336,17 +331,17 @@ disposable observable<T>::for_each(OnNext on_next, OnError on_error,
// -- observable::merge --------------------------------------------------------
template <class T>
template <class... Inputs>
template <class Out, class... Inputs>
auto observable<T>::merge(Inputs&&... xs) {
if constexpr (is_observable_v<output_type>) {
using value_t = output_type_t<output_type>;
if constexpr (is_observable_v<Out>) {
using value_t = output_type_t<Out>;
using impl_t = op::merge<value_t>;
return make_observable<impl_t>(ctx(), *this, std::forward<Inputs>(xs)...);
} else {
static_assert(
sizeof...(Inputs) > 0,
"merge without arguments expects this observable to emit observables");
using impl_t = op::merge<output_type>;
using impl_t = op::merge<Out>;
return make_observable<impl_t>(ctx(), *this, std::forward<Inputs>(xs)...);
}
}
......@@ -354,18 +349,16 @@ auto observable<T>::merge(Inputs&&... xs) {
// -- observable::flat_map -----------------------------------------------------
template <class T>
template <class F>
template <class Out, class F>
auto observable<T>::flat_map(F f) {
using res_t = decltype(f(std::declval<const output_type&>()));
using res_t = decltype(f(std::declval<const Out&>()));
if constexpr (is_observable_v<res_t>) {
return map([fn = std::move(f)](const output_type& x) mutable {
return map([fn = std::move(f)](const Out& x) mutable {
return fn(x).as_observable();
})
.merge();
} else if constexpr (detail::is_optional_v<res_t>) {
return map([fn = std::move(f)](const output_type& x) mutable {
return fn(x);
})
return map([fn = std::move(f)](const Out& x) mutable { return fn(x); })
.filter([](const res_t& x) { return x.has_value(); })
.map([](const res_t& x) { return *x; });
} else {
......@@ -374,39 +367,27 @@ auto observable<T>::flat_map(F f) {
// all available immediately, there is no good reason to mess up the emitted
// order of values.
static_assert(detail::is_iterable_v<res_t>);
return map([cptr = ctx(), fn = std::move(f)](const output_type& x) mutable {
return map([cptr = ctx(), fn = std::move(f)](const Out& x) mutable {
return cptr->make_observable().from_container(fn(x));
})
.concat();
}
}
// -- observable::flat_map_optional --------------------------------------------
template <class T>
template <class F>
transformation<flat_map_optional_step<F>>
observable<T>::flat_map_optional(F f) {
using step_type = flat_map_optional_step<F>;
static_assert(std::is_same_v<typename step_type::input_type, T>,
"flat_map_optional function does not match the input type");
return {*this, std::forward_as_tuple(step_type{std::move(f)})};
}
// -- observable::concat -------------------------------------------------------
template <class T>
template <class... Inputs>
template <class Out, class... Inputs>
auto observable<T>::concat(Inputs&&... xs) {
if constexpr (is_observable_v<output_type>) {
using value_t = output_type_t<output_type>;
if constexpr (is_observable_v<Out>) {
using value_t = output_type_t<Out>;
using impl_t = op::concat<value_t>;
return make_observable<impl_t>(ctx(), *this, std::forward<Inputs>(xs)...);
} else {
static_assert(
sizeof...(Inputs) > 0,
"merge without arguments expects this observable to emit observables");
using impl_t = op::concat<output_type>;
using impl_t = op::concat<Out>;
return make_observable<impl_t>(ctx(), *this, std::forward<Inputs>(xs)...);
}
}
......@@ -414,23 +395,21 @@ auto observable<T>::concat(Inputs&&... xs) {
// -- observable::concat_map ---------------------------------------------------
template <class T>
template <class F>
template <class Out, class F>
auto observable<T>::concat_map(F f) {
using res_t = decltype(f(std::declval<const output_type&>()));
using res_t = decltype(f(std::declval<const Out&>()));
if constexpr (is_observable_v<res_t>) {
return map([fn = std::move(f)](const output_type& x) mutable {
return map([fn = std::move(f)](const Out& x) mutable {
return fn(x).as_observable();
})
.concat();
} else if constexpr (detail::is_optional_v<res_t>) {
return map([fn = std::move(f)](const output_type& x) mutable {
return fn(x);
})
return map([fn = std::move(f)](const Out& x) mutable { return fn(x); })
.filter([](const res_t& x) { return x.has_value(); })
.map([](const res_t& x) { return *x; });
} else {
static_assert(detail::is_iterable_v<res_t>);
return map([cptr = ctx(), fn = std::move(f)](const output_type& x) mutable {
return map([cptr = ctx(), fn = std::move(f)](const Out& x) mutable {
return cptr->make_observable().from_container(fn(x));
})
.concat();
......
......@@ -137,28 +137,23 @@ public:
/// Combines the output of multiple @ref observable objects into one by
/// merging their outputs. May also be called without arguments if the `T` is
/// an @ref observable.
template <class... Inputs>
template <class Out = output_type, class... Inputs>
auto merge(Inputs&&... xs);
/// Combines the output of multiple @ref observable objects into one by
/// concatenating their outputs. May also be called without arguments if the
/// `T` is an @ref observable.
template <class... Inputs>
template <class Out = output_type, class... Inputs>
auto concat(Inputs&&...);
/// Returns a transformation that emits items by merging the outputs of all
/// observables returned by `f`.
template <class F>
template <class Out = output_type, class F>
auto flat_map(F f);
/// Returns a transformation that emits items from optional values returned by
/// `f`.
template <class F>
transformation<flat_map_optional_step<F>> flat_map_optional(F f);
/// Returns a transformation that emits items by concatenating the outputs of
/// all observables returned by `f`.
template <class F>
template <class Out = output_type, class F>
auto concat_map(F f);
/// Takes @p prefix_size elements from this observable and emits it in a tuple
......
......@@ -211,7 +211,7 @@ public:
sub_ = std::move(sub);
sub_.request(defaults::flow::buffer_size);
} else {
sub.cancel();
sub.dispose();
}
}
......@@ -355,7 +355,7 @@ public:
sub_ = std::move(sub);
} else {
CAF_LOG_DEBUG("already have a subscription or buffer no longer valid");
sub.cancel();
sub.dispose();
}
}
......@@ -399,7 +399,7 @@ private:
void on_cancel() {
CAF_LOG_TRACE("");
if (sub_) {
sub_.cancel();
sub_.dispose();
sub_ = nullptr;
}
buf_ = nullptr;
......@@ -457,7 +457,7 @@ public:
if (parent_) {
parent_->fwd_on_subscribe(token_, std::move(new_sub));
} else {
new_sub.cancel();
new_sub.dispose();
}
}
......@@ -491,7 +491,7 @@ public:
void on_complete() override {
if (sub) {
sub.cancel();
sub.dispose();
sub = nullptr;
}
state = observer_state::completed;
......@@ -499,7 +499,7 @@ public:
void on_error(const error& what) override {
if (sub) {
sub.cancel();
sub.dispose();
sub = nullptr;
}
err = what;
......@@ -512,7 +512,7 @@ public:
sub = std::move(new_sub);
state = observer_state::subscribed;
} else {
new_sub.cancel();
new_sub.dispose();
}
}
......@@ -577,7 +577,7 @@ public:
void on_complete() override {
if (sub) {
sub.cancel();
sub.dispose();
sub = nullptr;
}
state = observer_state::completed;
......@@ -585,7 +585,7 @@ public:
void on_error(const error& what) override {
if (sub) {
sub.cancel();
sub.dispose();
sub = nullptr;
}
err = what;
......@@ -599,7 +599,7 @@ public:
state = observer_state::subscribed;
sub.request(64);
} else {
new_sub.cancel();
new_sub.dispose();
}
}
......
......@@ -97,7 +97,7 @@ public:
return !state_;
}
void cancel() override {
void dispose() override {
if (state_) {
state_->drop(out_);
state_ = nullptr;
......
......@@ -83,7 +83,7 @@ public:
factory_sub_ = std::move(sub);
factory_sub_.request(1);
} else {
sub.cancel();
sub.dispose();
}
}
......@@ -134,7 +134,7 @@ public:
return !out_;
}
void cancel() override {
void dispose() override {
if (out_) {
ctx_->delay_fn([strong_this = intrusive_ptr<concat_sub>{this}] {
if (strong_this->out_) {
......@@ -156,11 +156,11 @@ private:
void fin() {
CAF_ASSERT(out_);
if (factory_sub_) {
factory_sub_.cancel();
factory_sub_.dispose();
factory_sub_ = nullptr;
}
if (active_sub_) {
active_sub_.cancel();
active_sub_.dispose();
active_sub_ = nullptr;
}
factory_key_ = 0;
......
......@@ -29,13 +29,13 @@ public:
return !out_;
}
void cancel() override {
void dispose() override {
if (out_)
ctx_->delay_fn([out = std::move(out_)]() mutable { out.on_complete(); });
}
void request(size_t) override {
cancel();
dispose();
}
private:
......
......@@ -53,7 +53,7 @@ public:
return !out_;
}
void cancel() override {
void dispose() override {
if (out_) {
completed_ = true;
buf_.clear();
......
......@@ -55,7 +55,7 @@ public:
return disposed_;
}
void cancel() override {
void dispose() override {
CAF_LOG_TRACE("");
if (!disposed_) {
disposed_ = true;
......
......@@ -42,13 +42,13 @@ public:
void on_complete() {
CAF_ASSERT(sub->in_.valid());
sub->in_.cancel();
sub->in_.dispose();
sub->in_ = nullptr;
}
void on_error(const error& what) {
CAF_ASSERT(sub->subscribed());
sub->in_.cancel();
sub->in_.dispose();
sub->in_ = nullptr;
sub->err_ = what;
}
......@@ -154,7 +154,7 @@ public:
in_ = std::move(in);
pull();
} else {
in.cancel();
in.dispose();
}
}
......@@ -164,7 +164,7 @@ public:
return disposed_;
}
void cancel() override {
void dispose() override {
CAF_LOG_TRACE("");
if (!disposed_) {
disposed_ = true;
......@@ -173,7 +173,7 @@ public:
ctx_->delay_fn([out = std::move(out_)]() mutable { out.on_complete(); });
}
if (in_) {
in_.cancel();
in_.dispose();
in_ = nullptr;
}
}
......
......@@ -110,7 +110,7 @@ public:
return !state_;
}
void cancel() override {
void dispose() override {
if (state_) {
ctx_->delay_fn([state = std::move(state_)]() { state->do_dispose(); });
}
......
......@@ -77,7 +77,7 @@ public:
sub.request(max_pending_);
ptr->sub = std::move(sub);
} else {
sub.cancel();
sub.dispose();
}
}
......@@ -102,7 +102,7 @@ public:
while (i != inputs_.end()) {
auto& input = *i->second;
if (auto& sub = input.sub) {
sub.cancel();
sub.dispose();
sub = nullptr;
}
if (input.buf.empty())
......@@ -138,11 +138,11 @@ public:
return !out_;
}
void cancel() override {
void dispose() override {
if (out_) {
for (auto& kvp : inputs_)
if (auto& sub = kvp.second->sub)
sub.cancel();
sub.dispose();
inputs_.clear();
run_later();
}
......@@ -215,7 +215,7 @@ private:
if (!inputs_.empty()) {
for (auto& kvp : inputs_)
if (auto& sub = kvp.second->sub)
sub.cancel();
sub.dispose();
inputs_.clear();
}
if (!err_) {
......
......@@ -28,7 +28,7 @@ public:
return !out_;
}
void cancel() override {
void dispose() override {
if (out_)
ctx_->delay_fn([out = std::move(out_)]() mutable { out.on_complete(); });
}
......
......@@ -93,7 +93,7 @@ public:
}
void on_subscribe(subscription in) override {
in.cancel();
in.dispose();
}
// -- implementation of subscription_impl ------------------------------------
......@@ -102,9 +102,9 @@ public:
return !out_;
}
void cancel() override {
void dispose() override {
if (in_) {
in_.cancel();
in_.dispose();
in_ = nullptr;
}
if (out_) {
......@@ -145,7 +145,7 @@ public:
~pipe() {
if (sub_)
sub_->cancel();
sub_->dispose();
}
disposable subscribe(observer<T> out) override {
......
......@@ -21,13 +21,13 @@
namespace caf::flow::op {
template <class F, class... Ts>
struct zip_wich_oracle {
struct zip_with_oracle {
using output_type
= decltype(std::declval<F&>()(std::declval<const Ts&>()...));
};
template <class F, class... Ts>
using zip_with_output_t = typename zip_wich_oracle<F, Ts...>::output_type;
using zip_with_output_t = typename zip_with_oracle<F, Ts...>::output_type;
template <size_t Index>
using zip_index = std::integral_constant<size_t, Index>;
......@@ -73,11 +73,11 @@ public:
return !out_;
}
void cancel() override {
void dispose() override {
if (out_) {
for_each_input([](auto, auto& input) {
if (input.sub) {
input.sub.cancel();
input.sub.dispose();
input.sub = nullptr;
}
input.buf.clear();
......@@ -143,10 +143,10 @@ public:
sub.request(demand_);
in.sub = std::move(sub);
} else {
sub.cancel();
sub.dispose();
}
} else {
sub.cancel();
sub.dispose();
}
}
......@@ -203,7 +203,7 @@ private:
void fin() {
for_each_input([](auto, auto& input) {
if (input.sub) {
input.sub.cancel();
input.sub.dispose();
input.sub = nullptr;
}
input.buf.clear();
......
......@@ -27,14 +27,8 @@ public:
public:
~impl() override;
/// Causes the publisher to stop producing items for the subscriber. Any
/// in-flight items may still get dispatched.
virtual void cancel() = 0;
/// Signals demand for `n` more items.
virtual void request(size_t n) = 0;
void dispose() final;
};
/// Simple base type for all subscription implementations that implements the
......@@ -71,7 +65,7 @@ public:
void request(size_t n) override;
void cancel() override;
void dispose() override;
auto* ctx() const noexcept {
return ctx_;
......@@ -125,10 +119,11 @@ public:
// -- demand signaling -------------------------------------------------------
/// @copydoc impl::cancel
void cancel() {
/// Causes the publisher to stop producing items for the subscriber. Any
/// in-flight items may still get dispatched.
void dispose() {
if (pimpl_) {
pimpl_->cancel();
pimpl_->dispose();
pimpl_ = nullptr;
}
}
......
......@@ -69,7 +69,7 @@ public:
// -- implementation of subscription_impl ------------------------------------
void cancel() override {
void dispose() override {
if (out_) {
ctx_->delay_fn([ptr = strong_this()] { ptr->do_cancel(); });
}
......
......@@ -12,10 +12,6 @@ subscription::impl::~impl() {
// nop
}
void subscription::impl::dispose() {
cancel();
}
void subscription::impl_base::ref_disposable() const noexcept {
this->ref();
}
......@@ -39,7 +35,7 @@ void subscription::fwd_impl::request(size_t n) {
});
}
void subscription::fwd_impl::cancel() {
void subscription::fwd_impl::dispose() {
if (src_) {
ctx()->delay_fn([src = src_, snk = snk_] { //
src->on_cancel(snk.get());
......
......@@ -42,7 +42,7 @@ SCENARIO("repeater sources repeat one value indefinitely") {
snk->sub.request(4);
ctx->run();
CHECK_EQ(snk->buf, ivec({42, 42, 42, 42, 42, 42, 42}));
snk->sub.cancel();
snk->sub.dispose();
ctx->run();
CHECK_EQ(snk->buf, ivec({42, 42, 42, 42, 42, 42, 42}));
}
......@@ -112,7 +112,7 @@ SCENARIO("callable sources stream values generated from a function object") {
snk->sub.request(4);
ctx->run();
CHECK_EQ(snk->buf, ivec({1, 2, 3, 4, 5, 6, 7}));
snk->sub.cancel();
snk->sub.dispose();
ctx->run();
CHECK_EQ(snk->buf, ivec({1, 2, 3, 4, 5, 6, 7}));
}
......
......@@ -23,20 +23,8 @@ struct fixture : test_coordinator_fixture<> {
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("the never operator never invokes callbacks except when disposed") {
GIVEN("a never<int32>") {
WHEN("an observer subscribes") {
THEN("the observer never receives any events") {
auto uut = ctx->make_observable().never<int32_t>();
auto snk = flow::make_auto_observer<int32_t>();
uut.subscribe(snk->as_observer());
ctx->run();
CHECK(snk->buf.empty());
CHECK_EQ(snk->state, flow::observer_state::subscribed);
}
}
}
GIVEN("a never<int32> that gets disposed") {
WHEN("an observer subscribes") {
GIVEN("a never operator") {
WHEN("an observer subscribes and disposing the subscription") {
THEN("the observer receives on_complete") {
auto uut = ctx->make_observable().never<int32_t>();
auto snk1 = flow::make_auto_observer<int32_t>();
......
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