Commit c316456d authored by Dominik Charousset's avatar Dominik Charousset

renamed `response_future` => `response_handle`

parent 9c9c4285
......@@ -207,7 +207,6 @@ set(LIBCPPA_SRC
src/unicast_network.cpp
src/uniform_type_info.cpp
src/uniform_type_info_map.cpp
src/untyped_actor_handle.cpp
src/weak_ptr_anchor.cpp
src/yield_interface.cpp)
......
......@@ -55,7 +55,7 @@ cppa/detail/proper_actor.hpp
cppa/detail/pseudo_tuple.hpp
cppa/detail/ptype_to_type.hpp
cppa/detail/raw_access.hpp
cppa/detail/response_future_util.hpp
cppa/detail/response_handle_util.hpp
cppa/detail/serialize_tuple.hpp
cppa/detail/singleton_manager.hpp
cppa/detail/singleton_mixin.hpp
......@@ -175,7 +175,6 @@ cppa/typed_actor_ptr.hpp
cppa/typed_behavior.hpp
cppa/uniform_type_info.hpp
cppa/unit.hpp
cppa/untyped_actor_handle.hpp
cppa/util/abstract_uniform_type_info.hpp
cppa/util/algorithm.hpp
cppa/util/arg_match_t.hpp
......@@ -315,7 +314,6 @@ src/type_lookup_table.cpp
src/unicast_network.cpp
src/uniform_type_info.cpp
src/uniform_type_info_map.cpp
src/untyped_actor_handle.cpp
src/weak_ptr_anchor.cpp
src/yield_interface.cpp
unit_testing/ping_pong.cpp
......
......@@ -56,7 +56,7 @@ class deserializer;
/**
* @brief A unique actor ID.
* @relates actor
* @relates abstract_actor
*/
typedef std::uint32_t actor_id;
......@@ -85,6 +85,20 @@ class abstract_actor : public abstract_channel {
*/
bool attach(attachable_ptr ptr);
/**
* @brief Convenience function that attaches the functor
* or function @p f to this actor.
*
* The actor executes <tt>f()</tt> on exit, or immediatley
* if it already finished execution.
* @param f A functor, function or lambda expression that's executed
* if the actor finishes execution.
* @returns @c true if @p f was successfully attached to the actor;
* otherwise (actor already exited) @p false.
*/
template<typename F>
bool attach_functor(F&& f);
/**
* @brief Returns the address of this actor.
*/
......@@ -96,21 +110,28 @@ class abstract_actor : public abstract_channel {
void detach(const attachable::token& what);
/**
* @brief Links this actor to @p other.
* @brief Links this actor to @p whom.
* @param other Actor instance that whose execution is coupled to the
* execution of this Actor.
*/
virtual void link_to(const actor_addr& other);
virtual void link_to(const actor_addr& whom);
/**
* @copydoc abstract_actor::link_to(const actor_addr&)
*/
void link_to(const actor& whom);
/**
* @brief Unlinks this actor from @p other.
* @brief Unlinks this actor from @p whom.
* @param other Linked Actor.
* @note Links are automatically removed if the actor finishes execution.
*/
virtual void unlink_from(const actor_addr& other);
virtual void unlink_from(const actor_addr& whom);
/**
* @copydoc abstract_actor::unlink_from(const actor_addr&)
*/
void unlink_from(const actor& whom);
/**
* @brief Establishes a link relation between this actor and @p other.
......@@ -232,6 +253,27 @@ inline const node_id& abstract_actor::node() const {
return *m_node;
}
template<class F>
struct functor_attachable : attachable {
F m_functor;
template<typename T>
inline functor_attachable(T&& arg) : m_functor(std::forward<T>(arg)) { }
void actor_exited(std::uint32_t reason) { m_functor(reason); }
bool matches(const attachable::token&) { return false; }
};
template<typename F>
bool abstract_actor::attach_functor(F&& f) {
typedef typename util::rm_const_and_ref<F>::type f_type;
typedef functor_attachable<f_type> impl;
return attach(attachable_ptr{new impl(std::forward<F>(f))});
}
} // namespace cppa
#endif // CPPA_ABSTRACT_ACTOR_HPP
......@@ -38,7 +38,6 @@
#include "cppa/intrusive_ptr.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/untyped_actor_handle.hpp"
#include "cppa/util/comparable.hpp"
#include "cppa/util/type_traits.hpp"
......@@ -68,6 +67,14 @@ struct invalid_actor_t { constexpr invalid_actor_t() { } };
*/
constexpr invalid_actor_t invalid_actor = invalid_actor_t{};
template<typename T>
struct is_convertible_to_actor {
static constexpr bool value = std::is_base_of<io::broker, T>::value
|| std::is_base_of<actor_proxy, T>::value
|| std::is_base_of<blocking_actor, T>::value
|| std::is_base_of<event_based_actor, T>::value;
};
/**
* @brief Identifies an untyped actor.
*
......@@ -84,34 +91,6 @@ class actor : util::comparable<actor>
public:
/**
* @brief Extends {@link untyped_actor_handle} to grant
* access to the enqueue member function.
*/
class handle : public untyped_actor_handle {
// untyped_actor_handle does not provide a virtual destructor
// -> no new members
friend class actor;
typedef untyped_actor_handle super;
public:
handle() = default;
/**
* @brief Enqueues @p msg to the receiver specified in @p hdr.
*/
void enqueue(const message_header& hdr, any_tuple msg) const;
private:
inline handle(abstract_actor_ptr ptr) : super(std::move(ptr)) { }
};
actor() = default;
actor(actor&&) = default;
......@@ -120,23 +99,13 @@ class actor : util::comparable<actor>
template<typename T>
actor(intrusive_ptr<T> ptr,
typename std::enable_if<
std::is_base_of<io::broker, T>::value
|| std::is_base_of<actor_proxy, T>::value
|| std::is_base_of<blocking_actor, T>::value
|| std::is_base_of<event_based_actor, T>::value
>::type* = 0)
: m_ops(std::move(ptr)) { }
typename std::enable_if<is_convertible_to_actor<T>::value>::type* = 0)
: m_ptr(std::move(ptr)) { }
template<typename T>
actor(T* ptr,
typename std::enable_if<
std::is_base_of<io::broker, T>::value
|| std::is_base_of<actor_proxy, T>::value
|| std::is_base_of<event_based_actor, T>::value
|| std::is_base_of<blocking_actor, T>::value
>::type* = 0)
: m_ops(ptr) { }
typename std::enable_if<is_convertible_to_actor<T>::value>::type* = 0)
: m_ptr(ptr) { }
actor(const invalid_actor_t&);
......@@ -145,13 +114,7 @@ class actor : util::comparable<actor>
actor& operator=(const actor&) = default;
template<typename T>
typename std::enable_if<
std::is_base_of<io::broker, T>::value
|| std::is_base_of<actor_proxy, T>::value
|| std::is_base_of<event_based_actor, T>::value
|| std::is_base_of<blocking_actor, T>::value,
actor&
>::type
typename std::enable_if<is_convertible_to_actor<T>::value, actor&>::type
operator=(intrusive_ptr<T> ptr) {
actor tmp{std::move(ptr)};
swap(tmp);
......@@ -159,13 +122,7 @@ class actor : util::comparable<actor>
}
template<typename T>
typename std::enable_if<
std::is_base_of<io::broker, T>::value
|| std::is_base_of<actor_proxy, T>::value
|| std::is_base_of<event_based_actor, T>::value
|| std::is_base_of<blocking_actor, T>::value,
actor&
>::type
typename std::enable_if<is_convertible_to_actor<T>::value, actor&>::type
operator=(T* ptr) {
actor tmp{ptr};
swap(tmp);
......@@ -175,23 +132,23 @@ class actor : util::comparable<actor>
actor& operator=(const invalid_actor_t&);
inline operator bool() const {
return static_cast<bool>(m_ops.m_ptr);
return static_cast<bool>(m_ptr);
}
inline bool operator!() const {
return !m_ops.m_ptr;
return !m_ptr;
}
/**
* @brief Returns a handle that grants access to
* actor operations such as enqueue.
*/
inline const handle* operator->() const {
return &m_ops;
inline abstract_actor* operator->() const {
return m_ptr.get();
}
inline const handle& operator*() const {
return m_ops;
inline abstract_actor& operator*() const {
return *m_ptr;
}
intptr_t compare(const actor& other) const;
......@@ -199,20 +156,25 @@ class actor : util::comparable<actor>
intptr_t compare(const actor_addr&) const;
inline intptr_t compare(const invalid_actor_t&) const {
return m_ops.m_ptr ? 1 : 0;
return m_ptr ? 1 : 0;
}
inline intptr_t compare(const invalid_actor_addr_t&) const {
return compare(invalid_actor);
}
/**
* @brief Queries the address of the stored actor.
*/
actor_addr address() const;
private:
void swap(actor& other);
actor(abstract_actor*);
handle m_ops;
abstract_actor_ptr m_ptr;
};
......
......@@ -37,12 +37,12 @@
#include "cppa/intrusive_ptr.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/untyped_actor_handle.hpp"
#include "cppa/util/comparable.hpp"
namespace cppa {
class actor;
class local_actor;
class actor_namespace;
......@@ -63,6 +63,7 @@ class actor_addr : util::comparable<actor_addr>
, util::comparable<actor_addr, abstract_actor*>
, util::comparable<actor_addr, abstract_actor_ptr> {
friend class actor;
friend class abstract_actor;
friend class detail::raw_access;
......@@ -83,15 +84,11 @@ class actor_addr : util::comparable<actor_addr>
actor_addr operator=(const invalid_actor_addr_t&);
inline explicit operator bool() const {
return valid();
return static_cast<bool>(m_ptr);
}
inline bool operator!() const {
return !valid();
}
inline bool valid() const {
return m_ops.valid();
return !m_ptr;
}
intptr_t compare(const actor_addr& other) const;
......@@ -102,19 +99,21 @@ class actor_addr : util::comparable<actor_addr>
return compare(other.get());
}
inline const untyped_actor_handle* operator->() const {
return &m_ops;
}
actor_id id() const;
inline const untyped_actor_handle& operator*() const {
return m_ops;
}
const node_id& node() const;
/**
* @brief Returns whether this is an address of a
* remote actor.
*/
bool is_remote() const;
private:
explicit actor_addr(abstract_actor*);
untyped_actor_handle m_ops;
abstract_actor_ptr m_ptr;
};
......
......@@ -44,7 +44,7 @@ class actor_proxy_cache;
/**
* @brief Represents a remote actor.
* @extends actor
* @extends abstract_actor
*/
class actor_proxy : public extend<abstract_actor>::with<enable_weak_ptr> {
......
......@@ -38,7 +38,7 @@
#include "cppa/mailbox_based.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/detail/response_future_util.hpp"
#include "cppa/detail/response_handle_util.hpp"
namespace cppa {
......@@ -49,11 +49,11 @@ class blocking_actor : public extend<local_actor>::with<mailbox_based> {
public:
class response_future {
class response_handle {
public:
response_future() = delete;
response_handle() = delete;
void await(behavior&);
......@@ -86,10 +86,10 @@ class blocking_actor : public extend<local_actor>::with<mailbox_based> {
*/
inline const message_id& id() const { return m_mid; }
response_future(const response_future&) = default;
response_future& operator=(const response_future&) = default;
response_handle(const response_handle&) = default;
response_handle& operator=(const response_handle&) = default;
inline response_future(const message_id& from, blocking_actor* self)
inline response_handle(const message_id& from, blocking_actor* self)
: m_mid(from), m_self(self) { }
private:
......@@ -103,7 +103,7 @@ class blocking_actor : public extend<local_actor>::with<mailbox_based> {
public:
inline sync_receive_helper(const response_future& mf) : m_mf(mf) { }
inline sync_receive_helper(const response_handle& mf) : m_mf(mf) { }
template<typename... Ts>
inline void operator()(Ts&&... args) {
......@@ -112,7 +112,7 @@ class blocking_actor : public extend<local_actor>::with<mailbox_based> {
private:
response_future m_mf;
response_handle m_mf;
};
......@@ -125,9 +125,9 @@ class blocking_actor : public extend<local_actor>::with<mailbox_based> {
* sent message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
response_future sync_send_tuple(const actor& dest, any_tuple what);
response_handle sync_send_tuple(const actor& dest, any_tuple what);
response_future timed_sync_send_tuple(const util::duration& rtime,
response_handle timed_sync_send_tuple(const util::duration& rtime,
const actor& dest,
any_tuple what);
......@@ -142,13 +142,13 @@ class blocking_actor : public extend<local_actor>::with<mailbox_based> {
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<typename... Ts>
inline response_future sync_send(const actor& dest, Ts&&... what) {
inline response_handle sync_send(const actor& dest, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(dest, make_any_tuple(std::forward<Ts>(what)...));
}
template<typename... Ts>
response_future timed_sync_send(const actor& dest,
response_handle timed_sync_send(const actor& dest,
const util::duration& rtime,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
......@@ -273,9 +273,9 @@ class blocking_actor : public extend<local_actor>::with<mailbox_based> {
* @param handle A future for a synchronous response.
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
* @relates response_future
* @relates response_handle
*/
inline sync_receive_helper receive_response(const response_future& f) {
inline sync_receive_helper receive_response(const response_handle& f) {
return {f};
}
......
......@@ -433,53 +433,61 @@
namespace cppa {
/**
* @brief Sends @p to a message under the identity of @p from.
*/
inline void send_tuple_as(const actor& from, const channel& to, any_tuple msg) {
to.enqueue({from->address(), to}, std::move(msg));
to.enqueue({from.address(), to}, std::move(msg));
}
/**
* @brief Sends @p to a message under the identity of @p from.
*/
template<typename... Ts>
void send_as(const actor& from, channel& to, Ts&&... args) {
void send_as(const actor& from, const channel& to, Ts&&... args) {
send_tuple_as(from, to, make_any_tuple(std::forward<Ts>(args)...));
}
/**
* @brief Anonymously sends @p to a message.
*/
inline void anon_send_tuple(const channel& to, any_tuple msg) {
send_tuple_as(invalid_actor, to, std::move(msg));
}
/**
* @brief Anonymously sends @p to a message.
*/
template<typename... Ts>
void send_as(const actor& from, const channel& to, Ts&&... args) {
send_tuple_as(from, to, make_any_tuple(std::forward<Ts>(args)...));
inline void anon_send(const channel& to, Ts&&... args) {
send_as(invalid_actor, to, std::forward<Ts>(args)...);
}
/**
* @brief Sets the maximum size of a message.
* @brief Sets the maximum size of a message over network.
* @param size The maximum number of bytes a message may occupy.
*/
void max_msg_size(size_t size);
/**
* @brief Retrieves the maximum size of messages.
* @brief Queries the maximum size of messages over network.
* @returns The number maximum number of bytes a message may occupy.
*/
size_t max_msg_size();
inline void anon_send_tuple(const channel& to, any_tuple msg) {
to.enqueue({invalid_actor_addr, to}, std::move(msg));
}
// implemented in local_actor.cpp
/**
* @brief Anonymously sends a message to @p receiver;
* @brief Anonymously sends @p whom an exit message.
*/
template<typename... Ts>
inline void anon_send(const channel& receiver, Ts&&... args) {
anon_send_tuple(receiver, make_any_tuple(std::forward<Ts>(args)...));
}
// implemented in local_actor.cpp
void anon_send_exit(const actor_addr& whom, std::uint32_t reason);
/**
* @brief Anonymously sends @p whom an exit message.
*/
inline void anon_send_exit(const actor& whom, std::uint32_t reason) {
whom->enqueue({invalid_actor_addr, whom, message_id{}.with_high_priority()},
make_any_tuple(exit_msg{invalid_actor_addr, reason}));
anon_send_exit(whom.address(), reason);
}
/**
* @brief Blocks execution of this actor until all
* other actors finished execution.
......@@ -528,7 +536,7 @@ inline actor remote_actor(const std::string& host, std::uint16_t port) {
}
/**
* @brief Establish a new connection to the actor via given @p connection.
* @brief Establish a new connection to a remote actor via @p connection.
* @param connection A connection to another libcppa process described by a pair
* of input and output stream.
* @returns An {@link actor_ptr} to the proxy instance
......@@ -550,10 +558,13 @@ actor spawn_io(Ts&&... args) {
}
/**
* @brief Spawns a new {@link actor} that evaluates given arguments.
* @param args A functor followed by its arguments.
* @brief Spawns a new, function-based IO actor.
* @param fun A functor implementing the actor's behavior.
* @param in The actor's input stream.
* @param out The actor's output stream.
* @param args Optional arguments for @p fun.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
* @returns A {@link actor handle} to the spawned actor.
*/
template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::broker*)>,
......@@ -567,14 +578,6 @@ actor spawn_io(F fun,
return {io::init_and_launch(std::move(ptr))};
}
/*
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor_ptr spawn_io(const char* host, uint16_t port, Ts&&... args) {
auto ptr = io::ipv4_io_stream::connect_to(host, port);
return spawn_io<Impl>(ptr, ptr, std::forward<Ts>(args)...);
}
*/
template<spawn_options Options = no_spawn_options,
typename F = std::function<void (io::broker*)>,
typename... Ts>
......@@ -619,7 +622,7 @@ struct hash<cppa::actor> {
template<>
struct hash<cppa::actor_addr> {
inline size_t operator()(const cppa::actor_addr& ref) const {
return static_cast<size_t>(ref->id());
return static_cast<size_t>(ref.id());
}
};
} // namespace std
......
......@@ -46,11 +46,11 @@ class raw_access {
public:
static abstract_actor* get(const actor& hdl) {
return hdl.m_ops.m_ptr.get();
return hdl.m_ptr.get();
}
static abstract_actor* get(const actor_addr& hdl) {
return hdl.m_ops.m_ptr.get();
return hdl.m_ptr.get();
}
static abstract_channel* get(const channel& hdl) {
......
......@@ -38,12 +38,16 @@
#include "cppa/mailbox_based.hpp"
#include "cppa/behavior_stack_based.hpp"
#include "cppa/detail/response_future_util.hpp"
#include "cppa/detail/response_handle_util.hpp"
namespace cppa {
class event_based_actor;
/**
* @brief Helper class to enable users to add continuations when dealing
* with synchronous sends.
*/
class continue_helper {
public:
......@@ -53,6 +57,11 @@ class continue_helper {
inline continue_helper(message_id mid, event_based_actor* self)
: m_mid(mid), m_self(self) { }
/**
* @brief Adds a continuation to the synchronous message handler
* that is invoked if the response handler successfully returned.
* @param fun The continuation as functor object.
*/
template<typename F>
continue_helper& continue_with(F fun) {
return continue_with(behavior::continuation_fun{partial_function{
......@@ -60,8 +69,16 @@ class continue_helper {
}});
}
/**
* @brief Adds a continuation to the synchronous message handler
* that is invoked if the response handler successfully returned.
* @param fun The continuation as functor object.
*/
continue_helper& continue_with(behavior::continuation_fun fun);
/**
* @brief Returns the ID of the expected response message.
*/
message_id get_message_id() const {
return m_mid;
}
......@@ -74,6 +91,12 @@ class continue_helper {
};
/**
* @brief A cooperatively scheduled, event-based actor implementation.
*
* This is the recommended base class for user-defined actors and is used
* implicitly when spawning functor-based actors without the
* {@link blocking_api} flag.
*
* @extends local_actor
*/
class event_based_actor : public extend<local_actor>::
......@@ -81,18 +104,37 @@ class event_based_actor : public extend<local_actor>::
protected:
/**
* @brief Returns the initial actor behavior.
*/
virtual behavior make_behavior() = 0;
void forward_to(const actor& other);
/**
* @brief Forwards the last received message to @p whom.
*/
void forward_to(const actor& whom);
public:
class response_future {
/**
* @brief This helper class identifies an expected response message
* enables <tt>sync_send(...).then(...)</tt>.
*/
class response_handle {
friend class event_based_actor;
public:
response_future() = delete;
response_handle() = delete;
response_handle(const response_handle&) = default;
response_handle& operator=(const response_handle&) = default;
/**
* @brief Sets @p bhvr as event-handler for the response message.
*/
inline continue_helper then(behavior bhvr) {
m_self->bhvr_stack().push_back(std::move(bhvr), m_mid);
return {m_mid, m_self};
......@@ -120,20 +162,13 @@ class event_based_actor : public extend<local_actor>::
return then(detail::fs2bhvr(m_self, fs...));
}
response_future(const response_future&) = default;
response_future& operator=(const response_future&) = default;
inline response_future(const message_id& from, event_based_actor* self)
: m_mid(from), m_self(self) { }
private:
response_handle(const message_id& from, event_based_actor* self);
message_id m_mid;
event_based_actor* m_self;
inline void check_consistency() { }
};
/**
......@@ -145,7 +180,7 @@ class event_based_actor : public extend<local_actor>::
* sent message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
response_future sync_send_tuple(const actor& dest, any_tuple what);
response_handle sync_send_tuple(const actor& whom, any_tuple what);
/**
......@@ -159,21 +194,34 @@ class event_based_actor : public extend<local_actor>::
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<typename... Ts>
inline response_future sync_send(const actor& dest, Ts&&... what) {
inline response_handle sync_send(const actor& whom, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(dest, make_any_tuple(std::forward<Ts>(what)...));
return sync_send_tuple(whom, make_any_tuple(std::forward<Ts>(what)...));
}
response_future timed_sync_send_tuple(const util::duration& rtime,
const actor& dest,
/**
* @brief Sends a synchronous message with timeout @p rtime to @p whom.
* @param whom Receiver of the message.
* @param rtime Relative time until this messages times out.
* @param what Message content as tuple.
*/
response_handle timed_sync_send_tuple(const util::duration& rtime,
const actor& whom,
any_tuple what);
/**
* @brief Sends a synchronous message with timeout @p rtime to @p whom.
* @param whom Receiver of the message.
* @param rtime Relative time until this messages times out.
* @param what Message elements.
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template<typename... Ts>
response_future timed_sync_send(const actor& dest,
const util::duration& rtime,
Ts&&... what) {
response_handle timed_sync_send(const actor& whom,
const util::duration& rtime,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return timed_sync_send_tuple(rtime, dest,
return timed_sync_send_tuple(rtime, whom,
make_any_tuple(std::forward<Ts>(what)...));
}
......
......@@ -66,7 +66,7 @@ namespace cppa {
// forward declarations
class scheduler;
class response_future;
class response_handle;
class local_scheduler;
class sync_handle_helper;
......@@ -91,7 +91,7 @@ actor spawn_in_group(const group&, Ts&&... args);
/**
* @brief Base class for local running Actors.
* @extends actor
* @extends abstract_actor
*/
class local_actor : public extend<abstract_actor>::with<memory_cached> {
......@@ -171,7 +171,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
void send_exit(const actor_addr& whom, std::uint32_t reason);
inline void send_exit(const actor& whom, std::uint32_t reason) {
send_exit(whom->address(), reason);
send_exit(whom.address(), reason);
}
/**
......@@ -266,7 +266,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
void monitor(const actor_addr& whom);
inline void monitor(const actor& whom) {
monitor(whom->address());
monitor(whom.address());
}
/**
......@@ -276,7 +276,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
void demonitor(const actor_addr& whom);
inline void demonitor(const actor& whom) {
demonitor(whom->address());
demonitor(whom.address());
}
/**
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* 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 2.1 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_COMMON_ACTOR_OPS_HPP
#define CPPA_COMMON_ACTOR_OPS_HPP
#include "cppa/abstract_actor.hpp"
#include "cppa/util/comparable.hpp"
namespace cppa {
class actor;
class actor_addr;
namespace detail { class raw_access; }
/**
* @brief Encapsulates actor operations that are valid for both {@link actor}
* and {@link actor_addr}.
*/
class untyped_actor_handle : util::comparable<untyped_actor_handle> {
friend class actor;
friend class actor_addr;
friend class detail::raw_access;
public:
untyped_actor_handle() = default;
inline bool attach(attachable_ptr ptr) const;
/**
* @brief Convenience function that attaches the functor
* or function @p f to this actor.
*
* The actor executes <tt>f()</tt> on exit, or immediatley
* if it already finished execution.
* @param f A functor, function or lambda expression that's executed
* if the actor finishes execution.
* @returns @c true if @p f was successfully attached to the actor;
* otherwise (actor already exited) @p false.
*/
template<typename F>
bool attach_functor(F&& f) const;
inline actor_id id() const;
actor_addr address() const;
const node_id& node() const;
/**
* @brief Returns whether this is an address of a
* remote actor.
*/
bool is_remote() const;
inline bool valid() const {
return static_cast<bool>(m_ptr);
}
intptr_t compare(const untyped_actor_handle& other) const;
protected:
inline untyped_actor_handle(abstract_actor_ptr ptr) : m_ptr(std::move(ptr)) { }
abstract_actor_ptr m_ptr;
};
template<class F>
struct functor_attachable : attachable {
F m_functor;
template<typename T>
inline functor_attachable(T&& arg) : m_functor(std::forward<T>(arg)) { }
void actor_exited(std::uint32_t reason) { m_functor(reason); }
bool matches(const attachable::token&) { return false; }
};
template<typename F>
bool untyped_actor_handle::attach_functor(F&& f) const {
typedef typename util::rm_const_and_ref<F>::type f_type;
typedef functor_attachable<f_type> impl;
return attach(attachable_ptr{new impl(std::forward<F>(f))});
}
inline actor_id untyped_actor_handle::id() const {
return (m_ptr) ? m_ptr->id() : 0;
}
inline bool untyped_actor_handle::attach(attachable_ptr ptr) const {
return m_ptr ? m_ptr->attach(std::move(ptr)) : false;
}
} // namespace cppa
#endif // CPPA_COMMON_ACTOR_OPS_HPP
......@@ -67,7 +67,11 @@ abstract_actor::abstract_actor()
, m_exit_reason(exit_reason::not_exited) { }
void abstract_actor::link_to(const actor& whom) {
link_to(whom->address());
link_to(whom.address());
}
void abstract_actor::unlink_from(const actor& whom) {
unlink_from(whom.address());
}
bool abstract_actor::link_to_impl(const actor_addr& other) {
......
......@@ -35,34 +35,34 @@
#include "cppa/actor_addr.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/blocking_actor.hpp"
#include "cppa/event_based_actor.hpp"
namespace cppa {
actor::actor(const invalid_actor_t&) : m_ops(nullptr) { }
actor::actor(const invalid_actor_t&) : m_ptr(nullptr) { }
actor::actor(abstract_actor* ptr) : m_ops(ptr) { }
actor::actor(abstract_actor* ptr) : m_ptr(ptr) { }
actor& actor::operator=(const invalid_actor_t&) {
m_ops.m_ptr.reset();
m_ptr.reset();
return *this;
}
void actor::handle::enqueue(const message_header& hdr, any_tuple msg) const {
if (m_ptr) m_ptr->enqueue(hdr, std::move(msg));
}
intptr_t actor::compare(const actor& other) const {
return channel::compare(m_ops.m_ptr.get(), other.m_ops.m_ptr.get());
return channel::compare(m_ptr.get(), other.m_ptr.get());
}
intptr_t actor::compare(const actor_addr& other) const {
return m_ops.compare(*other);
return m_ptr.compare(other.m_ptr);
}
void actor::swap(actor& other) {
m_ops.m_ptr.swap(other.m_ops.m_ptr);
m_ptr.swap(other.m_ptr);
}
actor_addr actor::address() const {
return m_ptr ? m_ptr->address() : actor_addr{};
}
} // namespace cppa
......@@ -42,21 +42,33 @@ intptr_t compare_impl(const abstract_actor* lhs, const abstract_actor* rhs) {
}
} // namespace <anonymous>
actor_addr::actor_addr(const invalid_actor_addr_t&) : m_ops(nullptr) { }
actor_addr::actor_addr(const invalid_actor_addr_t&) : m_ptr(nullptr) { }
actor_addr::actor_addr(abstract_actor* ptr) : m_ops(ptr) { }
actor_addr::actor_addr(abstract_actor* ptr) : m_ptr(ptr) { }
intptr_t actor_addr::compare(const actor_addr& other) const {
return compare_impl(m_ops.m_ptr.get(), other.m_ops.m_ptr.get());
return compare_impl(m_ptr.get(), other.m_ptr.get());
}
intptr_t actor_addr::compare(const abstract_actor* other) const {
return compare_impl(m_ops.m_ptr.get(), other);
return compare_impl(m_ptr.get(), other);
}
actor_addr actor_addr::operator=(const invalid_actor_addr_t&) {
m_ops.m_ptr.reset();
m_ptr.reset();
return *this;
}
actor_id actor_addr::id() const {
return (m_ptr) ? m_ptr->id() : 0;
}
const node_id& actor_addr::node() const {
return m_ptr ? m_ptr->node() : *node_id::get();
}
bool actor_addr::is_remote() const {
return m_ptr ? m_ptr->is_proxy() : false;
}
} // namespace cppa
......@@ -45,9 +45,9 @@
namespace cppa {
void actor_namespace::write(serializer* sink, const actor_addr& ptr) {
void actor_namespace::write(serializer* sink, const actor_addr& addr) {
CPPA_REQUIRE(sink != nullptr);
if (!ptr) {
if (!addr) {
node_id::host_id_type zero;
std::fill(zero.begin(), zero.end(), 0);
sink->write_value(static_cast<uint32_t>(0)); // actor id
......@@ -56,11 +56,11 @@ void actor_namespace::write(serializer* sink, const actor_addr& ptr) {
}
else {
// register locally running actors to be able to deserialize them later
if (!ptr->is_remote()) {
get_actor_registry()->put(ptr->id(), detail::raw_access::get(ptr));
if (!addr.is_remote()) {
get_actor_registry()->put(addr.id(), detail::raw_access::get(addr));
}
auto& pinf = ptr->node();
sink->write_value(ptr->id()); // actor id
auto& pinf = addr.node();
sink->write_value(addr.id()); // actor id
sink->write_value(pinf.process_id()); // process id
sink->write_raw(node_id::host_id_size, pinf.host_id().data()); // host id
}
......
......@@ -37,7 +37,7 @@
namespace cppa {
void blocking_actor::response_future::await(behavior& bhvr) {
void blocking_actor::response_handle::await(behavior& bhvr) {
m_self->dequeue_response(bhvr, m_mid);
}
......@@ -46,14 +46,14 @@ void blocking_actor::await_all_other_actors_done() {
get_actor_registry()->await_running_count_equal(1);
}
blocking_actor::response_future
blocking_actor::response_handle
blocking_actor::sync_send_tuple(const actor& dest, any_tuple what) {
auto nri = new_request_id();
dest->enqueue({address(), dest, nri}, std::move(what));
return {nri.response_id(), this};
}
blocking_actor::response_future
blocking_actor::response_handle
blocking_actor::timed_sync_send_tuple(const util::duration& rtime,
const actor& dest,
any_tuple what) {
......
......@@ -46,18 +46,22 @@ continue_helper::continue_with(behavior::continuation_fun fun) {
return *this;
}
event_based_actor::response_handle::response_handle(const message_id& from,
event_based_actor* self)
: m_mid(from), m_self(self) { }
void event_based_actor::forward_to(const actor& whom) {
forward_message(whom, message_priority::normal);
}
event_based_actor::response_future
event_based_actor::response_handle
event_based_actor::sync_send_tuple(const actor& dest, any_tuple what) {
auto nri = new_request_id();
dest->enqueue({address(), dest, nri}, std::move(what));
return {nri.response_id(), this};
}
event_based_actor::response_future
event_based_actor::response_handle
event_based_actor::timed_sync_send_tuple(const util::duration& rtime,
const actor& dest,
any_tuple what) {
......
......@@ -317,7 +317,7 @@ class local_group_module : public abstract_group::module {
actor broker;
m_actor_utype->deserialize(&broker, source);
if (!broker) return invalid_group;
if (!broker->is_remote()) {
if (!broker->is_proxy()) {
return this->get(identifier);
}
else {
......@@ -365,8 +365,9 @@ class remote_group : public abstract_group {
public:
remote_group(abstract_group::module_ptr parent, string id, local_group_ptr decorated)
: super(parent, move(id)), m_decorated(decorated) { }
remote_group(abstract_group::module_ptr parent, string id,
local_group_ptr decorated)
: super(parent, move(id)), m_decorated(decorated) { }
abstract_group::subscription subscribe(const channel& who) {
CPPA_LOG_TRACE(CPPA_TSARG(who));
......
......@@ -222,7 +222,7 @@ void peer::monitor(const actor_addr&,
else {
CPPA_LOG_DEBUG("attach functor to " << entry.first.get());
auto mm = parent();
entry.first->address()->attach_functor([=](uint32_t reason) {
entry.first->attach_functor([=](uint32_t reason) {
mm->run_later([=] {
CPPA_LOGC_TRACE("cppa::io::peer",
"monitor$kill_proxy_helper",
......@@ -266,7 +266,7 @@ void peer::kill_proxy(const actor_addr& sender,
void peer::deliver(const message_header& hdr, any_tuple msg) {
CPPA_LOG_TRACE("");
if (hdr.sender && hdr.sender->is_remote()) {
if (hdr.sender && hdr.sender.is_remote()) {
// is_remote() is guaranteed to return true if and only if
// the actor is derived from actor_proxy, hence we do not
// need to use a dynamic_cast here
......@@ -290,7 +290,7 @@ void peer::link(const actor_addr& lhs, const actor_addr& rhs) {
auto ptr = static_cast<actor_proxy*>(detail::raw_access::get(proxy));
ptr->local_link_to(addr);
};
switch ((lhs->is_remote() ? 0x10 : 0x00) | (rhs->is_remote() ? 0x01 : 0x00)) {
switch ((lhs.is_remote() ? 0x10 : 0x00) | (rhs.is_remote() ? 0x01 : 0x00)) {
case 0x00: // both local
case 0x11: // both remote
detail::raw_access::get(lhs)->link_to(rhs);
......@@ -316,7 +316,7 @@ void peer::unlink(const actor_addr& lhs, const actor_addr& rhs) {
auto ptr = static_cast<actor_proxy*>(detail::raw_access::get(proxy));
ptr->local_unlink_from(addr);
};
switch ((lhs->is_remote() ? 0x10 : 0x00) | (rhs->is_remote() ? 0x01 : 0x00)) {
switch ((lhs.is_remote() ? 0x10 : 0x00) | (rhs.is_remote() ? 0x01 : 0x00)) {
case 0x00: // both local
case 0x11: // both remote
detail::raw_access::get(lhs)->unlink_from(rhs);
......
......@@ -64,7 +64,7 @@ continue_reading_result peer_acceptor::continue_reading() {
auto& pself = node_id::get();
uint32_t process_id = pself->process_id();
try {
actor_id aid = published_actor()->id();
actor_id aid = published_actor().id();
pair.second->write(&aid, sizeof(actor_id));
pair.second->write(&process_id, sizeof(uint32_t));
pair.second->write(pself->host_id().data(),
......
......@@ -224,7 +224,7 @@ void scheduler_helper::printer_loop(blocking_actor* self) {
self->receive_while (gref(running)) (
on(atom("add"), arg_match) >> [&](std::string& str) {
auto s = self->last_sender();
if (!str.empty() && s.valid()) {
if (!str.empty() && s) {
auto i = out.find(s);
if (i == out.end()) {
i = out.insert(make_pair(s, move(str))).first;
......
......@@ -70,7 +70,7 @@ void publish(actor whom, std::unique_ptr<acceptor> aptr) {
if (!whom) return;
get_actor_registry()->put(whom->id(), detail::raw_access::get(whom));
auto mm = get_middleman();
auto addr = whom->address();
auto addr = whom.address();
mm->register_acceptor(addr, new peer_acceptor(mm, move(aptr), addr));
}
......
......@@ -185,7 +185,7 @@ void deserialize_impl(actor_addr& addr, deserializer* source) {
}
void serialize_impl(const actor& ptr, serializer* sink) {
serialize_impl(ptr->address(), sink);
serialize_impl(ptr.address(), sink);
}
void deserialize_impl(actor& ptr, deserializer* source) {
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* 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 2.1 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 "cppa/channel.hpp"
#include "cppa/actor_addr.hpp"
#include "cppa/untyped_actor_handle.hpp"
namespace cppa {
actor_addr untyped_actor_handle::address() const {
return m_ptr ? m_ptr->address() : actor_addr{};
}
const node_id& untyped_actor_handle::node() const {
return m_ptr ? m_ptr->node() : *node_id::get();
}
bool untyped_actor_handle::is_remote() const {
return m_ptr ? m_ptr->is_proxy() : false;
}
intptr_t untyped_actor_handle::compare(const untyped_actor_handle& other) const {
return channel::compare(m_ptr.get(), other.m_ptr.get());
}
} // namespace cppa
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