Commit 0731e31d authored by Dominik Charousset's avatar Dominik Charousset

Add `stream_manager::continuous` property

parent 74480828
...@@ -55,7 +55,7 @@ public: ...@@ -55,7 +55,7 @@ public:
} }
bool done() const override { bool done() const override {
return this->inbound_paths_.empty(); return !this->continuous() && this->inbound_paths_.empty();
} }
void handle(inbound_path*, downstream_msg::batch& x) override { void handle(inbound_path*, downstream_msg::batch& x) override {
......
...@@ -68,8 +68,8 @@ public: ...@@ -68,8 +68,8 @@ public:
} }
bool done() const override { bool done() const override {
return this->pending_handshakes_ == 0 && this->inbound_paths_.empty() return !this->continuous() && this->pending_handshakes_ == 0
&& out_.clean(); && this->inbound_paths_.empty() && out_.clean();
} }
void handle(inbound_path*, downstream_msg::batch& x) override { void handle(inbound_path*, downstream_msg::batch& x) override {
......
...@@ -143,6 +143,21 @@ public: ...@@ -143,6 +143,21 @@ public:
/// Calls `x.deliver()` /// Calls `x.deliver()`
void deliver_promises(message x); void deliver_promises(message x);
// -- properties -------------------------------------------------------------
/// Returns whether this stream remains open even if no in- or outbound paths
/// exist. The default is `false`. Does not keep a source alive past the
/// point where its driver returns `done() == true`.
inline bool continuous() const noexcept {
return continuous_;
}
/// Sets whether this stream remains open even if no in- or outbound paths
/// exist.
inline void continuous(bool x) noexcept {
continuous_ = x;
}
// -- inline functions ------------------------------------------------------- // -- inline functions -------------------------------------------------------
inline local_actor* self() { inline local_actor* self() {
...@@ -196,6 +211,10 @@ protected: ...@@ -196,6 +211,10 @@ protected:
/// Stores promises while a handshake is active. The sink at the associated /// Stores promises while a handshake is active. The sink at the associated
/// key becomes responsible for the promise after receiving `ack_open`. /// key becomes responsible for the promise after receiving `ack_open`.
std::map<stream_slot, response_promise> in_flight_promises_; std::map<stream_slot, response_promise> in_flight_promises_;
/// Stores whether this stream shall remain open even if no in- or outbound
/// paths exist.
bool continuous_;
}; };
/// A reference counting pointer to a `stream_manager`. /// A reference counting pointer to a `stream_manager`.
......
...@@ -37,7 +37,8 @@ namespace caf { ...@@ -37,7 +37,8 @@ namespace caf {
stream_manager::stream_manager(local_actor* selfptr, stream_priority prio) stream_manager::stream_manager(local_actor* selfptr, stream_priority prio)
: self_(selfptr), : self_(selfptr),
pending_handshakes_(0), pending_handshakes_(0),
priority_(prio) { priority_(prio),
continuous_(false) {
// nop // nop
} }
......
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