Commit c05472d2 authored by neverlord's avatar neverlord

refactored channel interface

parent e3526dee
...@@ -59,6 +59,13 @@ class actor : public channel ...@@ -59,6 +59,13 @@ class actor : public channel
~actor(); ~actor();
/**
* @brief Enqueues @p msg to the actor's mailbox and returns true if
* this actor is an scheduled actor that successfully changed
* its state to @p pending.
*/
virtual bool pending_enqueue(actor* sender, any_tuple msg);
/** /**
* @brief Attaches @p ptr to this actor. * @brief Attaches @p ptr to this actor.
* *
......
...@@ -54,9 +54,7 @@ class actor_proxy : public abstract_actor<actor> ...@@ -54,9 +54,7 @@ class actor_proxy : public abstract_actor<actor>
actor_proxy(std::uint32_t mid, process_information_ptr const& parent); actor_proxy(std::uint32_t mid, process_information_ptr const& parent);
void enqueue(actor* sender, any_tuple&& msg); void enqueue(actor* sender, any_tuple msg);
void enqueue(actor* sender, any_tuple const& msg);
void link_to(intrusive_ptr<actor>& other); void link_to(intrusive_ptr<actor>& other);
...@@ -76,8 +74,7 @@ class actor_proxy : public abstract_actor<actor> ...@@ -76,8 +74,7 @@ class actor_proxy : public abstract_actor<actor>
public: public:
void forward_message(process_information_ptr const&, void forward_message(process_information_ptr const&, actor*, any_tuple&&);
actor*, any_tuple const&);
}; };
......
...@@ -60,9 +60,7 @@ class channel : public ref_counted ...@@ -60,9 +60,7 @@ class channel : public ref_counted
/** /**
* @brief Enqueues @p msg to the list of received messages. * @brief Enqueues @p msg to the list of received messages.
*/ */
virtual void enqueue(actor* sender, any_tuple const& msg) = 0; virtual void enqueue(actor* sender, any_tuple msg) = 0;
virtual void enqueue(actor* sender, any_tuple&& msg) = 0;
private: private:
......
...@@ -139,16 +139,11 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> ...@@ -139,16 +139,11 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor>
throw actor_exited(reason); throw actor_exited(reason);
} }
void enqueue(actor* sender, any_tuple&& msg) void enqueue(actor* sender, any_tuple msg)
{ {
enqueue_node(super::fetch_node(sender, std::move(msg))); enqueue_node(super::fetch_node(sender, std::move(msg)));
} }
void enqueue(actor* sender, any_tuple const& msg)
{
enqueue_node(super::fetch_node(sender, msg));
}
int compare_exchange_state(int expected, int new_value) int compare_exchange_state(int expected, int new_value)
{ {
int e = expected; int e = expected;
......
...@@ -38,8 +38,6 @@ ...@@ -38,8 +38,6 @@
#include "cppa/ref_counted.hpp" #include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp" #include "cppa/intrusive_ptr.hpp"
#include "cppa/detail/channel.hpp"
namespace cppa { namespace detail { namespace cppa { namespace detail {
class addressed_message class addressed_message
......
...@@ -100,9 +100,7 @@ class converted_thread_context : public abstract_actor<local_actor> ...@@ -100,9 +100,7 @@ class converted_thread_context : public abstract_actor<local_actor>
void quit(std::uint32_t reason); //override void quit(std::uint32_t reason); //override
void enqueue(actor* sender, any_tuple&& msg); //override void enqueue(actor* sender, any_tuple msg); //override
void enqueue(actor* sender, any_tuple const& msg); //override
void dequeue(behavior& rules); //override void dequeue(behavior& rules); //override
......
...@@ -63,6 +63,12 @@ actor::actor(std::uint32_t aid, const process_information_ptr& pptr) ...@@ -63,6 +63,12 @@ actor::actor(std::uint32_t aid, const process_information_ptr& pptr)
} }
} }
bool actor::pending_enqueue(actor* sender, any_tuple msg)
{
enqueue(sender, std::move(msg));
return false;
}
actor::actor(const process_information_ptr& pptr) actor::actor(const process_information_ptr& pptr)
: m_id(registry().next_id()), m_is_proxy(false), m_parent_process(pptr) : m_id(registry().next_id()), m_is_proxy(false), m_parent_process(pptr)
{ {
......
...@@ -54,19 +54,13 @@ actor_proxy::actor_proxy(std::uint32_t mid, const process_information_ptr& pptr) ...@@ -54,19 +54,13 @@ actor_proxy::actor_proxy(std::uint32_t mid, const process_information_ptr& pptr)
void actor_proxy::forward_message(const process_information_ptr& piptr, void actor_proxy::forward_message(const process_information_ptr& piptr,
actor* sender, actor* sender,
const any_tuple& msg) any_tuple&& msg)
{ {
auto mailman_msg = new detail::mailman_job(piptr, sender, this, msg); auto mailman_msg = new detail::mailman_job(piptr, sender, this, std::move(msg));
detail::mailman_queue().push_back(mailman_msg); detail::mailman_queue().push_back(mailman_msg);
} }
void actor_proxy::enqueue(actor* sender, any_tuple&& msg) void actor_proxy::enqueue(actor* sender, any_tuple msg)
{
any_tuple tmp(std::move(msg));
enqueue(sender, tmp);
}
void actor_proxy::enqueue(actor* sender, const any_tuple& msg)
{ {
if ( msg.size() == 2 if ( msg.size() == 2
&& *(msg.type_at(0)) == typeid(atom_value) && *(msg.type_at(0)) == typeid(atom_value)
...@@ -76,7 +70,7 @@ void actor_proxy::enqueue(actor* sender, const any_tuple& msg) ...@@ -76,7 +70,7 @@ void actor_proxy::enqueue(actor* sender, const any_tuple& msg)
cleanup(msg.get_as<std::uint32_t>(1)); cleanup(msg.get_as<std::uint32_t>(1));
return; return;
} }
forward_message(parent_process_ptr(), sender, msg); forward_message(parent_process_ptr(), sender, std::move(msg));
} }
void actor_proxy::link_to(intrusive_ptr<actor>& other) void actor_proxy::link_to(intrusive_ptr<actor>& other)
......
...@@ -59,16 +59,11 @@ void converted_thread_context::cleanup(std::uint32_t reason) ...@@ -59,16 +59,11 @@ void converted_thread_context::cleanup(std::uint32_t reason)
super::cleanup(reason); super::cleanup(reason);
} }
void converted_thread_context::enqueue(actor* sender, any_tuple&& msg) void converted_thread_context::enqueue(actor* sender, any_tuple msg)
{ {
m_mailbox.push_back(fetch_node(sender, std::move(msg))); m_mailbox.push_back(fetch_node(sender, std::move(msg)));
} }
void converted_thread_context::enqueue(actor* sender, const any_tuple& msg)
{
m_mailbox.push_back(fetch_node(sender, msg));
}
void converted_thread_context::dequeue(partial_function& fun) // override void converted_thread_context::dequeue(partial_function& fun) // override
{ {
if (m_invoke.invoke_from_cache(fun) == false) if (m_invoke.invoke_from_cache(fun) == false)
......
...@@ -60,7 +60,7 @@ class group_impl : public group ...@@ -60,7 +60,7 @@ class group_impl : public group
public: public:
void enqueue(actor* sender, const any_tuple& msg) /*override*/ void enqueue(actor* sender, any_tuple msg) /*override*/
{ {
shared_guard guard(m_shared_mtx); shared_guard guard(m_shared_mtx);
for (auto i = m_subscribers.begin(); i != m_subscribers.end(); ++i) for (auto i = m_subscribers.begin(); i != m_subscribers.end(); ++i)
...@@ -71,12 +71,6 @@ class group_impl : public group ...@@ -71,12 +71,6 @@ class group_impl : public group
} }
} }
void enqueue(actor* sender, any_tuple&& msg) /*override*/
{
any_tuple tmp(std::move(msg));
enqueue(sender, tmp);
}
group::subscription subscribe(const channel_ptr& who) /*override*/ group::subscription subscribe(const channel_ptr& who) /*override*/
{ {
group::subscription result; group::subscription 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