Commit e1473d7f authored by Dominik Charousset's avatar Dominik Charousset

Implement round-trip from string input to response

parent f2395f58
...@@ -19,11 +19,13 @@ ...@@ -19,11 +19,13 @@
#pragma once #pragma once
#include "caf/actor_traits.hpp" #include "caf/actor_traits.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp" #include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp" #include "caf/intrusive/fifo_inbox.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/none.hpp" #include "caf/none.hpp"
...@@ -33,7 +35,8 @@ namespace caf::net { ...@@ -33,7 +35,8 @@ namespace caf::net {
/// ///
class actor_shell class actor_shell
: public extend<local_actor, actor_shell>::with<mixin::sender>, : public extend<local_actor, actor_shell>::with<mixin::sender,
mixin::requester>,
public dynamically_typed_actor_base, public dynamically_typed_actor_base,
public non_blocking_actor_base { public non_blocking_actor_base {
public: public:
...@@ -43,7 +46,8 @@ public: ...@@ -43,7 +46,8 @@ public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = extend<local_actor, actor_shell>::with<mixin::sender>; using super
= extend<local_actor, actor_shell>::with<mixin::sender, mixin::requester>;
using signatures = none_t; using signatures = none_t;
...@@ -82,6 +86,12 @@ public: ...@@ -82,6 +86,12 @@ public:
/// `nullptr` if the mailbox is empty. /// `nullptr` if the mailbox is empty.
mailbox_element_ptr next_message(); mailbox_element_ptr next_message();
/// Tries to put the mailbox into the `blocked` state, causing the next
/// enqueue to register the owning socket manager for write events.
bool try_block_mailbox();
// -- message processing -----------------------------------------------------
/// Dequeues and processes the next message from the mailbox. /// Dequeues and processes the next message from the mailbox.
/// @param bhvr Available message handlers. /// @param bhvr Available message handlers.
/// @param fallback Callback for processing message that failed to match /// @param fallback Callback for processing message that failed to match
...@@ -91,9 +101,8 @@ public: ...@@ -91,9 +101,8 @@ public:
bool consume_message(behavior& bhvr, bool consume_message(behavior& bhvr,
callback<result<message>(message&)>& fallback); callback<result<message>(message&)>& fallback);
/// Tries to put the mailbox into the `blocked` state, causing the next /// Adds a callback for a multiplexed response.
/// enqueue to register the owning socket manager for write events. void add_multiplexed_response_handler(message_id response_id, behavior bhvr);
bool try_block_mailbox();
// -- overridden functions of abstract_actor --------------------------------- // -- overridden functions of abstract_actor ---------------------------------
...@@ -120,6 +129,9 @@ private: ...@@ -120,6 +129,9 @@ private:
// Points to the owning manager (nullptr after quit was called). // Points to the owning manager (nullptr after quit was called).
socket_manager* owner_; socket_manager* owner_;
// Stores callbacks for multiplexed responses.
detail::unordered_flat_map<message_id, behavior> multiplexed_responses_;
}; };
/// An "owning" pointer to an actor shell in the sense that it calls `quit()` on /// An "owning" pointer to an actor shell in the sense that it calls `quit()` on
......
...@@ -29,6 +29,8 @@ ...@@ -29,6 +29,8 @@
namespace caf::net { namespace caf::net {
// -- constructors, destructors, and assignment operators ----------------------
actor_shell::actor_shell(actor_config& cfg, socket_manager* owner) actor_shell::actor_shell(actor_config& cfg, socket_manager* owner)
: super(cfg), mailbox_(policy::normal_messages{}), owner_(owner) { : super(cfg), mailbox_(policy::normal_messages{}), owner_(owner) {
mailbox_.try_block(); mailbox_.try_block();
...@@ -38,10 +40,14 @@ actor_shell::~actor_shell() { ...@@ -38,10 +40,14 @@ actor_shell::~actor_shell() {
// nop // nop
} }
// -- state modifiers ----------------------------------------------------------
void actor_shell::quit(error reason) { void actor_shell::quit(error reason) {
cleanup(std::move(reason), nullptr); cleanup(std::move(reason), nullptr);
} }
// -- mailbox access -----------------------------------------------------------
mailbox_element_ptr actor_shell::next_message() { mailbox_element_ptr actor_shell::next_message() {
mailbox_.fetch_more(); mailbox_.fetch_more();
auto& q = mailbox_.queue(); auto& q = mailbox_.queue();
...@@ -52,6 +58,12 @@ mailbox_element_ptr actor_shell::next_message() { ...@@ -52,6 +58,12 @@ mailbox_element_ptr actor_shell::next_message() {
return nullptr; return nullptr;
} }
bool actor_shell::try_block_mailbox() {
return mailbox_.try_block();
}
// -- message processing -------------------------------------------------------
bool actor_shell::consume_message( bool actor_shell::consume_message(
behavior& bhvr, callback<result<message>(message&)>& fallback) { behavior& bhvr, callback<result<message>(message&)>& fallback) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
...@@ -59,12 +71,26 @@ bool actor_shell::consume_message( ...@@ -59,12 +71,26 @@ bool actor_shell::consume_message(
current_element_ = msg.get(); current_element_ = msg.get();
CAF_LOG_RECEIVE_EVENT(current_element_); CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, *msg); CAF_BEFORE_PROCESSING(this, *msg);
detail::default_invoke_result_visitor<actor_shell> visitor{this}; auto mid = msg->mid;
if (auto result = bhvr(msg->payload)) { if (!mid.is_response()) {
visitor(*result); detail::default_invoke_result_visitor<actor_shell> visitor{this};
} else { if (auto result = bhvr(msg->payload)) {
auto fallback_result = fallback(msg->payload); visitor(*result);
visit(visitor, fallback_result); } else {
auto fallback_result = fallback(msg->payload);
visit(visitor, fallback_result);
}
} else if (auto i = multiplexed_responses_.find(mid);
i != multiplexed_responses_.end()) {
auto bhvr = std::move(i->second);
multiplexed_responses_.erase(i);
auto res = bhvr(msg->payload);
if (!res) {
CAF_LOG_DEBUG("got unexpected_response");
auto err_msg = make_message(
make_error(sec::unexpected_response, std::move(msg->payload)));
bhvr(err_msg);
}
} }
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed); CAF_AFTER_PROCESSING(this, invoke_message_result::consumed);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(invoke_message_result::consumed); CAF_LOG_SKIP_OR_FINALIZE_EVENT(invoke_message_result::consumed);
...@@ -73,10 +99,15 @@ bool actor_shell::consume_message( ...@@ -73,10 +99,15 @@ bool actor_shell::consume_message(
return false; return false;
} }
bool actor_shell::try_block_mailbox() { void actor_shell::add_multiplexed_response_handler(message_id response_id,
return mailbox_.try_block(); behavior bhvr) {
if (bhvr.timeout() != infinite)
request_response_timeout(bhvr.timeout(), response_id);
multiplexed_responses_.emplace(response_id, std::move(bhvr));
} }
// -- overridden functions of abstract_actor -----------------------------------
void actor_shell::enqueue(mailbox_element_ptr ptr, execution_unit*) { void actor_shell::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(!getf(is_blocking_flag)); CAF_ASSERT(!getf(is_blocking_flag));
......
...@@ -30,6 +30,8 @@ ...@@ -30,6 +30,8 @@
using namespace caf; using namespace caf;
using namespace std::string_literals;
namespace { namespace {
using byte_span = span<const byte>; using byte_span = span<const byte>;
...@@ -39,8 +41,14 @@ using svec = std::vector<std::string>; ...@@ -39,8 +41,14 @@ using svec = std::vector<std::string>;
struct app_t { struct app_t {
using input_tag = tag::stream_oriented; using input_tag = tag::stream_oriented;
app_t() = default;
explicit app_t(actor hdl) : worker(std::move(hdl)) {
// nop
}
template <class LowerLayer> template <class LowerLayer>
error init(net::socket_manager* mgr, LowerLayer&, const settings&) { error init(net::socket_manager* mgr, LowerLayer& down, const settings&) {
self = mgr->make_actor_shell(); self = mgr->make_actor_shell();
bhvr = behavior{ bhvr = behavior{
[this](std::string& line) { lines.emplace_back(std::move(line)); }, [this](std::string& line) { lines.emplace_back(std::move(line)); },
...@@ -49,6 +57,7 @@ struct app_t { ...@@ -49,6 +57,7 @@ struct app_t {
CAF_FAIL("unexpected message: " << msg); CAF_FAIL("unexpected message: " << msg);
return make_error(sec::unexpected_message); return make_error(sec::unexpected_message);
}); });
down.configure_read(net::receive_policy::up_to(2048));
return none; return none;
} }
...@@ -70,18 +79,80 @@ struct app_t { ...@@ -70,18 +79,80 @@ struct app_t {
} }
template <class LowerLayer> template <class LowerLayer>
ptrdiff_t consume(LowerLayer&, byte_span, byte_span) { ptrdiff_t consume(LowerLayer& down, byte_span buf, byte_span) {
CAF_FAIL("received unexpected data"); // Seek newline character.
return -1; constexpr auto nl = byte{'\n'};
if (auto i = std::find(buf.begin(), buf.end(), nl); i != buf.end()) {
// Skip empty lines.
if (i == buf.begin()) {
consumed_bytes += 1;
auto sub_res = consume(down, buf.subspan(1), {});
return sub_res >= 0 ? sub_res + 1 : sub_res;
}
// Deserialize config value from received line.
auto num_bytes = std::distance(buf.begin(), i) + 1;
string_view line{reinterpret_cast<const char*>(buf.data()),
static_cast<size_t>(num_bytes) - 1};
config_value val;
if (auto parsed_res = config_value::parse(line)) {
val = std::move(*parsed_res);
} else {
down.abort_reason(std::move(parsed_res.error()));
return -1;
}
if (!holds_alternative<settings>(val)) {
down.abort_reason(
make_error(pec::type_mismatch,
"expected a dictionary, got a "s + val.type_name()));
return -1;
}
// Deserialize message from received dictionary.
config_value_reader reader{&val};
caf::message msg;
if (!reader.apply_object(msg)) {
down.abort_reason(reader.get_error());
return -1;
}
// Dispatch message to worker.
CAF_MESSAGE("app received a message from its socket: " << msg);
self->request(worker, std::chrono::seconds{1}, std::move(msg))
.then(
[this](int32_t value) {
++received_responses;
CAF_CHECK_EQUAL(value, 246);
},
[](error&) {
// TODO: implement me
});
// Try consuming more from the buffer.
consumed_bytes += static_cast<size_t>(num_bytes);
auto sub_buf = buf.subspan(num_bytes);
auto sub_res = consume(down, sub_buf, {});
return sub_res >= 0 ? num_bytes + sub_res : sub_res;
}
return 0;
} }
// Handle to the worker-under-test.
actor worker;
// Lines received as asynchronous messages.
std::vector<std::string> lines; std::vector<std::string> lines;
// Actor shell representing this app.
net::actor_shell_ptr self; net::actor_shell_ptr self;
// Handler for consuming messages from the mailbox.
behavior bhvr; behavior bhvr;
// Handler for unexpected messages.
unique_callback_ptr<result<message>(message&)> fallback; unique_callback_ptr<result<message>(message&)> fallback;
// Counts how many bytes we've consumed in total.
size_t consumed_bytes = 0;
// Counts how many response messages we've received from the worker.
size_t received_responses = 0;
}; };
struct fixture : host_fixture, test_coordinator_fixture<> { struct fixture : host_fixture, test_coordinator_fixture<> {
...@@ -108,12 +179,23 @@ struct fixture : host_fixture, test_coordinator_fixture<> { ...@@ -108,12 +179,23 @@ struct fixture : host_fixture, test_coordinator_fixture<> {
} }
CAF_FAIL("reached max repeat rate without meeting the predicate"); CAF_FAIL("reached max repeat rate without meeting the predicate");
} }
void send(string_view str) {
auto res = write(self_socket_guard.socket(), as_bytes(make_span(str)));
if (res != static_cast<ptrdiff_t>(str.size()))
CAF_FAIL("expected write() to return " << str.size() << ", got: " << res);
}
net::middleman mm; net::middleman mm;
net::multiplexer mpx; net::multiplexer mpx;
net::socket_guard<net::stream_socket> self_socket_guard; net::socket_guard<net::stream_socket> self_socket_guard;
net::socket_guard<net::stream_socket> testee_socket_guard; net::socket_guard<net::stream_socket> testee_socket_guard;
}; };
constexpr std::string_view input = R"__(
{ values = [ { "@type" : "int32_t", value: 123 } ] }
)__";
} // namespace } // namespace
CAF_TEST_FIXTURE_SCOPE(actor_shell_tests, fixture) CAF_TEST_FIXTURE_SCOPE(actor_shell_tests, fixture)
...@@ -132,4 +214,22 @@ CAF_TEST(actor shells expose their mailbox to their owners) { ...@@ -132,4 +214,22 @@ CAF_TEST(actor shells expose their mailbox to their owners) {
CAF_CHECK_EQUAL(app.lines, svec({"line 1", "line 2", "line 3"})); CAF_CHECK_EQUAL(app.lines, svec({"line 1", "line 2", "line 3"}));
} }
CAF_TEST(actor shells can send requests and receive responses) {
auto worker = sys.spawn([] {
return behavior{
[](int32_t value) { return value * 2; },
};
});
auto sck = testee_socket_guard.release();
auto mgr = net::make_socket_manager<app_t, net::stream_transport>(sck, &mpx,
worker);
if (auto err = mgr->init(content(cfg)))
CAF_FAIL("mgr->init() failed: " << err);
auto& app = mgr->top_layer();
send(input);
run_while([&] { return app.consumed_bytes != input.size(); });
expect((int32_t), to(worker).with(123));
run_while([&] { return app.received_responses != 1; });
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
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