Commit 293a0153 authored by Dominik Charousset's avatar Dominik Charousset

added `response_handle` for delayed responses

allows event-based actors to delay a synchronous response,
e.g., to reply to a previous message in a sync_send(...).then(...).
parent 060f9144
......@@ -135,6 +135,7 @@ set(LIBCPPA_SRC
src/protocol.cpp
src/receive.cpp
src/ref_counted.cpp
src/response_handle.cpp
src/ripemd_160.cpp
src/scheduled_actor.cpp
src/scheduled_actor_dummy.cpp
......
......@@ -295,3 +295,5 @@ cppa/weak_ptr_anchor.hpp
src/weak_ptr_anchor.cpp
cppa/actor_companion_mixin.hpp
cppa/detail/singleton_mixin.hpp
cppa/response_handle.hpp
src/response_handle.cpp
......@@ -56,6 +56,7 @@
#include "cppa/exit_reason.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/message_future.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/scheduling_hint.hpp"
#include "cppa/event_based_actor.hpp"
......@@ -528,6 +529,27 @@ inline void reply(Args&&... what) {
self->reply_message(make_any_tuple(std::forward<Args>(what)...));
}
/**
* @brief Sends a message as reply to @p handle.
*/
template<typename... Args>
inline void reply_to(response_handle& handle, Args&&... what) {
if (handle.valid()) {
handle.apply(make_any_tuple(std::forward<Args>(what)...));
}
}
/**
* @brief Sends a message to the sender of the last received message.
* @param what Message content as a tuple.
*/
inline void reply_tuple_to(response_handle& handle, any_tuple what) {
handle.apply(std::move(what));
}
/**
* @brief Forwards the last received message to @p whom.
*/
inline void forward_to(const actor_ptr& whom) {
self->forward_message(whom);
}
......@@ -543,9 +565,7 @@ template<class Rep, class Period, typename... Args>
inline void delayed_send_tuple(const channel_ptr& whom,
const std::chrono::duration<Rep,Period>& rtime,
any_tuple what) {
if (whom) {
get_scheduler()->delayed_send(whom, rtime, what);
}
if (whom) get_scheduler()->delayed_send(whom, rtime, what);
}
/**
......
......@@ -39,6 +39,7 @@
#include "cppa/any_tuple.hpp"
#include "cppa/match_expr.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/partial_function.hpp"
#include "cppa/message_id.hpp"
......@@ -382,10 +383,14 @@ class local_actor : public actor {
void forward_message(const actor_ptr& new_receiver);
inline actor_ptr& chained_actor() {
inline const actor_ptr& chained_actor() {
return m_chained_actor;
}
inline void chained_actor(const actor_ptr& value) {
m_chained_actor = value;
}
inline bool awaits(message_id_t response_id) {
CPPA_REQUIRE(response_id.is_response());
return std::any_of(m_pending_responses.begin(),
......@@ -405,6 +410,12 @@ class local_actor : public actor {
virtual void become_waiting_for(behavior&&, message_id_t) = 0;
/**
* @brief Creates a {@link response_handle} to allow actors to response
* to a request later on.
*/
response_handle make_response_handle();
protected:
// true if this actor uses the chained_send optimization
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_RESPONSE_HANDLE_HPP
#define CPPA_RESPONSE_HANDLE_HPP
#include "cppa/actor.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/message_id.hpp"
namespace cppa {
/**
* @brief Identifies a
*/
class response_handle {
response_handle(const response_handle&) = delete;
response_handle& operator=(const response_handle&) = delete;
public:
response_handle() = default;
response_handle(response_handle&&) = default;
response_handle& operator=(response_handle&&) = default;
response_handle(const actor_ptr& from,
const actor_ptr& to,
const message_id_t& response_id);
/**
* @brief Queries whether response message is still outstanding.
*/
bool valid() const;
/**
* @brief Queries whether this is a response
* handle for a synchronous request.
*/
bool synchronous() const;
/**
* @brief Sends @p response_message and invalidates this handle afterwards.
*/
void apply(any_tuple response_message);
private:
actor_ptr m_from;
actor_ptr m_to;
message_id_t m_id;
};
} // namespace cppa
#endif // CPPA_RESPONSE_HANDLE_HPP
......@@ -150,4 +150,11 @@ sync_recv_helper local_actor::handle_response(const message_future& handle) {
}};
}
response_handle local_actor::make_response_handle() {
auto n = m_current_node;
response_handle result(this, n->sender, n->mid.response_id());
n->mid.mark_as_answered();
return std::move(result);
}
} // namespace cppa
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include <utility>
#include "cppa/self.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/response_handle.hpp"
using std::move;
namespace cppa {
response_handle::response_handle(const actor_ptr& from,
const actor_ptr& to,
const message_id_t& id)
: m_from(from), m_to(to), m_id(id) {
CPPA_REQUIRE(id.is_response());
}
bool response_handle::valid() const {
return m_to != nullptr;
}
bool response_handle::synchronous() const {
return m_id.valid();
}
void response_handle::apply(any_tuple msg) {
if (valid()) {
local_actor* sptr = self.unchecked();
if (sptr && sptr == m_from) {
if (sptr->chaining_enabled()) {
if (m_to->chained_sync_enqueue(sptr, m_id, move(msg))) {
sptr->chained_actor(m_to);
}
}
}
else m_to->enqueue(m_from.get(), move(msg));
m_to.reset();
}
}
} // namespace cppa
......@@ -114,7 +114,7 @@ struct thread_pool_scheduler::worker {
CPPA_REQUIRE(job != nullptr);
auto ptr = job->chained_actor().get();
if (ptr) {
job->chained_actor().reset();
job->chained_actor(nullptr);
return static_cast<scheduled_actor*>(ptr);
}
return nullptr;
......
......@@ -3,8 +3,20 @@
using namespace cppa;
struct popular_actor : event_based_actor { // popular actors have a buddy
actor_ptr m_buddy;
popular_actor(const actor_ptr& buddy) : m_buddy(buddy) { }
inline const actor_ptr& buddy() const { return m_buddy; }
};
void report_failure() {
local_actor* s = self;
send(static_cast<popular_actor*>(s)->buddy(), atom("failure"));
self->quit();
}
/******************************************************************************\
* test case: *
* test case 1: *
* *
* A B C *
* | | | *
......@@ -17,19 +29,14 @@ using namespace cppa;
* X X *
\******************************************************************************/
struct A : event_based_actor {
actor_ptr m_parent;
A(const actor_ptr& parent) : m_parent(parent) { }
struct A : popular_actor {
A(const actor_ptr& buddy) : popular_actor(buddy) { }
void init() {
auto report_failure = [=] {
send(m_parent, atom("failure"));
quit();
};
become (
on(atom("go"), arg_match) >> [=](const actor_ptr& next) {
sync_send(next, atom("gogo")).then (
on(atom("gogogo")) >> [=] {
send(m_parent, atom("success"));
send(buddy(), atom("success"));
quit();
},
others() >> report_failure,
......@@ -41,13 +48,12 @@ struct A : event_based_actor {
}
};
struct B : event_based_actor {
actor_ptr m_buddy;
B(const actor_ptr& buddy) : m_buddy(buddy) { }
struct B : popular_actor {
B(const actor_ptr& buddy) : popular_actor(buddy) { }
void init() {
become (
others() >> [=] {
forward_to(m_buddy);
forward_to(buddy());
quit();
}
);
......@@ -65,6 +71,41 @@ struct C : event_based_actor {
}
};
/******************************************************************************\
* test case 2: *
* *
* A D C *
* | | | *
* | --(sync_send)--> | | *
* | | --(sync_send)--> | *
* | | |---\ *
* | | | | *
* | | |<--/ *
* | | <---(reply)----- | *
* | <---(reply)----- | *
* X X *
\******************************************************************************/
struct D : popular_actor {
response_handle m_handle;
D(const actor_ptr& buddy) : popular_actor(buddy) { }
void init() {
become (
others() >> [=] {
m_handle = make_response_handle();
sync_send_tuple(buddy(), last_dequeued()).then(
others() >> [=] {
m_handle.apply(last_dequeued());
quit();
},
after(std::chrono::seconds(1)) >> report_failure
);
}
);
}
};
int main() {
CPPA_TEST(test__sync_send);
send(spawn<A>(self), atom("go"), spawn<B>(spawn<C>()));
......@@ -75,5 +116,14 @@ int main() {
}
);
await_all_others_done();
send(spawn<A>(self), atom("go"), spawn<D>(spawn<C>()));
receive (
on(atom("success")) >> [&] { },
on(atom("failure")) >> [&] {
CPPA_ERROR("A didn't receive a sync response");
}
);
await_all_others_done();
shutdown();
return CPPA_TEST_RESULT;
}
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