Commit 6c2e5c5d authored by neverlord's avatar neverlord

future_send and delayed_reply

parent 77b1e3d0
...@@ -297,6 +297,28 @@ void reply(const Arg0& arg0, const Args&... args) ...@@ -297,6 +297,28 @@ void reply(const Arg0& arg0, const Args&... args)
if (whom) whom->enqueue(message(sptr, whom, arg0, args...)); if (whom) whom->enqueue(message(sptr, whom, arg0, args...));
} }
/**
* @brief Send a message that is delayed by @p rel_time.
*/
template<typename Duration, typename... Data>
void future_send(actor_ptr whom, const Duration& rel_time, const Data&... data)
{
get_scheduler()->future_send(self(), whom, rel_time, data...);
}
/**
*
*/
template<typename Duration, typename... Data>
void delayed_reply(const Duration& rel_time, const Data&... data)
{
auto whom = last_received().sender();
if (whom)
{
get_scheduler()->future_send(self(), whom, rel_time, data...);
}
}
/** /**
* @brief Blocks execution of this actor until all * @brief Blocks execution of this actor until all
* other actors finished execution. * other actors finished execution.
......
...@@ -86,11 +86,12 @@ class scheduler ...@@ -86,11 +86,12 @@ class scheduler
virtual void await_others_done(); virtual void await_others_done();
template<typename Duration, typename... Data> template<typename Duration, typename... Data>
void future_send(actor_ptr whom, const Duration& d, const Data&... data) void future_send(actor_ptr from, actor_ptr to,
const Duration& rel_time, const Data&... data)
{ {
static_assert(sizeof...(Data) > 0, "no message to send"); static_assert(sizeof...(Data) > 0, "no message to send");
any_tuple tup = make_tuple(util::duration(d), data...); any_tuple tup = make_tuple(util::duration(rel_time), data...);
future_send_helper()->enqueue(message(whom, whom, tup)); future_send_helper()->enqueue(message(from, to, tup));
} }
}; };
......
...@@ -4,6 +4,7 @@ ...@@ -4,6 +4,7 @@
#include <atomic> #include <atomic>
#include <iostream> #include <iostream>
#include "cppa/cppa.hpp"
#include "cppa/match.hpp" #include "cppa/match.hpp"
#include "cppa/context.hpp" #include "cppa/context.hpp"
#include "cppa/scheduler.hpp" #include "cppa/scheduler.hpp"
...@@ -180,7 +181,7 @@ bool yielding_message_queue_impl::dequeue_impl(timed_invoke_rules& rules, queue_ ...@@ -180,7 +181,7 @@ bool yielding_message_queue_impl::dequeue_impl(timed_invoke_rules& rules, queue_
{ {
if (m_queue.empty() && !m_has_pending_timeout_request) if (m_queue.empty() && !m_has_pending_timeout_request)
{ {
get_scheduler()->future_send(self(), rules.timeout(), future_send(self(), rules.timeout(),
atom(":Timeout"), ++m_active_timeout_id); atom(":Timeout"), ++m_active_timeout_id);
m_has_pending_timeout_request = true; m_has_pending_timeout_request = true;
} }
......
...@@ -50,6 +50,24 @@ void testee2(actor_ptr other) ...@@ -50,6 +50,24 @@ void testee2(actor_ptr other)
); );
} }
void testee3(actor_ptr parent)
{
// test a future_send / delayed_reply based loop
future_send(self(), std::chrono::milliseconds(50), atom("Poll"));
int polls = 0;
receive_while([&polls]() { return ++polls <= 5; })
(
on(atom("Poll")) >> [&]()
{
if (polls < 5)
{
delayed_reply(std::chrono::milliseconds(50), atom("Poll"));
}
send(parent, atom("Push"), polls);
}
);
}
size_t test__spawn() size_t test__spawn()
{ {
CPPA_TEST(test__spawn); CPPA_TEST(test__spawn);
...@@ -65,8 +83,7 @@ size_t test__spawn() ...@@ -65,8 +83,7 @@ size_t test__spawn()
monitor(spawn(testee2, spawn(testee1))); monitor(spawn(testee2, spawn(testee1)));
int i = 0; int i = 0;
int flags = 0; int flags = 0;
get_scheduler()->future_send(self(), std::chrono::seconds(1), future_send(self(), std::chrono::seconds(1), atom("FooBar"));
atom("FooBar"));
// wait for :Down and :Exit messages of pong // wait for :Down and :Exit messages of pong
receive_while([&i]() { return ++i <= 4; }) receive_while([&i]() { return ++i <= 4; })
( (
...@@ -115,5 +132,25 @@ size_t test__spawn() ...@@ -115,5 +132,25 @@ size_t test__spawn()
} }
// verify pong messages // verify pong messages
CPPA_CHECK_EQUAL(pongs(), 5); CPPA_CHECK_EQUAL(pongs(), 5);
spawn(testee3, self());
i = 0;
// testee3 sends 5 { "Push", int } messages in a 50 milliseconds interval;
// allow for a maximum error of 5ms
receive_while([&i]() { return ++i <= 5; })
(
on<atom("Push"), int>() >> [&](int val)
{
CPPA_CHECK_EQUAL(i, val);
//cout << "{ Push, " << val << " } ..." << endl;
},
after(std::chrono::milliseconds(55)) >> [&]()
{
cout << "Push " << i
<< " was delayed more than 55 milliseconds" << endl;
CPPA_CHECK(false);
}
);
spawn(dancing_kirby);
await_all_others_done();
return CPPA_TEST_RESULT; 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