Commit c62ce9ee authored by Dominik Charousset's avatar Dominik Charousset

Add stream_source::generate function

This function allows users to trigger creation of new downstream
messages, e.g., after filling a buffer that the source reads.
parent 7178e464
...@@ -47,6 +47,9 @@ public: ...@@ -47,6 +47,9 @@ public:
return *out_ptr_; return *out_ptr_;
} }
/// Convenience function to trigger generation of new elements.
void generate();
protected: protected:
/// Queries the current amount of elements in the output buffer. /// Queries the current amount of elements in the output buffer.
virtual long buf_size() const = 0; virtual long buf_size() const = 0;
......
...@@ -66,4 +66,13 @@ void stream_source::abort(strong_actor_ptr& cause, const error& reason) { ...@@ -66,4 +66,13 @@ void stream_source::abort(strong_actor_ptr& cause, const error& reason) {
out_ptr_->abort(cause, reason); out_ptr_->abort(cause, reason);
} }
void stream_source::generate() {
if (!at_end()) {
auto current_size = buf_size();
auto size_hint = out().total_net_credit();
if (current_size < size_hint)
generate(size_hint - current_size);
}
}
} // namespace caf } // namespace caf
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