Commit 899a8449 authored by Dominik Charousset's avatar Dominik Charousset

Allow streams to remain without upstream path

parent a3b43745
...@@ -71,9 +71,22 @@ public: ...@@ -71,9 +71,22 @@ public:
return self_; return self_;
} }
/// Queries whether all upstream paths were closed. /// Returns `true` if all upstream paths are closed and this upstream is not
/// flagged as `continuous`, `false` otherwise.
inline bool closed() const { inline bool closed() const {
return paths_.empty(); return paths_.empty() && !continuous_;
}
/// Returns whether this upstream remains open even if no more upstream path
/// exists.
inline bool continuous() const {
return continuous_;
}
/// Sets whether this upstream remains open even if no more upstream path
/// exists.
inline void continuous(bool value) {
continuous_ = value;
} }
optional<path&> find(const strong_actor_ptr& x) const; optional<path&> find(const strong_actor_ptr& x) const;
...@@ -90,6 +103,10 @@ protected: ...@@ -90,6 +103,10 @@ protected:
/// An assignment vector that's re-used whenever calling the policy. /// An assignment vector that's re-used whenever calling the policy.
upstream_policy::assignment_vec policy_vec_; upstream_policy::assignment_vec policy_vec_;
/// Stores whether this stream remains open even if all paths have been
/// closed.
bool continuous_;
}; };
} // namespace caf } // namespace caf
......
...@@ -33,7 +33,7 @@ namespace mixin { ...@@ -33,7 +33,7 @@ namespace mixin {
template <class Base, class Subtype> template <class Base, class Subtype>
class has_downstreams : public Base { class has_downstreams : public Base {
public: public:
error add_downstream(strong_actor_ptr& ptr) final { error add_downstream(strong_actor_ptr& ptr) override {
CAF_LOG_TRACE(CAF_ARG(ptr)); CAF_LOG_TRACE(CAF_ARG(ptr));
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
if (out().add_path(ptr)) if (out().add_path(ptr))
...@@ -54,7 +54,8 @@ public: ...@@ -54,7 +54,8 @@ public:
return sec::invalid_downstream; return sec::invalid_downstream;
} }
error push(size_t* hint = nullptr) final { error push(size_t* hint = nullptr) override {
CAF_LOG_TRACE("");
if (out().buf_size() > 0) if (out().buf_size() > 0)
out().policy().push(out(), hint); out().policy().push(out(), hint);
return none; return none;
......
...@@ -86,6 +86,9 @@ public: ...@@ -86,6 +86,9 @@ public:
/// A reference-counting pointer to a `stream_handler`. /// A reference-counting pointer to a `stream_handler`.
using stream_handler_ptr = intrusive_ptr<stream_handler>; using stream_handler_ptr = intrusive_ptr<stream_handler>;
/// A container for associating stream IDs to handlers.
using streams_map = std::unordered_map<stream_id, stream_handler_ptr>;
/// The message ID of an outstanding response with its callback. /// The message ID of an outstanding response with its callback.
using pending_response = std::pair<const message_id, behavior>; using pending_response = std::pair<const message_id, behavior>;
...@@ -515,7 +518,7 @@ public: ...@@ -515,7 +518,7 @@ public:
return {in.id(), std::move(ptr)}; return {in.id(), std::move(ptr)};
} }
inline std::unordered_map<stream_id, stream_handler_ptr>& streams() { inline streams_map& streams() {
return streams_; return streams_;
} }
...@@ -596,6 +599,25 @@ public: ...@@ -596,6 +599,25 @@ public:
return bhvr_stack_; return bhvr_stack_;
} }
template <class T, class... Ts>
void fwd_stream_handshake(const stream_id& sid, std::tuple<Ts...>& xs) {
auto mptr = current_mailbox_element();
auto& stages = mptr->stages;
CAF_ASSERT(!stages.empty());
CAF_ASSERT(stages.back() != nullptr);
auto next = std::move(stages.back());
stages.pop_back();
stream<T> token{sid};
auto ys = std::tuple_cat(std::forward_as_tuple(token), std::move(xs));;
next->enqueue(
make_mailbox_element(
mptr->sender, mptr->mid, std::move(stages),
make<stream_msg::open>(sid, make_message_from_tuple(std::move(ys)),
ctrl(), next, stream_priority::normal, false)),
context());
mptr->mid.mark_as_answered();
}
/// @endcond /// @endcond
protected: protected:
...@@ -631,25 +653,6 @@ protected: ...@@ -631,25 +653,6 @@ protected:
swap(g, f); swap(g, f);
} }
template <class T, class... Ts>
void fwd_stream_handshake(const stream_id& sid, std::tuple<Ts...>& xs) {
auto mptr = current_mailbox_element();
auto& stages = mptr->stages;
CAF_ASSERT(!stages.empty());
CAF_ASSERT(stages.back() != nullptr);
auto next = std::move(stages.back());
stages.pop_back();
stream<T> token{sid};
auto ys = std::tuple_cat(std::forward_as_tuple(token), std::move(xs));;
next->enqueue(
make_mailbox_element(
mptr->sender, mptr->mid, std::move(stages),
make<stream_msg::open>(sid, make_message_from_tuple(std::move(ys)),
ctrl(), next, stream_priority::normal, false)),
context());
mptr->mid.mark_as_answered();
}
bool handle_stream_msg(mailbox_element& x, behavior* active_behavior); bool handle_stream_msg(mailbox_element& x, behavior* active_behavior);
// -- Member Variables ------------------------------------------------------- // -- Member Variables -------------------------------------------------------
...@@ -683,7 +686,7 @@ protected: ...@@ -683,7 +686,7 @@ protected:
// TODO: this type is quite heavy in terms of memory, maybe use vector? // TODO: this type is quite heavy in terms of memory, maybe use vector?
/// Holds state for all streams running through this actor. /// Holds state for all streams running through this actor.
std::unordered_map<stream_id, stream_handler_ptr> streams_; streams_map streams_;
# ifndef CAF_NO_EXCEPTIONS # ifndef CAF_NO_EXCEPTIONS
/// Customization point for setting a default exception callback. /// Customization point for setting a default exception callback.
......
...@@ -27,7 +27,8 @@ namespace caf { ...@@ -27,7 +27,8 @@ namespace caf {
abstract_upstream::abstract_upstream(local_actor* selfptr, abstract_upstream::abstract_upstream(local_actor* selfptr,
abstract_upstream::policy_ptr p) abstract_upstream::policy_ptr p)
: self_(selfptr), : self_(selfptr),
policy_(std::move(p)) { policy_(std::move(p)),
continuous_(false) {
// nop // nop
} }
......
...@@ -58,6 +58,7 @@ error stream_stage::upstream_batch(strong_actor_ptr& hdl, size_t xs_size, ...@@ -58,6 +58,7 @@ error stream_stage::upstream_batch(strong_actor_ptr& hdl, size_t xs_size,
} }
error stream_stage::downstream_demand(strong_actor_ptr& hdl, size_t value) { error stream_stage::downstream_demand(strong_actor_ptr& hdl, size_t value) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(value));
auto path = out_ptr_->find(hdl); auto path = out_ptr_->find(hdl);
if (path) { if (path) {
path->open_credit += value; path->open_credit += value;
......
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