Commit 290f7d8e authored by Dominik Charousset's avatar Dominik Charousset

Add finalize with error argument to all drivers

parent 64c6a025
...@@ -54,7 +54,7 @@ public: ...@@ -54,7 +54,7 @@ public:
return trait::process::invoke(process_, state_, std::move(xs)); return trait::process::invoke(process_, state_, std::move(xs));
} }
message finalize() override { message make_final_result() override {
return trait::finalize::invoke(finalize_, state_); return trait::finalize::invoke(finalize_, state_);
} }
......
...@@ -71,7 +71,11 @@ public: ...@@ -71,7 +71,11 @@ public:
protected: protected:
message make_final_result() override { message make_final_result() override {
return driver_.finalize(); return driver_.make_final_result();
}
void finalize(const error& reason) override {
driver_.finalize(reason);
} }
private: private:
......
...@@ -89,6 +89,11 @@ public: ...@@ -89,6 +89,11 @@ public:
return make_message_from_tuple(driver_.make_handshake(slot)); return make_message_from_tuple(driver_.make_handshake(slot));
} }
protected:
void finalize(const error& reason) override {
driver_.finalize(reason);
}
private: private:
bool at_end_; bool at_end_;
Driver driver_; Driver driver_;
......
...@@ -74,7 +74,7 @@ public: ...@@ -74,7 +74,7 @@ public:
trait::process::invoke(process_, state_, std::move(batch), out); trait::process::invoke(process_, state_, std::move(batch), out);
} }
void finalize() override { void finalize(const error&) override {
return fin_(state_); return fin_(state_);
} }
......
...@@ -93,6 +93,11 @@ public: ...@@ -93,6 +93,11 @@ public:
return out_.capacity() == 0; return out_.capacity() == 0;
} }
protected:
void finalize(const error& reason) override {
driver_.finalize(reason);
}
private: private:
driver_type driver_; driver_type driver_;
scatterer_type out_; scatterer_type out_;
......
...@@ -173,6 +173,10 @@ public: ...@@ -173,6 +173,10 @@ public:
} }
protected: protected:
// -- implementation hooks ---------------------------------------------------
virtual void finalize(const error& reason);
// -- implementation hooks for sinks ----------------------------------------- // -- implementation hooks for sinks -----------------------------------------
/// Called when the gatherer closes to produce the final stream result for /// Called when the gatherer closes to produce the final stream result for
......
...@@ -46,10 +46,14 @@ public: ...@@ -46,10 +46,14 @@ public:
// -- virtual functions ------------------------------------------------------ // -- virtual functions ------------------------------------------------------
/// Cleans up any state and produces a result message. /// Cleans up any state and produces a result message.
virtual message finalize() { virtual message make_final_result() {
return make_message(); return make_message();
} }
virtual void finalize(const error&) {
// nop
}
// -- pure virtual functions ------------------------------------------------- // -- pure virtual functions -------------------------------------------------
/// Processes a single batch. /// Processes a single batch.
......
...@@ -53,7 +53,7 @@ public: ...@@ -53,7 +53,7 @@ public:
// -- virtual functions ------------------------------------------------------ // -- virtual functions ------------------------------------------------------
/// Cleans up any state. /// Cleans up any state.
virtual void finalize() { virtual void finalize(const error&) {
// nop // nop
} }
......
...@@ -56,7 +56,7 @@ public: ...@@ -56,7 +56,7 @@ public:
// -- virtual functions ------------------------------------------------------ // -- virtual functions ------------------------------------------------------
/// Cleans up any state. /// Cleans up any state.
virtual void finalize() { virtual void finalize(const error&) {
// nop // nop
} }
......
...@@ -91,6 +91,8 @@ void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) { ...@@ -91,6 +91,8 @@ void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) {
void stream_manager::stop() { void stream_manager::stop() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
out().close(); out().close();
error tmp;
finalize(tmp);
self_->erase_inbound_paths_later(this); self_->erase_inbound_paths_later(this);
if (!promises_.empty()) if (!promises_.empty())
deliver_promises(make_final_result()); deliver_promises(make_final_result());
...@@ -106,6 +108,7 @@ void stream_manager::abort(error reason) { ...@@ -106,6 +108,7 @@ void stream_manager::abort(error reason) {
in_flight_promises_.clear(); in_flight_promises_.clear();
} }
out().abort(reason); out().abort(reason);
finalize(reason);
self_->erase_inbound_paths_later(this, std::move(reason)); self_->erase_inbound_paths_later(this, std::move(reason));
} }
...@@ -182,6 +185,10 @@ void stream_manager::deliver_promises(message x) { ...@@ -182,6 +185,10 @@ void stream_manager::deliver_promises(message x) {
promises_.clear(); promises_.clear();
} }
void stream_manager::finalize(const error&) {
// nop
}
message stream_manager::make_final_result() { message stream_manager::make_final_result() {
return none; return none;
} }
......
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