Commit 9934dbc1 authored by Dominik Charousset's avatar Dominik Charousset

Remove clumsy `.continue_with()` feature

parent 1c447b62
...@@ -44,11 +44,6 @@ class behavior { ...@@ -44,11 +44,6 @@ class behavior {
public: public:
friend class message_handler; friend class message_handler;
/**
* The type of continuations that can be used to extend an existing behavior.
*/
using continuation_fun = std::function<optional<message>(message&)>;
behavior() = default; behavior() = default;
behavior(behavior&&) = default; behavior(behavior&&) = default;
behavior(const behavior&) = default; behavior(const behavior&) = default;
...@@ -115,12 +110,6 @@ class behavior { ...@@ -115,12 +110,6 @@ class behavior {
return (m_impl) ? m_impl->invoke(std::forward<T>(arg)) : none; return (m_impl) ? m_impl->invoke(std::forward<T>(arg)) : none;
} }
/**
* Adds a continuation to this behavior that is executed
* whenever this behavior was successfully applied to a message.
*/
behavior add_continuation(continuation_fun fun);
/** /**
* Checks whether this behavior is not empty. * Checks whether this behavior is not empty.
*/ */
......
...@@ -38,35 +38,17 @@ class local_actor; ...@@ -38,35 +38,17 @@ class local_actor;
class continue_helper { class continue_helper {
public: public:
using message_id_wrapper_tag = int; using message_id_wrapper_tag = int;
using getter = std::function<optional<behavior&> (message_id msg_id)>; continue_helper(message_id mid);
continue_helper(message_id mid, getter get_sync_handler);
/**
* Adds the continuation `fun` to the synchronous message handler
* that is invoked if the response handler successfully returned.
*/
template <class F>
continue_helper& continue_with(F fun) {
return continue_with(behavior::continuation_fun{
message_handler{on(any_vals, arg_match) >> fun}});
}
/**
* Adds the continuation `fun` to the synchronous message handler
* that is invoked if the response handler successfully returned.
*/
continue_helper& continue_with(behavior::continuation_fun fun);
/** /**
* Returns the ID of the expected response message. * Returns the ID of the expected response message.
*/ */
message_id get_message_id() const { return m_mid; } message_id get_message_id() const {
return m_mid;
}
private: private:
message_id m_mid; message_id m_mid;
getter m_getter;
//local_actor* m_self;
}; };
} // namespace caf } // namespace caf
......
...@@ -153,36 +153,22 @@ class invoke_policy { ...@@ -153,36 +153,22 @@ class invoke_policy {
auto id = res->template get_as<uint64_t>(1); auto id = res->template get_as<uint64_t>(1);
auto msg_id = message_id::from_integer_value(id); auto msg_id = message_id::from_integer_value(id);
auto ref_opt = self->sync_handler(msg_id); auto ref_opt = self->sync_handler(msg_id);
// calls self->response_promise() if hdl is a dummy // install a behavior that calls the user-defined behavior
// argument, forwards hdl otherwise to reply to the // and using the result of its inner behavior as response
// original request message
auto fhdl = fetch_response_promise(self, hdl);
if (ref_opt) { if (ref_opt) {
behavior cpy = *ref_opt; auto fhdl = fetch_response_promise(self, hdl);
*ref_opt = behavior inner = *ref_opt;
cpy.add_continuation([=](message & intermediate) *ref_opt = behavior {
->optional<message> { others() >> [=] {
if (!intermediate.empty()) { // inner is const inside this lambda and mutable a C++14 feature
// do no use lamba expresion type to behavior cpy = inner;
// avoid recursive template instantiaton auto inner_res = cpy(self->last_dequeued());
behavior::continuation_fun f2 = [=]( if (inner_res) {
message & m)->optional<message> { fhdl.deliver(*inner_res);
return std::move(m);
};
auto mutable_mid = mid;
// recursively call invoke_fun on the
// result to correctly handle stuff like
// sync_send(...).then(...).then(...)...
return this->invoke_fun(self, intermediate,
mutable_mid, f2, fhdl);
} }
return none; }
}); };
} }
// reset res to prevent "outer" invoke_fun
// from handling the result again
res->reset();
} else { } else {
// respond by using the result of 'fun' // respond by using the result of 'fun'
CAF_LOG_DEBUG("respond via response_promise"); CAF_LOG_DEBUG("respond via response_promise");
......
...@@ -80,8 +80,7 @@ class response_handle<Self, message, nonblocking_response_handle_tag> { ...@@ -80,8 +80,7 @@ class response_handle<Self, message, nonblocking_response_handle_tag> {
} }
}; };
m_self->bhvr_stack().push_back(std::move(bhvr), m_mid); m_self->bhvr_stack().push_back(std::move(bhvr), m_mid);
auto ptr = m_self; return {m_mid};
return {m_mid, [ptr](message_id mid) { return ptr->sync_handler(mid); }};
} }
private: private:
...@@ -126,8 +125,7 @@ class response_handle<Self, TypedOutputPair, nonblocking_response_handle_tag> { ...@@ -126,8 +125,7 @@ class response_handle<Self, TypedOutputPair, nonblocking_response_handle_tag> {
} }
}; };
m_self->bhvr_stack().push_back(std::move(tmp), m_mid); m_self->bhvr_stack().push_back(std::move(tmp), m_mid);
auto get = [selfptr](message_id mid) { return selfptr->sync_handler(mid); }; return {m_mid};
return {m_mid, get};
} }
private: private:
......
...@@ -32,10 +32,8 @@ template <class OutputList> ...@@ -32,10 +32,8 @@ template <class OutputList>
class typed_continue_helper { class typed_continue_helper {
public: public:
using message_id_wrapper_tag = int; using message_id_wrapper_tag = int;
using getter = std::function<optional<behavior&> (message_id msg_id)>;
typed_continue_helper(message_id mid, getter get_sync_handler) typed_continue_helper(message_id mid) : m_ch(mid) {
: m_ch(mid, std::move(get_sync_handler)) {
// nop // nop
} }
...@@ -43,14 +41,6 @@ class typed_continue_helper { ...@@ -43,14 +41,6 @@ class typed_continue_helper {
// nop // nop
} }
template <class F>
typed_continue_helper<typename detail::get_callable_trait<F>::result_type>
continue_with(F fun) {
detail::type_checker<OutputList, F>::check();
m_ch.continue_with(std::move(fun));
return {m_ch};
}
message_id get_message_id() const { message_id get_message_id() const {
return m_ch.get_message_id(); return m_ch.get_message_id();
} }
......
...@@ -24,57 +24,8 @@ ...@@ -24,57 +24,8 @@
namespace caf { namespace caf {
namespace {
class continuation_decorator : public detail::behavior_impl {
public:
using super = behavior_impl;
using continuation_fun = behavior::continuation_fun;
using pointer = typename behavior_impl::pointer;
continuation_decorator(continuation_fun fun, pointer ptr)
: super(ptr->timeout()), m_fun(fun), m_decorated(std::move(ptr)) {
CAF_REQUIRE(m_decorated != nullptr);
}
template <class T>
inline bhvr_invoke_result invoke_impl(T& tup) {
auto res = m_decorated->invoke(tup);
if (res) {
return m_fun(*res);
}
return none;
}
bhvr_invoke_result invoke(message& tup) {
return invoke_impl(tup);
}
bhvr_invoke_result invoke(const message& tup) {
return invoke_impl(tup);
}
pointer copy(const generic_timeout_definition& tdef) const {
return new continuation_decorator(m_fun, m_decorated->copy(tdef));
}
void handle_timeout() {
m_decorated->handle_timeout();
}
private:
continuation_fun m_fun;
pointer m_decorated;
};
} // namespace <anonymous>
behavior::behavior(const message_handler& mh) : m_impl(mh.as_behavior_impl()) { behavior::behavior(const message_handler& mh) : m_impl(mh.as_behavior_impl()) {
// nop // nop
} }
behavior behavior::add_continuation(continuation_fun fun) {
return behavior::impl_ptr{new continuation_decorator(std::move(fun), m_impl)};
}
} // namespace caf } // namespace caf
...@@ -22,18 +22,8 @@ ...@@ -22,18 +22,8 @@
namespace caf { namespace caf {
continue_helper::continue_helper(message_id mid, getter g) continue_helper::continue_helper(message_id mid) : m_mid(mid) {
: m_mid(mid), m_getter(std::move(g)) {} // nop
continue_helper& continue_helper::continue_with(behavior::continuation_fun f) {
auto ref_opt = m_getter(m_mid); //m_self->sync_handler(m_mid);
if (ref_opt) {
behavior cpy = *ref_opt;
*ref_opt = cpy.add_continuation(std::move(f));
} else {
CAF_LOGF_ERROR("failed to add continuation");
}
return *this;
} }
} // namespace caf } // namespace caf
...@@ -30,7 +30,6 @@ add_unit_test(simple_reply_response) ...@@ -30,7 +30,6 @@ add_unit_test(simple_reply_response)
add_unit_test(serial_reply) add_unit_test(serial_reply)
add_unit_test(or_else) add_unit_test(or_else)
add_unit_test(either) add_unit_test(either)
add_unit_test(continuation)
add_unit_test(constructor_attach) add_unit_test(constructor_attach)
add_unit_test(custom_exception_handler) add_unit_test(custom_exception_handler)
add_unit_test(typed_spawn) add_unit_test(typed_spawn)
......
#include "test.hpp"
#include "caf/all.hpp"
using namespace caf;
behavior simple_mirror(event_based_actor* self) {
return {
others() >> [=] {
return self->last_dequeued();
}
};
}
void test_continuation() {
auto mirror = spawn(simple_mirror);
spawn([=](event_based_actor* self) {
self->sync_send(mirror, 42).then(
on(42) >> [] {
return "fourty-two";
}
).continue_with(
[=](const std::string& ref) {
CAF_CHECK_EQUAL(ref, "fourty-two");
return 4.2f;
}
).continue_with(
[=](float f) {
CAF_CHECK_EQUAL(f, 4.2f);
self->send_exit(mirror, exit_reason::user_shutdown);
self->quit();
}
);
});
await_all_actors_done();
}
int main() {
CAF_TEST(test_continuation);
test_continuation();
return CAF_TEST_RESULT();
}
...@@ -51,6 +51,44 @@ void spawn5_server_impl(event_based_actor* self, actor client, group grp) { ...@@ -51,6 +51,44 @@ void spawn5_server_impl(event_based_actor* self, actor client, group grp) {
CAF_PRINT("monitor actor: " << to_string(a)); CAF_PRINT("monitor actor: " << to_string(a));
self->monitor(a); self->monitor(a);
} }
CAF_PRINT("wait for reflected messages");
// receive seven reply messages (2 local, 5 remote)
auto replies = std::make_shared<int>(0);
self->become(
on("Hello reflectors!", 5.0) >> [=] {
if (++*replies == 7) {
CAF_PRINT("wait for DOWN messages");
auto downs = std::make_shared<int>(0);
self->become(
[=](const down_msg& dm) {
if (dm.reason != exit_reason::normal) {
CAF_PRINTERR("reflector exited for non-normal exit reason!");
}
if (++*downs == 5) {
CAF_CHECKPOINT();
self->send(client, atom("Spawn5Done"));
self->quit();
}
},
others() >> [=] {
CAF_UNEXPECTED_MSG(self);
self->quit(exit_reason::user_defined);
},
after(chrono::seconds(2)) >> [=] {
CAF_UNEXPECTED_TOUT();
CAF_LOGF_ERROR("did only receive " << *downs << " down messages");
self->quit(exit_reason::user_defined);
}
);
}
},
after(std::chrono::seconds(2)) >> [=] {
CAF_UNEXPECTED_TOUT();
CAF_LOGF_ERROR("did only receive " << *replies
<< " responses to 'Hello reflectors!'");
self->quit(exit_reason::user_defined);
}
);
}, },
others() >> [=] { others() >> [=] {
CAF_UNEXPECTED_MSG(self); CAF_UNEXPECTED_MSG(self);
...@@ -58,48 +96,9 @@ void spawn5_server_impl(event_based_actor* self, actor client, group grp) { ...@@ -58,48 +96,9 @@ void spawn5_server_impl(event_based_actor* self, actor client, group grp) {
}, },
after(chrono::seconds(10)) >> [=] { after(chrono::seconds(10)) >> [=] {
CAF_UNEXPECTED_TOUT(); CAF_UNEXPECTED_TOUT();
self->quit(exit_reason::user_defined); self->quit(exit_reason::user_defined);
} }
).continue_with([=] { );
CAF_PRINT("wait for reflected messages");
// receive seven reply messages (2 local, 5 remote)
auto replies = std::make_shared<int>(0);
self->become(
on("Hello reflectors!", 5.0) >> [=] {
if (++*replies == 7) {
CAF_PRINT("wait for DOWN messages");
auto downs = std::make_shared<int>(0);
self->become(
[=](const down_msg& dm) {
if (dm.reason != exit_reason::normal) {
CAF_PRINTERR("reflector exited for non-normal exit reason!");
}
if (++*downs == 5) {
CAF_CHECKPOINT();
self->send(client, atom("Spawn5Done"));
self->quit();
}
},
others() >> [=] {
CAF_UNEXPECTED_MSG(self);
self->quit(exit_reason::user_defined);
},
after(chrono::seconds(2)) >> [=] {
CAF_UNEXPECTED_TOUT();
CAF_LOGF_ERROR("did only receive " << *downs << " down messages");
self->quit(exit_reason::user_defined);
}
);
}
},
after(std::chrono::seconds(2)) >> [=] {
CAF_UNEXPECTED_TOUT();
CAF_LOGF_ERROR("did only receive " << *replies
<< " responses to 'Hello reflectors!'");
self->quit(exit_reason::user_defined);
}
);
});
} }
// receive seven reply messages (2 local, 5 remote) // receive seven reply messages (2 local, 5 remote)
......
...@@ -66,18 +66,17 @@ class typed_server3 : public server_type::base { ...@@ -66,18 +66,17 @@ class typed_server3 : public server_type::base {
}; };
void client(event_based_actor* self, actor parent, server_type serv) { void client(event_based_actor* self, actor parent, server_type serv) {
self->sync_send(serv, my_request{0, 0}) self->sync_send(serv, my_request{0, 0}).then(
.then([](bool value)->int { [=](bool value) {
CAF_CHECK_EQUAL(value, true); CAF_CHECK_EQUAL(value, true);
return 42; self->sync_send(serv, my_request{10, 20}).then(
}) [=](bool value) {
.continue_with([=](int ival) { CAF_CHECK_EQUAL(value, false);
CAF_CHECK_EQUAL(ival, 42); self->send(parent, atom("passed"));
self->sync_send(serv, my_request{10, 20}).then([=](bool value) { }
CAF_CHECK_EQUAL(value, false); );
self->send(parent, atom("passed")); }
}); );
});
} }
void test_typed_spawn(server_type ts) { void test_typed_spawn(server_type ts) {
......
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