Commit 3635f1c8 authored by Dominik Charousset's avatar Dominik Charousset

Fix error propagation in fused operators

parent e6f5f968
...@@ -9,6 +9,8 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -9,6 +9,8 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- The new classes `json_value`, `json_array` and `json_object` allow working - The new classes `json_value`, `json_array` and `json_object` allow working
with JSON inputs directly. Actors can also pass around JSON values safely. with JSON inputs directly. Actors can also pass around JSON values safely.
- Fused stages now properly forward errors during the initial subscription to
their observer.
### Fixed ### Fixed
......
...@@ -47,9 +47,10 @@ public: ...@@ -47,9 +47,10 @@ public:
} }
void on_error(const error& what) { void on_error(const error& what) {
CAF_ASSERT(sub->in_.valid()); if (sub->in_) {
sub->in_.dispose(); sub->in_.dispose();
sub->in_ = nullptr; sub->in_ = nullptr;
}
sub->err_ = what; sub->err_ = what;
} }
}; };
...@@ -137,6 +138,19 @@ public: ...@@ -137,6 +138,19 @@ public:
void on_error(const error& what) override { void on_error(const error& what) override {
if (in_) { if (in_) {
if (!err_) {
auto fn = [this, &what](auto& step, auto&... steps) {
term_step term{this};
step.on_error(what, steps..., term);
};
std::apply(fn, steps_);
if (!running_) {
running_ = true;
do_run();
}
}
} else if (out_) {
// This may only happen if subscribing to the input fails.
auto fn = [this, &what](auto& step, auto&... steps) { auto fn = [this, &what](auto& step, auto&... steps) {
term_step term{this}; term_step term{this};
step.on_error(what, steps..., term); step.on_error(what, steps..., term);
...@@ -273,19 +287,10 @@ public: ...@@ -273,19 +287,10 @@ public:
auto ptr = make_counted<sub_t>(super::ctx_, out, steps_); auto ptr = make_counted<sub_t>(super::ctx_, out, steps_);
input_->subscribe(observer<input_type>{ptr}); input_->subscribe(observer<input_type>{ptr});
if (ptr->subscribed()) { if (ptr->subscribed()) {
auto sub = subscription{std::move(ptr)}; out.on_subscribe(subscription{ptr});
out.on_subscribe(sub); return ptr->as_disposable();
return std::move(sub).as_disposable();
} else if (auto& fail_reason = ptr->fail_reason()) {
out.on_error(fail_reason);
return disposable{};
} else {
auto err = make_error(sec::invalid_observable,
"flow operator from_steps failed "
"to subscribe to its input");
out.on_error(err);
return disposable{};
} }
return disposable{};
} }
private: private:
......
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