Commit 5572f4fd authored by Dominik Charousset's avatar Dominik Charousset

Remove local_actor::forward_to

parent 2bcb1d2e
...@@ -301,14 +301,14 @@ behavior curl_master(stateful_actor<master_state>* self) { ...@@ -301,14 +301,14 @@ behavior curl_master(stateful_actor<master_state>* self) {
self->state.print() << "spawned " << self->state.idle.size() self->state.print() << "spawned " << self->state.idle.size()
<< " worker(s)" << color::reset_endl; << " worker(s)" << color::reset_endl;
return { return {
[=](read_atom, const std::string&, uint64_t, uint64_t) { [=](read_atom rd, std::string& str, uint64_t x, uint64_t y) {
auto& st = self->state; auto& st = self->state;
st.print() << "received {'read'}" << color::reset_endl; st.print() << "received {'read'}" << color::reset_endl;
// forward job to an idle worker // forward job to an idle worker
actor worker = st.idle.back(); actor worker = st.idle.back();
st.idle.pop_back(); st.idle.pop_back();
st.busy.push_back(worker); st.busy.push_back(worker);
self->forward_to(worker); self->delegate(worker, rd, std::move(str), x, y);
st.print() << st.busy.size() << " active jobs" << color::reset_endl; st.print() << st.busy.size() << " active jobs" << color::reset_endl;
if (st.idle.empty()) { if (st.idle.empty()) {
// wait until at least one worker finished its job // wait until at least one worker finished its job
......
...@@ -42,10 +42,6 @@ public: ...@@ -42,10 +42,6 @@ public:
~event_based_actor(); ~event_based_actor();
/// Forwards the last received message to `whom`.
void forward_to(const actor& whom,
message_priority = message_priority::normal);
void initialize() override; void initialize() override;
protected: protected:
......
...@@ -381,11 +381,9 @@ public: ...@@ -381,11 +381,9 @@ public:
/// @endcond /// @endcond
/// Returns the address of the sender of the current message. /// Returns the address of the sender of the current message.
/// @warning Only set during callback invocation. Calling this member function
/// is undefined behavior (dereferencing a `nullptr`) when not in a
/// callback or `forward_to` has been called previously.
inline actor_addr current_sender() { inline actor_addr current_sender() {
return actor_cast<actor_addr>(current_element_->sender); return current_element_ ? actor_cast<actor_addr>(current_element_->sender)
: invalid_actor_addr;
} }
/// Adds a unidirectional `monitor` to `whom`. /// Adds a unidirectional `monitor` to `whom`.
...@@ -534,10 +532,6 @@ public: ...@@ -534,10 +532,6 @@ public:
return (mid.is_request()) ? mid.response_id() : message_id(); return (mid.is_request()) ? mid.response_id() : message_id();
} }
void forward_current_message(const actor& dest);
void forward_current_message(const actor& dest, message_priority mp);
template <class... Ts> template <class... Ts>
void delegate(message_priority mp, const actor& dest, Ts&&... xs) { void delegate(message_priority mp, const actor& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
......
...@@ -30,11 +30,6 @@ event_based_actor::~event_based_actor() { ...@@ -30,11 +30,6 @@ event_based_actor::~event_based_actor() {
// nop // nop
} }
void event_based_actor::forward_to(const actor& whom,
message_priority prio) {
forward_current_message(whom, prio);
}
void event_based_actor::initialize() { void event_based_actor::initialize() {
CAF_LOG_TRACE("subtype =" << logger::render_type_name(typeid(*this)).c_str()); CAF_LOG_TRACE("subtype =" << logger::render_type_name(typeid(*this)).c_str());
is_initialized(true); is_initialized(true);
......
...@@ -160,23 +160,6 @@ std::vector<group> local_actor::joined_groups() const { ...@@ -160,23 +160,6 @@ std::vector<group> local_actor::joined_groups() const {
return result; return result;
} }
void local_actor::forward_current_message(const actor& dest) {
if (! dest)
return;
dest->enqueue(std::move(current_element_), context());
}
void local_actor::forward_current_message(const actor& dest,
message_priority prio) {
if (! dest)
return;
auto mid = current_element_->mid;
current_element_->mid = prio == message_priority::high
? mid.with_high_priority()
: mid.with_normal_priority();
dest->enqueue(std::move(current_element_), context());
}
uint32_t local_actor::request_timeout(const duration& d) { uint32_t local_actor::request_timeout(const duration& d) {
if (! d.valid()) { if (! d.valid()) {
has_timeout(false); has_timeout(false);
......
...@@ -555,13 +555,6 @@ CAF_TEST(constructor_attach) { ...@@ -555,13 +555,6 @@ CAF_TEST(constructor_attach) {
behavior make_behavior() { behavior make_behavior() {
trap_exit(true); trap_exit(true);
testee_ = spawn<testee, monitored>(this); testee_ = spawn<testee, monitored>(this);
auto f = [=](local_actor*, const type_erased_tuple*) -> result<message> {
CAF_MESSAGE("forward to testee");
//auto msg = message::from(x);
forward_to(testee_);
return delegated<message>{};
};
set_unexpected_handler(f);
return { return {
[=](const down_msg& msg) { [=](const down_msg& msg) {
CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown); CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown);
...@@ -574,6 +567,9 @@ CAF_TEST(constructor_attach) { ...@@ -574,6 +567,9 @@ CAF_TEST(constructor_attach) {
if (++downs_ == 2) { if (++downs_ == 2) {
quit(reason); quit(reason);
} }
},
[=](exit_msg& msg) {
delegate(testee_, std::move(msg));
} }
}; };
} }
......
...@@ -137,26 +137,26 @@ public: ...@@ -137,26 +137,26 @@ public:
); );
return {}; return {};
}, },
[=](unpublish_atom, const actor_addr&, uint16_t) -> del_res { [=](unpublish_atom atm, actor_addr addr, uint16_t p) -> del_res {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
forward_current_message(broker_); delegate(broker_, atm, std::move(addr), p);
return {}; return {};
}, },
[=](close_atom, uint16_t) -> del_res { [=](close_atom atm, uint16_t p) -> del_res {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
forward_current_message(broker_); delegate(broker_, atm, p);
return {}; return {};
}, },
[=](spawn_atom, const node_id&, const std::string&, const message&) [=](spawn_atom atm, node_id& nid, std::string& str, message& msg)
-> delegated<ok_atom, strong_actor_ptr, mpi_set> { -> delegated<ok_atom, strong_actor_ptr, mpi_set> {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
forward_current_message(broker_); delegate(broker_, atm, std::move(nid), std::move(str), std::move(msg));
return {}; return {};
}, },
[=](get_atom, const node_id&) [=](get_atom atm, node_id nid)
-> delegated<node_id, std::string, uint16_t> { -> delegated<node_id, std::string, uint16_t> {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
forward_current_message(broker_); delegate(broker_, atm, std::move(nid));
return {}; return {};
}, },
[=](const down_msg& dm) { [=](const down_msg& dm) {
......
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