Commit bd235b6b authored by neverlord's avatar neverlord

added 'then' & 'await' member functions to message_future

parent 0c5a0c34
......@@ -262,3 +262,4 @@ src/ipv4_acceptor.cpp
cppa/detail/middleman.hpp
src/middleman.cpp
src/buffer.cpp
cppa/message_future.hpp
......@@ -54,6 +54,7 @@
#include "cppa/tuple_cast.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/message_future.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/scheduling_hint.hpp"
#include "cppa/event_based_actor.hpp"
......@@ -466,10 +467,6 @@ operator<<(const intrusive_ptr<C>& whom, any_tuple what) {
return whom;
}
#ifndef CPPA_DOCUMENTATION
typedef message_id_t message_future;
#endif // CPPA_DOCUMENTATION
/**
* @brief Sends @p what as a synchronous message to @p whom.
* @param whom Receiver of the message.
......@@ -508,8 +505,8 @@ inline message_future sync_send(const actor_ptr& whom, Args&&... what) {
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
*/
inline sync_recv_helper receive_response(message_future handle) {
return {handle, [](behavior& bhvr, message_future mf) {
inline sync_recv_helper receive_response(const message_future& handle) {
return {handle.id(), [](behavior& bhvr, message_id_t mf) {
if (!self->awaits(mf)) {
throw std::logic_error("response already received");
}
......
......@@ -49,7 +49,7 @@ struct scheduled_actor_dummy : abstract_scheduled_actor {
bool attach(attachable*);
void unbecome();
void do_become(behavior&&, bool);
void become_waiting_for(behavior&&, message_future);
void become_waiting_for(behavior&&, message_id_t);
bool has_behavior();
scheduled_actor_type impl_type();
};
......
......@@ -88,7 +88,7 @@ class stacked_actor_mixin : public Base {
become_impl(std::move(bhvr), discard_old, message_id_t());
}
virtual void become_waiting_for(behavior&& bhvr, message_future mid) {
virtual void become_waiting_for(behavior&& bhvr, message_id_t mid) {
become_impl(std::move(bhvr), false, mid);
}
......
......@@ -134,7 +134,7 @@ class event_based_actor : public detail::abstract_scheduled_actor {
void do_become(behavior&& bhvr, bool discard_old);
void become_waiting_for(behavior&& bhvr, message_future mf);
void become_waiting_for(behavior&& bhvr, message_id_t mf);
private:
......
......@@ -46,12 +46,9 @@
namespace cppa {
#ifndef CPPA_DOCUMENTATION
typedef message_id_t message_future;
#endif // CPPA_DOCUMENTATION
// forward declarations
class scheduler;
class message_future;
class local_scheduler;
template<bool DiscardOld>
......@@ -84,10 +81,10 @@ constexpr keep_behavior_t keep_behavior = keep_behavior_t();
#endif // CPPA_DOCUMENTATION
struct sync_recv_helper {
typedef void (*callback_type)(behavior&, message_future);
message_future m_handle;
typedef void (*callback_type)(behavior&, message_id_t);
message_id_t m_handle;
callback_type m_fun;
inline sync_recv_helper(message_future handle, callback_type fun)
inline sync_recv_helper(message_id_t handle, callback_type fun)
: m_handle(handle), m_fun(fun) { }
template<typename... Expression>
inline void operator()(Expression&&... mexpr) const {
......@@ -104,7 +101,9 @@ struct sync_recv_helper {
}
};
inline sync_recv_helper receive_response(message_future);
class message_future;
//inline sync_recv_helper receive_response(message_future);
/**
* @brief Base class for local running Actors.
......@@ -112,7 +111,7 @@ inline sync_recv_helper receive_response(message_future);
class local_actor : public actor {
friend class scheduler;
friend inline sync_recv_helper receive_response(message_future);
//friend inline sync_recv_helper receive_response(message_future);
public:
......@@ -303,14 +302,7 @@ class local_actor : public actor {
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
*/
inline sync_recv_helper handle_response(message_future handle) {
return {handle, [](behavior& bhvr, message_future mf) {
if (!self->awaits(mf)) {
throw std::logic_error("response already received");
}
self->become_waiting_for(std::move(bhvr), mf);
}};
}
sync_recv_helper handle_response(const message_future& handle);
/**
* @brief Returns to a previous behavior if available.
......@@ -409,6 +401,10 @@ class local_actor : public actor {
if (i != last) m_pending_responses.erase(i);
}
virtual void dequeue_response(behavior&, message_id_t) = 0;
virtual void become_waiting_for(behavior&&, message_id_t) = 0;
protected:
// true if this actor uses the chained_send optimization
......@@ -452,9 +448,6 @@ class local_actor : public actor {
* Use always the {@link cppa::receive_response receive_response}
* function.
*/
virtual void dequeue_response(behavior&, message_future) = 0;
virtual void become_waiting_for(behavior&&, message_future) = 0;
};
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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 MESSAGE_FUTURE_HPP
#define MESSAGE_FUTURE_HPP
#include "cppa/behavior.hpp"
#include "cppa/match_expr.hpp"
#include "cppa/message_id.hpp"
#include "cppa/local_actor.hpp"
namespace cppa {
/**
* @brief Represents the result of a synchronous send.
*/
class message_future {
public:
/**
* @brief Sets @p mexpr as event-handler for the response message.
*/
template<typename... Expression>
void then(Expression&&... mexpr) {
auto f = [](behavior& bhvr, message_id_t mid) {
self->become_waiting_for(std::move(bhvr), mid);
};
apply(f, std::forward<Expression>(mexpr)...);
}
/**
* @brief Blocks until the response arrives and then executes @p mexpr.
*/
template<typename... Expression>
void await(Expression&&... mexpr) {
auto f = [](behavior& bhvr, message_id_t mid) {
self->dequeue_response(bhvr, mid);
};
apply(f, std::forward<Expression>(mexpr)...);
}
message_future(const message_future&) = default;
message_future& operator=(const message_future&) = default;
# ifndef CPPA_DOCUMENTATION
inline message_future(const message_id_t& from) : m_id(from) { }
inline const message_id_t& id() const { return m_id; }
# endif
private:
message_id_t m_id;
template<typename Fun, typename... Args>
void apply(Fun& fun, Args&&... args) {
auto bhvr = match_expr_convert(std::forward<Args>(args)...);
static_assert(std::is_same<decltype(bhvr), behavior>::value,
"no timeout specified");
if (bhvr.timeout().valid() == false || bhvr.timeout().is_zero()) {
throw std::invalid_argument("specified timeout is invalid or zero");
}
else if (!m_id.valid() || !m_id.is_response()) {
throw std::logic_error("handle does not point to a response");
}
else if (!self->awaits(m_id)) {
throw std::logic_error("response already received");
}
fun(bhvr, m_id);
}
};
} // namespace cppa
#endif // MESSAGE_FUTURE_HPP
......@@ -103,7 +103,7 @@ void event_based_actor::do_become(behavior&& bhvr, bool discard_old) {
m_bhvr_stack.push_back(std::move(bhvr));
}
void event_based_actor::become_waiting_for(behavior&& bhvr, message_future mf) {
void event_based_actor::become_waiting_for(behavior&& bhvr, message_id_t mf) {
CPPA_REQUIRE(bhvr.timeout().valid());
reset_timeout();
request_timeout(bhvr.timeout());
......
......@@ -140,4 +140,13 @@ void local_actor::forward_message(const actor_ptr& new_receiver) {
}
}
sync_recv_helper local_actor::handle_response(const message_future& handle) {
return {handle.id(), [](behavior& bhvr, message_id_t mf) {
if (!self->awaits(mf)) {
throw std::logic_error("response already received");
}
self->become_waiting_for(std::move(bhvr), mf);
}};
}
} // namespace cppa
......@@ -42,7 +42,7 @@ void scheduled_actor_dummy::detach(const attachable::token&) { }
bool scheduled_actor_dummy::attach(attachable*) { return false; }
void scheduled_actor_dummy::unbecome() { }
void scheduled_actor_dummy::do_become(behavior&&, bool) { }
void scheduled_actor_dummy::become_waiting_for(behavior&&, message_future) { }
void scheduled_actor_dummy::become_waiting_for(behavior&&, message_id_t) { }
bool scheduled_actor_dummy::has_behavior() { return false; }
resume_result scheduled_actor_dummy::resume(util::fiber*) {
......
......@@ -27,7 +27,7 @@ struct A : event_based_actor {
};
become (
on(atom("go"), arg_match) >> [=](const actor_ptr& next) {
handle_response(sync_send(next, atom("gogo"))) (
sync_send(next, atom("gogo")).then (
on(atom("gogogo")) >> [=] {
send(m_parent, atom("success"));
quit();
......
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