Commit ce7a5a50 authored by Dominik Charousset's avatar Dominik Charousset

Enable actors to receive streams as response

This patch is only one half of enabling actors to "request" streams from
others. Response IDs are currently not accounted for. This requires to
somehow signalize the *intent* of setting a message ID. The current use
case for this is starting a stream and then receiving the result.
However, when attaching oneself to a stream as sink and wanting to
provide the sink event handlers via response handler (e.g. via .then)
then actors in the pipeline need to be able to figure out which of the
two options is true.
parent 8a1d8241
......@@ -650,7 +650,9 @@ protected:
mptr->mid.mark_as_answered();
}
// -- member variables -------------------------------------------------------
bool handle_stream_msg(mailbox_element& x, behavior* active_behavior);
// -- Member Variables -------------------------------------------------------
/// Stores user-defined callbacks for message handling.
detail::behavior_stack bhvr_stack_;
......
......@@ -37,7 +37,7 @@ public:
using result_type = std::pair<error, iterator>;
stream_msg_visitor(scheduled_actor* self, stream_id& sid,
iterator i, iterator last);
iterator i, iterator last, behavior* bhvr);
result_type operator()(stream_msg::open& x);
......@@ -60,6 +60,7 @@ private:
stream_id& sid_;
iterator i_;
iterator e_;
behavior* bhvr_;
};
} // namespace caf
......
......@@ -389,23 +389,8 @@ scheduled_actor::categorize(mailbox_element& x) {
return message_category::internal;
}
case make_type_token<stream_msg>(): {
auto& sm = content.get_mutable_as<stream_msg>(0);
auto e = streams_.end();
stream_msg_visitor f{this, sm.sid, streams_.find(sm.sid), e};
auto res = apply_visitor(f, sm.content);
auto i = res.second;
if (res.first) {
if (i != e) {
i->second->abort(current_sender(), std::move(res.first));
streams_.erase(i);
}
} else if (i != e) {
if (i->second->done()) {
streams_.erase(i);
if (streams_.empty() && !has_behavior())
quit(exit_reason::normal);
}
}
auto& bs = bhvr_stack();
handle_stream_msg(x, bs.empty() ? nullptr : &bs.back());
return message_category::internal;
}
default:
......@@ -417,15 +402,34 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x));
current_element_ = &x;
CAF_LOG_RECEIVE_EVENT(current_element_);
// short-circuit awaited responses
// Helper function for dispatching a message to a response handler.
using ptr_t = scheduled_actor*;
using fun_t = bool (*)(ptr_t, behavior&, mailbox_element&);
auto ordinary_invoke = [](ptr_t, behavior& f, mailbox_element& in) -> bool {
return f(in.content()) != none;
};
auto stream_invoke = [](ptr_t p, behavior& f, mailbox_element& in) -> bool {
// The only legal stream message in a response is `stream_open`.
auto& var = in.content().get_as<stream_msg>(0).content;
if (holds_alternative<stream_msg::open>(var))
return p->handle_stream_msg(in, &f);
return false;
};
auto select_invoke_fun = [&]() -> fun_t {
if (x.content().type_token() != make_type_token<stream_msg>())
return ordinary_invoke;
return stream_invoke;
};
// Short-circuit awaited responses.
if (!awaited_responses_.empty()) {
auto invoke = select_invoke_fun();
auto& pr = awaited_responses_.front();
// skip all messages until we receive the currently awaited response
if (x.mid != pr.first)
return im_skipped;
auto f = std::move(pr.second);
awaited_responses_.pop_front();
if (!f(x.content())) {
if (!invoke(this, f, x)) {
// try again with error if first attempt failed
auto msg = make_message(make_error(sec::unexpected_response,
x.move_content_to_message()));
......@@ -433,13 +437,14 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
}
return im_success;
}
// handle multiplexed responses
// Handle multiplexed responses.
if (x.mid.is_response()) {
auto invoke = select_invoke_fun();
auto mrh = multiplexed_responses_.find(x.mid);
// neither awaited nor multiplexed, probably an expired timeout
if (mrh == multiplexed_responses_.end())
return im_dropped;
if (!mrh->second(x.content())) {
if (!invoke(this, mrh->second, x)) {
// try again with error if first attempt failed
auto msg = make_message(make_error(sec::unexpected_response,
x.move_content_to_message()));
......@@ -448,7 +453,7 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
multiplexed_responses_.erase(mrh);
return im_success;
}
// dispatch on the content of x
// Dispatch on the content of x.
switch (categorize(x)) {
case message_category::expired_timeout:
CAF_LOG_DEBUG("dropped expired timeout message");
......@@ -504,7 +509,7 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
return !skipped ? im_success : im_skipped;
}
}
// should be unreachable
// Unreachable.
CAF_CRITICAL("invalid message type");
}
......@@ -643,4 +648,28 @@ bool scheduled_actor::finalize() {
return true;
}
bool scheduled_actor::handle_stream_msg(mailbox_element& x,
behavior* active_behavior) {
CAF_ASSERT(x.content().match_elements<stream_msg>());
auto& sm = x.content().get_mutable_as<stream_msg>(0);
auto e = streams_.end();
stream_msg_visitor f{this, sm.sid, streams_.find(sm.sid), e, active_behavior};
auto res = apply_visitor(f, sm.content);
auto success = (res.first == none);
auto i = res.second;
if (!success) {
if (i != e) {
i->second->abort(current_sender(), std::move(res.first));
streams_.erase(i);
}
} else if (i != e) {
if (i->second->done()) {
streams_.erase(i);
if (streams_.empty() && !has_behavior())
quit(exit_reason::normal);
}
}
return success;
}
} // namespace caf
......@@ -26,37 +26,39 @@
namespace caf {
stream_msg_visitor::stream_msg_visitor(scheduled_actor* self, stream_id& sid,
iterator i, iterator last)
iterator i, iterator last,
behavior* bhvr)
: self_(self),
sid_(sid),
i_(i),
e_(last) {
e_(last),
bhvr_(bhvr) {
// nop
}
auto stream_msg_visitor::operator()(stream_msg::open& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
CAF_ASSERT(self_->current_mailbox_element() != nullptr);
// Make sure to not add an actor twice.
if (i_ != e_)
return {sec::downstream_already_exists, e_};
auto& predecessor = x.prev_stage;
// Convenience function for aborting the stream on error.
auto fail = [&](error reason) -> result_type {
unsafe_send_as(self_, predecessor, make<stream_msg::abort>(sid_, reason));
auto rp = self_->make_response_promise();
rp.deliver(reason);
return {std::move(reason), e_};
};
// Sanity checks.
if (bhvr_ == nullptr)
return fail(sec::stream_init_failed);
if (!predecessor) {
CAF_LOG_WARNING("received stream_msg::open with empty prev_stage");
return fail(sec::invalid_upstream);
}
auto& bs = self_->bhvr_stack();
if (bs.empty()) {
CAF_LOG_WARNING("cannot open stream in actor without behavior");
return fail(sec::stream_init_failed);
}
auto bhvr = self_->bhvr_stack().back();
auto res = bhvr(x.msg);
// Invoke behavior of parent to perform handshake.
auto res = (*bhvr_)(x.msg);
if (!res) {
CAF_LOG_WARNING("actor did not respond to handshake:" << CAF_ARG(x.msg));
return fail(sec::stream_init_failed);
......@@ -69,10 +71,10 @@ auto stream_msg_visitor::operator()(stream_msg::open& x) -> result_type {
return fail(sec::stream_init_failed);
}
auto& handler = i_->second;
// store upstream actor
// Store upstream actor.
auto initial_credit = handler->add_upstream(x.prev_stage, sid_, x.priority);
if (initial_credit) {
// send ACK to predecessor
// Send ACK to predecessor.
auto ic = static_cast<int32_t>(*initial_credit);
unsafe_send_as(self_, predecessor,
make_message(make<stream_msg::ack_open>(
......
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