Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
A
Actor Framework
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Operations
Operations
Metrics
Environments
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
cpp-libs
Actor Framework
Commits
5e5ccb08
Commit
5e5ccb08
authored
Jan 30, 2014
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
implemented send to typed actors
parent
fe9fa2ce
Changes
29
Hide whitespace changes
Inline
Side-by-side
Showing
29 changed files
with
775 additions
and
348 deletions
+775
-348
CMakeLists.txt
CMakeLists.txt
+1
-0
cppa.files
cppa.files
+3
-0
cppa/actor.hpp
cppa/actor.hpp
+4
-4
cppa/behavior_stack_based.hpp
cppa/behavior_stack_based.hpp
+1
-1
cppa/blocking_actor.hpp
cppa/blocking_actor.hpp
+90
-96
cppa/continue_helper.hpp
cppa/continue_helper.hpp
+91
-0
cppa/cppa_fwd.hpp
cppa/cppa_fwd.hpp
+17
-0
cppa/detail/response_handle_util.hpp
cppa/detail/response_handle_util.hpp
+2
-2
cppa/detail/uniform_type_info_map.hpp
cppa/detail/uniform_type_info_map.hpp
+1
-0
cppa/event_based_actor.hpp
cppa/event_based_actor.hpp
+17
-105
cppa/local_actor.hpp
cppa/local_actor.hpp
+179
-46
cppa/policy/invoke_policy.hpp
cppa/policy/invoke_policy.hpp
+1
-1
cppa/response_handle.hpp
cppa/response_handle.hpp
+204
-0
cppa/response_promise.hpp
cppa/response_promise.hpp
+3
-3
cppa/spawn.hpp
cppa/spawn.hpp
+4
-4
cppa/spawn_options.hpp
cppa/spawn_options.hpp
+1
-10
cppa/system_messages.hpp
cppa/system_messages.hpp
+14
-0
cppa/typed_actor.hpp
cppa/typed_actor.hpp
+26
-4
cppa/typed_behavior_stack_based.hpp
cppa/typed_behavior_stack_based.hpp
+1
-1
src/blocking_actor.cpp
src/blocking_actor.cpp
+0
-21
src/continue_helper.cpp
src/continue_helper.cpp
+49
-0
src/event_based_actor.cpp
src/event_based_actor.cpp
+0
-31
src/local_actor.cpp
src/local_actor.cpp
+15
-6
src/sync_request_bouncer.cpp
src/sync_request_bouncer.cpp
+2
-1
src/uniform_type_info_map.cpp
src/uniform_type_info_map.cpp
+13
-6
unit_testing/test_spawn.cpp
unit_testing/test_spawn.cpp
+2
-2
unit_testing/test_sync_send.cpp
unit_testing/test_sync_send.cpp
+3
-3
unit_testing/test_typed_spawn.cpp
unit_testing/test_typed_spawn.cpp
+30
-1
unit_testing/test_uniform_type.cpp
unit_testing/test_uniform_type.cpp
+1
-0
No files found.
CMakeLists.txt
View file @
5e5ccb08
...
...
@@ -152,6 +152,7 @@ set(LIBCPPA_SRC
src/channel.cpp
src/context_switching_resume.cpp
src/continuable.cpp
src/continue_helper.cpp
src/decorated_tuple.cpp
src/actor_proxy.cpp
src/peer.cpp
...
...
cppa.files
View file @
5e5ccb08
...
...
@@ -20,6 +20,7 @@ cppa/binary_serializer.hpp
cppa/blocking_actor.hpp
cppa/channel.hpp
cppa/config.hpp
cppa/continue_helper.hpp
cppa/cow_ptr.hpp
cppa/cow_tuple.hpp
cppa/cppa.hpp
...
...
@@ -155,6 +156,7 @@ cppa/primitive_variant.hpp
cppa/qtsupport/actor_widget_mixin.hpp
cppa/ref_counted.hpp
cppa/replies_to.hpp
cppa/response_handle.hpp
cppa/response_promise.hpp
cppa/resumable.hpp
cppa/sb_actor.hpp
...
...
@@ -256,6 +258,7 @@ src/buffer.cpp
src/channel.cpp
src/context_switching_resume.cpp
src/continuable.cpp
src/continue_helper.cpp
src/decorated_tuple.cpp
src/demangle.cpp
src/deserializer.cpp
...
...
cppa/actor.hpp
View file @
5e5ccb08
...
...
@@ -69,10 +69,10 @@ 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
;
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
;
};
/**
...
...
cppa/behavior_stack_based.hpp
View file @
5e5ccb08
...
...
@@ -92,7 +92,7 @@ class behavior_stack_based : public single_timeout<Base, Subtype> {
return
m_bhvr_stack
;
}
inline
optional
<
behavior
&>
sync_handler
(
message_id
msg_id
)
{
optional
<
behavior
&>
sync_handler
(
message_id
msg_id
)
override
{
return
m_bhvr_stack
.
sync_handler
(
msg_id
);
}
...
...
cppa/blocking_actor.hpp
View file @
5e5ccb08
...
...
@@ -34,87 +34,32 @@
#include "cppa/on.hpp"
#include "cppa/extend.hpp"
#include "cppa/behavior.hpp"
#include "cppa/typed_actor.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/mailbox_based.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/detail/response_handle_util.hpp"
namespace
cppa
{
/**
* @brief A thread-mapped or context-switching actor using a blocking
* receive rather than a behavior-stack based message processing.
* @extends local_actor
*/
class
blocking_actor
:
public
extend
<
local_actor
>::
with
<
mailbox_based
>
{
public:
class
response_handle
{
typedef
blocking_response_handle_tag
response_handle_tag
;
public:
typedef
response_handle
<
blocking_actor
>
response_handle_type
;
response_handle
()
=
delete
;
void
await
(
behavior
&
);
inline
void
await
(
behavior
&&
bhvr
)
{
behavior
arg
{
std
::
move
(
bhvr
)};
await
(
arg
);
}
/**
* @brief Blocks until the response arrives and then executes @p mexpr.
*/
template
<
typename
...
Cs
,
typename
...
Ts
>
void
await
(
const
match_expr
<
Cs
...
>&
arg
,
const
Ts
&
...
args
)
{
await
(
match_expr_convert
(
arg
,
args
...));
}
/**
* @brief Blocks until the response arrives and then executes @p @p fun,
* calls <tt>self->handle_sync_failure()</tt> if the response
* message is an 'EXITED' or 'VOID' message.
*/
template
<
typename
...
Fs
>
typename
std
::
enable_if
<
util
::
all_callable
<
Fs
...
>::
value
>::
type
await
(
Fs
...
fs
)
{
await
(
detail
::
fs2bhvr
(
m_self
,
fs
...));
}
/**
* @brief Returns the awaited response ID.
*/
inline
const
message_id
&
id
()
const
{
return
m_mid
;
}
response_handle
(
const
response_handle
&
)
=
default
;
response_handle
&
operator
=
(
const
response_handle
&
)
=
default
;
inline
response_handle
(
const
message_id
&
from
,
blocking_actor
*
self
)
:
m_mid
(
from
),
m_self
(
self
)
{
}
private:
message_id
m_mid
;
blocking_actor
*
m_self
;
};
class
sync_receive_helper
{
public:
inline
sync_receive_helper
(
const
response_handle
&
mf
)
:
m_mf
(
mf
)
{
}
template
<
typename
...
Ts
>
inline
void
operator
()(
Ts
&&
...
args
)
{
m_mf
.
await
(
std
::
forward
<
Ts
>
(
args
)...);
}
private:
response_handle
m_mf
;
};
/**************************************************************************
* sync_send[_tuple](actor, ...) *
**************************************************************************/
/**
* @brief Sends @p what as a synchronous message to @p whom.
...
...
@@ -125,11 +70,15 @@ 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_handle
sync_send_tuple
(
const
actor
&
dest
,
any_tuple
what
);
inline
response_handle_type
sync_send_tuple
(
message_priority
prio
,
const
actor
&
dest
,
any_tuple
what
)
{
return
{
sync_send_tuple_impl
(
prio
,
dest
,
std
::
move
(
what
)),
this
};
}
response_handle
timed_sync_send_tuple
(
const
util
::
duration
&
rtime
,
const
actor
&
dest
,
any_tuple
what
);
inline
response_handle_type
sync_send_tuple
(
const
actor
&
dest
,
any_tuple
what
)
{
return
sync_send_tuple
(
message_priority
::
normal
,
dest
,
std
::
move
(
what
));
}
/**
* @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom.
...
...
@@ -142,13 +91,35 @@ class blocking_actor : public extend<local_actor>::with<mailbox_based> {
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template
<
typename
...
Ts
>
inline
response_handle
sync_send
(
const
actor
&
dest
,
Ts
&&
...
what
)
{
inline
response_handle_type
sync_send
(
message_priority
prio
,
const
actor
&
dest
,
Ts
&&
...
what
)
{
static_assert
(
sizeof
...(
Ts
)
>
0
,
"no message to send"
);
return
sync_send_tuple
(
prio
,
dest
,
make_any_tuple
(
std
::
forward
<
Ts
>
(
what
)...));
}
template
<
typename
...
Ts
>
inline
response_handle_type
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
)...));
return
sync_send_tuple
(
message_priority
::
normal
,
dest
,
make_any_tuple
(
std
::
forward
<
Ts
>
(
what
)...));
}
/**************************************************************************
* timed_sync_send[_tuple](actor, ...) *
**************************************************************************/
inline
response_handle_type
timed_sync_send_tuple
(
const
util
::
duration
&
rtime
,
const
actor
&
dest
,
any_tuple
what
)
{
return
{
timed_sync_send_tuple_impl
(
message_priority
::
normal
,
dest
,
rtime
,
std
::
move
(
what
)),
this
};
}
template
<
typename
...
Ts
>
response_handle
timed_sync_send
(
const
actor
&
dest
,
response_handle
_type
timed_sync_send
(
const
actor
&
dest
,
const
util
::
duration
&
rtime
,
Ts
&&
...
what
)
{
static_assert
(
sizeof
...(
Ts
)
>
0
,
"no message to send"
);
...
...
@@ -156,6 +127,27 @@ class blocking_actor : public extend<local_actor>::with<mailbox_based> {
make_any_tuple
(
std
::
forward
<
Ts
>
(
what
)...));
}
/**************************************************************************
* sync_send[_tuple](typed_actor<...>, ...) *
**************************************************************************/
template
<
typename
...
Rs
,
typename
...
Ts
>
inline
response_handle_type
sync_send_tuple
(
message_priority
prio
,
const
typed_actor
<
Rs
...
>&
dest
,
cow_tuple
<
Ts
...
>
what
)
{
return
{
sync_send_tuple_impl
(
prio
,
dest
,
std
::
move
(
what
)),
this
};
}
template
<
typename
...
Rs
,
typename
...
Ts
>
inline
response_handle_type
sync_send_tuple
(
const
typed_actor
<
Rs
...
>&
dest
,
cow_tuple
<
Ts
...
>
what
)
{
return
sync_send_tuple
(
message_priority
::
normal
,
dest
,
std
::
move
(
what
));
}
/**************************************************************************
* utility stuff and receive() member function family *
**************************************************************************/
typedef
std
::
chrono
::
high_resolution_clock
::
time_point
timeout_type
;
struct
receive_while_helper
{
...
...
@@ -267,18 +259,6 @@ class blocking_actor : public extend<local_actor>::with<mailbox_based> {
return
{
make_dequeue_callback
(),
stmt
};
}
/**
* @brief Handles a synchronous response message in an event-based way.
* @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_handle
*/
inline
sync_receive_helper
receive_response
(
const
response_handle
&
f
)
{
return
{
f
};
}
/**
* @brief Receives messages until @p stmt returns true.
*
...
...
@@ -304,44 +284,58 @@ class blocking_actor : public extend<local_actor>::with<mailbox_based> {
,
match_expr_convert
(
std
::
forward
<
Ts
>
(
args
)...)};
}
inline
optional
<
behavior
&>
sync_handler
(
message_id
msg_id
)
{
optional
<
behavior
&>
sync_handler
(
message_id
msg_id
)
override
{
auto
i
=
m_sync_handler
.
find
(
msg_id
);
if
(
i
!=
m_sync_handler
.
end
())
return
i
->
second
;
return
none
;
}
// unused in blocking actors
/**
* @brief Blocks this actor until all other actors are done.
*/
void
await_all_other_actors_done
();
/**
* @brief Implements the actor's behavior.
*/
virtual
void
act
()
=
0
;
/**
* @brief Unwinds the stack by throwing an actor_exited exception.
* @throws actor_exited
*/
virtual
void
quit
(
std
::
uint32_t
reason
=
exit_reason
::
normal
);
/** @cond PRIVATE */
// required from invoke_policy; unused in blocking actors
inline
void
remove_handler
(
message_id
)
{
}
// required by receive() member function family
inline
void
dequeue
(
behavior
&&
bhvr
)
{
behavior
tmp
{
std
::
move
(
bhvr
)};
dequeue
(
tmp
);
}
// required by receive() member function family
inline
void
dequeue
(
behavior
&
bhvr
)
{
dequeue_response
(
bhvr
,
message_id
::
invalid
);
}
// implemented by detail::proper_actor
virtual
void
dequeue_response
(
behavior
&
bhvr
,
message_id
mid
)
=
0
;
void
await_all_other_actors_done
();
virtual
void
act
()
=
0
;
/**
* @brief Unwinds the stack by throwing an actor_exited exception.
* @throws actor_exited
*/
virtual
void
quit
(
std
::
uint32_t
reason
=
exit_reason
::
normal
);
/** @endcond */
private:
std
::
map
<
message_id
,
behavior
>
m_sync_handler
;
// helper function to implement receive_(for|while) and do_receive
std
::
function
<
void
(
behavior
&
)
>
make_dequeue_callback
()
{
return
[
=
](
behavior
&
bhvr
)
{
dequeue
(
bhvr
);
};
}
std
::
map
<
message_id
,
behavior
>
m_sync_handler
;
};
}
// namespace cppa
...
...
cppa/continue_helper.hpp
0 → 100644
View file @
5e5ccb08
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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 CONTINUE_HELPER_HPP
#define CONTINUE_HELPER_HPP
#include <functional>
#include "cppa/on.hpp"
#include "cppa/behavior.hpp"
#include "cppa/message_id.hpp"
namespace
cppa
{
class
local_actor
;
/**
* @brief Helper class to enable users to add continuations
* when dealing with synchronous sends.
*/
class
continue_helper
{
public:
typedef
int
message_id_wrapper_tag
;
continue_helper
(
message_id
mid
,
local_actor
*
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
{
on
(
any_vals
,
arg_match
)
>>
fun
}});
}
/**
* @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
;
}
private:
message_id
m_mid
;
local_actor
*
m_self
;
};
}
// namespace cppa
#endif // CONTINUE_HELPER_HPP
cppa/cppa_fwd.hpp
View file @
5e5ccb08
...
...
@@ -33,6 +33,8 @@
#include <cstdint>
#include "cppa/spawn_options.hpp"
namespace
cppa
{
// classes
...
...
@@ -74,6 +76,21 @@ typedef intrusive_ptr<node_id> node_id_ptr;
// weak intrusive pointer typedefs
typedef
weak_intrusive_ptr
<
actor_proxy
>
weak_actor_proxy_ptr
;
// prototype definitions of the spawn function famility;
// implemented in spawn.hpp (this header is included there)
template
<
class
Impl
,
spawn_options
Options
=
no_spawn_options
,
typename
...
Ts
>
actor
spawn
(
Ts
&&
...
args
);
template
<
spawn_options
Options
=
no_spawn_options
,
typename
...
Ts
>
actor
spawn
(
Ts
&&
...
args
);
template
<
class
Impl
,
spawn_options
Options
=
no_spawn_options
,
typename
...
Ts
>
actor
spawn_in_group
(
const
group
&
,
Ts
&&
...
args
);
template
<
spawn_options
Options
=
no_spawn_options
,
typename
...
Ts
>
actor
spawn_in_group
(
const
group
&
,
Ts
&&
...
args
);
}
// namespace cppa
#endif // CPPA_FWD_HPP
cppa/detail/response_handle_util.hpp
View file @
5e5ccb08
...
...
@@ -48,8 +48,8 @@ behavior fs2bhvr(Actor* self, Fs... fs) {
};
return
behavior
{
on
<
sync_timeout_msg
>
()
>>
handle_sync_timeout
,
on
(
atom
(
"VOID"
)
)
>>
skip_message
,
on
(
atom
(
"EXITED"
)
)
>>
skip_message
,
on
<
unit_t
>
(
)
>>
skip_message
,
on
<
sync_exited_msg
>
(
)
>>
skip_message
,
(
on
(
any_vals
,
arg_match
)
>>
std
::
move
(
fs
))...
};
}
...
...
cppa/detail/uniform_type_info_map.hpp
View file @
5e5ccb08
...
...
@@ -73,6 +73,7 @@ using mapped_type_list = util::type_list<
io
::
accept_handle
,
io
::
connection_handle
,
message_header
,
sync_exited_msg
,
sync_timeout_msg
,
timeout_msg
,
unit_t
,
...
...
cppa/event_based_actor.hpp
View file @
5e5ccb08
...
...
@@ -36,6 +36,7 @@
#include "cppa/logging.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/mailbox_based.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/behavior_stack_based.hpp"
#include "cppa/detail/response_handle_util.hpp"
...
...
@@ -44,52 +45,6 @@ namespace cppa {
class
event_based_actor
;
/**
* @brief Helper class to enable users to add continuations when dealing
* with synchronous sends.
*/
class
continue_helper
{
public:
typedef
int
message_id_wrapper_tag
;
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
{
on
(
any_vals
,
arg_match
)
>>
fun
}});
}
/**
* @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
;
}
private:
message_id
m_mid
;
event_based_actor
*
m_self
;
};
/**
* @brief A cooperatively scheduled, event-based actor implementation.
*
...
...
@@ -116,60 +71,9 @@ class event_based_actor : public extend<local_actor>::
public:
/**
* @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_handle
()
=
delete
;
response_handle
(
const
response_handle
&
)
=
default
;
response_handle
&
operator
=
(
const
response_handle
&
)
=
default
;
typedef
nonblocking_response_handle_tag
response_handle_tag
;
/**
* @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
};
}
/**
* @brief Sets @p mexpr as event-handler for the response message.
*/
template
<
typename
...
Cs
,
typename
...
Ts
>
continue_helper
then
(
const
match_expr
<
Cs
...
>&
arg
,
const
Ts
&
...
args
)
{
return
then
(
match_expr_convert
(
arg
,
args
...));
}
/**
* @brief Sets @p fun as event-handler for the response message, calls
* <tt>self->handle_sync_failure()</tt> if the response message
* is an 'EXITED' or 'VOID' message.
*/
template
<
typename
...
Fs
>
typename
std
::
enable_if
<
util
::
all_callable
<
Fs
...
>::
value
,
continue_helper
>::
type
then
(
Fs
...
fs
)
{
return
then
(
detail
::
fs2bhvr
(
m_self
,
fs
...));
}
private:
response_handle
(
const
message_id
&
from
,
event_based_actor
*
self
);
message_id
m_mid
;
event_based_actor
*
m_self
;
};
typedef
response_handle
<
event_based_actor
>
untyped_response_handle
;
/**
* @brief Sends @p what as a synchronous message to @p whom.
...
...
@@ -180,8 +84,12 @@ 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_handle
sync_send_tuple
(
const
actor
&
whom
,
any_tuple
what
);
inline
untyped_response_handle
sync_send_tuple
(
const
actor
&
whom
,
any_tuple
what
)
{
return
{
sync_send_tuple_impl
(
message_priority
::
normal
,
whom
,
std
::
move
(
what
)),
this
};
}
/**
* @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom.
...
...
@@ -194,7 +102,7 @@ class event_based_actor : public extend<local_actor>::
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template
<
typename
...
Ts
>
inline
response_handle
sync_send
(
const
actor
&
whom
,
Ts
&&
...
what
)
{
inline
untyped_
response_handle
sync_send
(
const
actor
&
whom
,
Ts
&&
...
what
)
{
static_assert
(
sizeof
...(
Ts
)
>
0
,
"no message to send"
);
return
sync_send_tuple
(
whom
,
make_any_tuple
(
std
::
forward
<
Ts
>
(
what
)...));
}
...
...
@@ -205,9 +113,13 @@ class event_based_actor : public extend<local_actor>::
* @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
,
untyped_
response_handle
timed_sync_send_tuple
(
const
util
::
duration
&
rtime
,
const
actor
&
whom
,
any_tuple
what
);
any_tuple
what
)
{
return
{
timed_sync_send_tuple_impl
(
message_priority
::
normal
,
whom
,
rtime
,
std
::
move
(
what
)),
this
};
}
/**
* @brief Sends a synchronous message with timeout @p rtime to @p whom.
...
...
@@ -217,7 +129,7 @@ class event_based_actor : public extend<local_actor>::
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template
<
typename
...
Ts
>
response_handle
timed_sync_send
(
const
actor
&
whom
,
untyped_
response_handle
timed_sync_send
(
const
actor
&
whom
,
const
util
::
duration
&
rtime
,
Ts
&&
...
what
)
{
static_assert
(
sizeof
...(
Ts
)
>
0
,
"no message to send"
);
...
...
cppa/local_actor.hpp
View file @
5e5ccb08
...
...
@@ -36,12 +36,12 @@
#include <functional>
#include "cppa/actor.hpp"
#include "cppa/abstract_group.hpp"
#include "cppa/extend.hpp"
#include "cppa/channel.hpp"
#include "cppa/behavior.hpp"
#include "cppa/cppa_fwd.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/cow_tuple.hpp"
#include "cppa/message_id.hpp"
#include "cppa/match_expr.hpp"
#include "cppa/actor_state.hpp"
...
...
@@ -51,6 +51,7 @@
#include "cppa/memory_cached.hpp"
#include "cppa/message_header.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/abstract_group.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/response_promise.hpp"
#include "cppa/message_priority.hpp"
...
...
@@ -59,6 +60,7 @@
#include "cppa/util/duration.hpp"
#include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/typed_actor_util.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
...
...
@@ -66,7 +68,6 @@ namespace cppa {
// forward declarations
class
scheduler
;
class
response_handle
;
class
local_scheduler
;
class
sync_handle_helper
;
...
...
@@ -74,21 +75,6 @@ namespace util {
struct
fiber
;
}
// namespace util
// prototype definitions of the spawn function famility;
// implemented in spawn.hpp (this header is included there)
template
<
class
Impl
,
spawn_options
Options
=
no_spawn_options
,
typename
...
Ts
>
actor
spawn
(
Ts
&&
...
args
);
template
<
spawn_options
Options
=
no_spawn_options
,
typename
...
Ts
>
actor
spawn
(
Ts
&&
...
args
);
template
<
class
Impl
,
spawn_options
Options
=
no_spawn_options
,
typename
...
Ts
>
actor
spawn_in_group
(
const
group
&
,
Ts
&&
...
args
);
template
<
spawn_options
Options
=
no_spawn_options
,
typename
...
Ts
>
actor
spawn_in_group
(
const
group
&
,
Ts
&&
...
args
);
/**
* @brief Base class for local running Actors.
* @extends abstract_actor
...
...
@@ -105,16 +91,6 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
~
local_actor
();
inline
actor
eval_opts
(
spawn_options
opts
,
actor
res
)
{
if
(
has_monitor_flag
(
opts
))
{
monitor
(
res
);
}
if
(
has_link_flag
(
opts
))
{
link_to
(
res
);
}
return
res
;
}
template
<
class
Impl
,
spawn_options
Options
=
no_spawn_options
,
typename
...
Ts
>
actor
spawn
(
Ts
&&
...
args
)
{
...
...
@@ -143,52 +119,168 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
/**
* @brief Sends @p what to the receiver specified in @p dest.
*/
void
send_tuple
(
message_priority
prio
,
const
channel
&
dest
,
any_tuple
what
);
void
send_tuple
(
message_priority
prio
,
const
channel
&
whom
,
any_tuple
what
);
/**
* @brief Sends @p what to the receiver specified in @p dest.
*/
inline
void
send_tuple
(
const
channel
&
whom
,
any_tuple
what
)
{
send_tuple
(
message_priority
::
normal
,
whom
,
std
::
move
(
what
));
}
/**
* @brief Sends <tt>{what...}</tt> to the receiver specified in @p hdr.
* @pre <tt>sizeof...(Ts) > 0</tt>
* @brief Sends <tt>{what...}</tt> to @p whom.
* @param prio Priority of the message.
* @param whom Receiver of the message.
* @param what Message elements.
* @pre <tt>sizeof...(Ts) > 0</tt>.
*/
template
<
typename
...
Ts
>
void
send
(
message_priority
prio
,
const
channel
&
whom
,
Ts
&&
...
what
)
{
inline
void
send
(
message_priority
prio
,
const
channel
&
whom
,
Ts
&&
...
what
)
{
static_assert
(
sizeof
...(
Ts
)
>
0
,
"sizeof...(Ts) == 0"
);
send_tuple
(
prio
,
whom
,
make_any_tuple
(
std
::
forward
<
Ts
>
(
what
)...));
}
/**
* @brief Sends @p what to the receiver specified in @p dest.
* @brief Sends <tt>{what...}</tt> to @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @pre <tt>sizeof...(Ts) > 0</tt>.
*/
void
send_tuple
(
const
channel
&
dest
,
any_tuple
what
);
template
<
typename
...
Ts
>
inline
void
send
(
const
channel
&
whom
,
Ts
&&
...
what
)
{
static_assert
(
sizeof
...(
Ts
)
>
0
,
"sizeof...(Ts) == 0"
);
send_tuple
(
message_priority
::
normal
,
whom
,
make_any_tuple
(
std
::
forward
<
Ts
>
(
what
)...));
}
/**
* @brief Sends <tt>{what...}</tt> to the receiver specified in @p hdr.
* @pre <tt>sizeof...(Ts) > 0</tt>
* @brief Sends @p what to @p whom.
* @param prio Priority of the message.
* @param whom Receiver of the message.
* @param what Message content as a tuple.
*/
template
<
typename
...
Ts
>
void
send
(
const
channel
&
whom
,
Ts
&&
...
what
)
{
send_tuple
(
whom
,
make_any_tuple
(
std
::
forward
<
Ts
>
(
what
)...));
template
<
typename
...
Signatures
,
typename
...
Ts
>
void
send_tuple
(
message_priority
prio
,
const
typed_actor
<
Signatures
...
>&
whom
,
cow_tuple
<
Ts
...
>
what
)
{
check_typed_input
(
whom
,
what
);
send_tuple
(
prio
,
whom
.
m_ptr
,
any_tuple
{
std
::
move
(
what
)});
}
/**
* @brief Sends @p what to @p whom.
* @param whom Receiver of the message.
* @param what Message content as a tuple.
*/
template
<
typename
...
Signatures
,
typename
...
Ts
>
void
send_tuple
(
const
typed_actor
<
Signatures
...
>&
whom
,
cow_tuple
<
Ts
...
>
what
)
{
check_typed_input
(
whom
,
what
);
send_tuple_impl
(
message_priority
::
normal
,
whom
,
std
::
move
(
what
));
}
/**
* @brief Sends <tt>{what...}</tt> to @p whom.
* @param prio Priority of the message.
* @param whom Receiver of the message.
* @param what Message elements.
* @pre <tt>sizeof...(Ts) > 0</tt>.
*/
template
<
typename
...
Signatures
,
typename
...
Ts
>
void
send
(
message_priority
prio
,
const
typed_actor
<
Signatures
...
>&
whom
,
cow_tuple
<
Ts
...
>
what
)
{
send_tuple
(
prio
,
whom
,
make_cow_tuple
(
std
::
forward
<
Ts
>
(
what
)...));
}
/**
* @brief Sends <tt>{what...}</tt> to @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @pre <tt>sizeof...(Ts) > 0</tt>.
*/
template
<
typename
...
Signatures
,
typename
...
Ts
>
void
send
(
const
typed_actor
<
Signatures
...
>&
whom
,
Ts
...
what
)
{
send_tuple
(
message_priority
::
normal
,
whom
,
make_cow_tuple
(
std
::
forward
<
Ts
>
(
what
)...));
}
/**
* @brief Sends an exit message to @p whom.
*/
void
send_exit
(
const
actor_addr
&
whom
,
std
::
uint32_t
reason
);
/**
* @copydoc send_exit(const actor_addr&, std::uint32_t)
*/
inline
void
send_exit
(
const
actor
&
whom
,
std
::
uint32_t
reason
)
{
send_exit
(
whom
.
address
(),
reason
);
}
/**
* @copydoc send_exit(const actor_addr&, std::uint32_t)
*/
template
<
typename
...
Signatures
>
void
send_exit
(
const
typed_actor
<
Signatures
...
>&
whom
,
std
::
uint32_t
reason
)
{
send_exit
(
whom
.
address
(),
reason
);
}
/**
* @brief Sends a message to @p whom that is delayed by @p rel_time.
* @param prio Priority of the message.
* @param whom Receiver of the message.
* @param rtime Relative time
duration
to delay the message in
* @param rtime Relative time to delay the message in
* microseconds, milliseconds, seconds or minutes.
* @param data Message content as a tuple.
*/
void
delayed_send_tuple
(
const
channel
&
dest
,
void
delayed_send_tuple
(
message_priority
prio
,
const
channel
&
whom
,
const
util
::
duration
&
rtime
,
any_tuple
data
);
/**
* @brief Sends a message to @p whom that is delayed by @p rel_time.
* @param whom Receiver of the message.
* @param rtime Relative time to delay the message in
* microseconds, milliseconds, seconds or minutes.
* @param data Message content as a tuple.
*/
inline
void
delayed_send_tuple
(
const
channel
&
whom
,
const
util
::
duration
&
rtime
,
any_tuple
data
)
{
delayed_send_tuple
(
message_priority
::
normal
,
whom
,
rtime
,
std
::
move
(
data
));
}
/**
* @brief Sends a message to @p whom that is delayed by @p rel_time.
* @param prio Priority of the message.
* @param whom Receiver of the message.
* @param rtime Relative time to delay the message in
* microseconds, milliseconds, seconds or minutes.
* @param data Message content as a tuple.
*/
template
<
typename
...
Ts
>
void
delayed_send
(
const
channel
&
dest
,
const
util
::
duration
&
rtime
,
void
delayed_send
(
message_priority
prio
,
const
channel
&
whom
,
const
util
::
duration
&
rtime
,
Ts
&&
...
args
)
{
delayed_send_tuple
(
prio
,
whom
,
rtime
,
make_any_tuple
(
std
::
forward
<
Ts
>
(
args
)...));
}
/**
* @brief Sends a message to @p whom that is delayed by @p rel_time.
* @param whom Receiver of the message.
* @param rtime Relative time to delay the message in
* microseconds, milliseconds, seconds or minutes.
* @param data Message content as a tuple.
*/
template
<
typename
...
Ts
>
void
delayed_send
(
const
channel
&
whom
,
const
util
::
duration
&
rtime
,
Ts
&&
...
args
)
{
delayed_send_tuple
(
dest
,
rtime
,
make_any_tuple
(
std
::
forward
<
Ts
>
(
args
)...));
delayed_send_tuple
(
message_priority
::
normal
,
whom
,
rtime
,
make_any_tuple
(
std
::
forward
<
Ts
>
(
args
)...));
}
/**
...
...
@@ -265,6 +357,9 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
*/
void
monitor
(
const
actor_addr
&
whom
);
/**
* @copydoc monitor(const actor_addr&)
*/
inline
void
monitor
(
const
actor
&
whom
)
{
monitor
(
whom
.
address
());
}
...
...
@@ -329,6 +424,18 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
/** @cond PRIVATE */
local_actor
();
inline
actor
eval_opts
(
spawn_options
opts
,
actor
res
)
{
if
(
has_monitor_flag
(
opts
))
{
monitor
(
res
);
}
if
(
has_link_flag
(
opts
))
{
link_to
(
res
);
}
return
res
;
}
inline
void
current_node
(
mailbox_element
*
ptr
)
{
this
->
m_current_node
=
ptr
;
}
...
...
@@ -353,15 +460,26 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
else
quit
(
exit_reason
::
unhandled_sync_failure
);
}
local_actor
();
inline
bool
chaining_enabled
();
// returns the response ID
message_id
timed_sync_send_tuple_impl
(
message_priority
mp
,
const
actor
&
whom
,
const
util
::
duration
&
rel_time
,
any_tuple
&&
what
);
// returns the response ID
message_id
sync_send_tuple_impl
(
message_priority
mp
,
const
actor
&
whom
,
any_tuple
&&
what
);
// returns the response ID
template
<
typename
...
Signatures
,
typename
...
Ts
>
message_id
sync_send_tuple_impl
(
message_priority
mp
,
const
typed_actor
<
Signatures
...
>&
whom
,
cow_tuple
<
Ts
...
>&&
what
)
{
check_typed_input
(
whom
,
what
);
return
sync_send_tuple_impl
(
mp
,
whom
.
m_ptr
,
any_tuple
{
std
::
move
(
what
)});
}
// returns 0 if last_dequeued() is an asynchronous or sync request message,
// a response id generated from the request id otherwise
inline
message_id
get_response_id
();
...
...
@@ -399,6 +517,8 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
return
&
m_dummy_node
;
}
virtual
optional
<
behavior
&>
sync_handler
(
message_id
msg_id
)
=
0
;
protected:
template
<
typename
...
Ts
>
...
...
@@ -435,6 +555,19 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
private:
template
<
typename
...
Signatures
,
typename
...
Ts
>
static
void
check_typed_input
(
const
typed_actor
<
Signatures
...
>&
,
const
cow_tuple
<
Ts
...
>&
)
{
static
constexpr
int
input_pos
=
util
::
tl_find_if
<
util
::
type_list
<
Signatures
...
>
,
detail
::
input_is
<
util
::
type_list
<
Ts
...
>
>::
template
eval
>
::
value
;
static_assert
(
input_pos
>=
0
,
"typed actor does not support given input"
);
}
std
::
function
<
void
()
>
m_sync_failure_handler
;
std
::
function
<
void
()
>
m_sync_timeout_handler
;
...
...
cppa/policy/invoke_policy.hpp
View file @
5e5ccb08
...
...
@@ -221,7 +221,7 @@ class invoke_policy {
<<
" did not reply to a "
"synchronous request message"
);
auto
fhdl
=
fetch_response_promise
(
self
,
hdl
);
if
(
fhdl
)
fhdl
.
deliver
(
make_any_tuple
(
atom
(
"VOID"
)
));
if
(
fhdl
)
fhdl
.
deliver
(
make_any_tuple
(
unit
));
}
}
else
{
if
(
detail
::
matches
<
atom_value
,
std
::
uint64_t
>
(
*
res
)
...
...
cppa/response_handle.hpp
0 → 100644
View file @
5e5ccb08
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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_RESPONSE_HANDLE_HPP
#define CPPA_RESPONSE_HANDLE_HPP
#include <type_traits>
#include "cppa/message_id.hpp"
#include "cppa/typed_behavior.hpp"
#include "cppa/continue_helper.hpp"
#include "cppa/util/type_list.hpp"
#include "cppa/detail/response_handle_util.hpp"
namespace
cppa
{
/**
* @brief This tag identifies response handles featuring a
* nonblocking API by providing a @p then member function.
*/
struct
nonblocking_response_handle_tag
{
};
/**
* @brief This tag identifies response handles featuring a
* blocking API by providing an @p await member function.
*/
struct
blocking_response_handle_tag
{
};
/**
* @brief This helper class identifies an expected response message
* and enables <tt>sync_send(...).then(...)</tt>.
*
* @tparam Self The type of the actor this handle belongs to.
* @tparam Result Either any_tuple or type_list<R1, R2, ... Rn>.
* @tparam Tag Either {@link nonblocking_response_handle_tag} or
* {@link blocking_response_handle_tag}.
*/
template
<
class
Self
,
class
Result
=
any_tuple
,
class
Tag
=
typename
Self
::
response_handle_tag
>
class
response_handle
{
// default impl for nonblocking_response_handle_tag
friend
Self
;
public:
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
)
{
return
then_impl
<
Result
>
(
std
::
move
(
bhvr
));
}
/**
* @brief Sets @p mexpr as event-handler for the response message.
*/
template
<
typename
...
Cs
,
typename
...
Ts
>
continue_helper
then
(
const
match_expr
<
Cs
...
>&
arg
,
const
Ts
&
...
args
)
{
return
then_impl
<
Result
>
({
arg
,
args
...});
}
/**
* @brief Sets @p fun as event-handler for the response message, calls
* <tt>self->handle_sync_failure()</tt> if the response message
* is an 'EXITED' or 'VOID' message.
*/
template
<
typename
...
Fs
>
typename
std
::
enable_if
<
util
::
all_callable
<
Fs
...
>::
value
,
continue_helper
>::
type
then
(
Fs
...
fs
)
{
return
then_impl
<
Result
>
(
detail
::
fs2bhvr
(
m_self
,
fs
...));
}
private:
template
<
typename
R
>
typename
std
::
enable_if
<
std
::
is_same
<
R
,
any_tuple
>::
value
,
continue_helper
>::
type
then_impl
(
behavior
&&
bhvr
)
{
m_self
->
bhvr_stack
().
push_back
(
std
::
move
(
bhvr
),
m_mid
);
return
{
m_mid
,
m_self
};
}
template
<
typename
R
>
typename
std
::
enable_if
<
util
::
is_type_list
<
R
>::
value
,
continue_helper
>::
type
then_impl
(
behavior
&&
bhvr
)
{
m_self
->
bhvr_stack
().
push_back
(
std
::
move
(
bhvr
),
m_mid
);
return
{
m_mid
,
m_self
};
}
response_handle
(
const
message_id
&
mid
,
Self
*
self
)
:
m_mid
(
mid
),
m_self
(
self
)
{
}
message_id
m_mid
;
Self
*
m_self
;
};
template
<
class
Self
,
typename
Result
>
class
response_handle
<
Self
,
Result
,
blocking_response_handle_tag
>
{
friend
Self
;
public:
response_handle
()
=
delete
;
response_handle
(
const
response_handle
&
)
=
default
;
response_handle
&
operator
=
(
const
response_handle
&
)
=
default
;
/**
* @brief Blocks until the response arrives and then executes @p pfun.
*/
void
await
(
behavior
&
pfun
)
{
m_self
->
dequeue_response
(
pfun
,
m_mid
);
}
/**
* @copydoc await(behavior&)
*/
inline
void
await
(
behavior
&&
pfun
)
{
behavior
arg
{
std
::
move
(
pfun
)};
await
(
arg
);
}
/**
* @brief Blocks until the response arrives and then executes
* the corresponding handler.
*/
template
<
typename
...
Cs
,
typename
...
Ts
>
void
await
(
const
match_expr
<
Cs
...
>&
arg
,
const
Ts
&
...
args
)
{
await
(
match_expr_convert
(
arg
,
args
...));
}
/**
* @brief Blocks until the response arrives and then executes
* the corresponding handler.
* @note Calls <tt>self->handle_sync_failure()</tt> if the response
* message is an 'EXITED' or 'VOID' message.
*/
template
<
typename
...
Fs
>
typename
std
::
enable_if
<
util
::
all_callable
<
Fs
...
>::
value
>::
type
await
(
Fs
...
fs
)
{
await
(
detail
::
fs2bhvr
(
m_self
,
fs
...));
}
private:
response_handle
(
const
message_id
&
mid
,
Self
*
self
)
:
m_mid
(
mid
),
m_self
(
self
)
{
}
message_id
m_mid
;
Self
*
m_self
;
};
}
// namespace cppa
#endif // CPPA_RESPONSE_HANDLE_HPP
cppa/response_promise.hpp
View file @
5e5ccb08
...
...
@@ -28,8 +28,8 @@
\******************************************************************************/
#ifndef CPPA_RESPONSE_
HANDL
E_HPP
#define CPPA_RESPONSE_
HANDL
E_HPP
#ifndef CPPA_RESPONSE_
PROMIS
E_HPP
#define CPPA_RESPONSE_
PROMIS
E_HPP
#include "cppa/actor.hpp"
#include "cppa/any_tuple.hpp"
...
...
@@ -78,4 +78,4 @@ class response_promise {
}
// namespace cppa
#endif // CPPA_RESPONSE_
HANDL
E_HPP
#endif // CPPA_RESPONSE_
PROMIS
E_HPP
cppa/spawn.hpp
View file @
5e5ccb08
...
...
@@ -35,8 +35,8 @@
#include "cppa/policy.hpp"
#include "cppa/logging.hpp"
#include "cppa/cppa_fwd.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/typed_actor.hpp"
#include "cppa/spawn_options.hpp"
#include "cppa/typed_event_based_actor.hpp"
...
...
@@ -157,7 +157,7 @@ intrusive_ptr<Impl> spawn_fwd_args(BeforeLaunch before_launch_fun, Ts&&... args)
template
<
class
Impl
,
spawn_options
Opts
,
typename
...
Ts
>
actor
spawn
(
Ts
&&
...
args
)
{
return
detail
::
spawn_fwd_args
<
Impl
,
Opts
>
(
[](
local_actor
*
)
{
/* no-op as BeforeLaunch callback */
},
[](
Impl
*
)
{
/* no-op as BeforeLaunch callback */
},
std
::
forward
<
Ts
>
(
args
)...);
}
...
...
@@ -196,7 +196,7 @@ actor spawn_in_group(const group& grp, Ts&&... args) {
detail
::
functor_based_actor
>::
type
;
return
detail
::
spawn_fwd_args
<
base_class
,
Opts
>
(
[
&
](
local_actor
*
ptr
)
{
ptr
->
join
(
grp
);
},
[
&
](
base_class
*
ptr
)
{
ptr
->
join
(
grp
);
},
std
::
forward
<
Ts
>
(
args
)...);
}
...
...
@@ -211,7 +211,7 @@ actor spawn_in_group(const group& grp, Ts&&... args) {
template
<
class
Impl
,
spawn_options
Opts
,
typename
...
Ts
>
actor
spawn_in_group
(
const
group
&
grp
,
Ts
&&
...
args
)
{
return
detail
::
spawn_fwd_args
<
Impl
,
Opts
>
(
[
&
](
local_actor
*
ptr
)
{
ptr
->
join
(
grp
);
},
[
&
](
Impl
*
ptr
)
{
ptr
->
join
(
grp
);
},
std
::
forward
<
Ts
>
(
args
)...);
}
...
...
cppa/spawn_options.hpp
View file @
5e5ccb08
...
...
@@ -65,10 +65,6 @@ constexpr spawn_options operator+(const spawn_options& lhs,
|
static_cast
<
int
>
(
rhs
));
}
#ifndef CPPA_DOCUMENTATION
namespace
{
#endif
/**
* @brief Denotes default settings.
*/
...
...
@@ -111,10 +107,6 @@ constexpr spawn_options blocking_api = spawn_options::blocking_api_flag;
constexpr
spawn_options
priority_aware
=
spawn_options
::
priority_aware_flag
+
spawn_options
::
detach_flag
;
#ifndef CPPA_DOCUMENTATION
}
// namespace <anonymous>
#endif
/**
* @brief Checks wheter @p haystack contains @p needle.
* @relates spawn_options
...
...
@@ -182,8 +174,7 @@ constexpr bool is_unbound(spawn_options opts) {
constexpr
spawn_options
make_unbound
(
spawn_options
opts
)
{
return
static_cast
<
spawn_options
>
(
(
static_cast
<
int
>
(
opts
)
&
~
static_cast
<
int
>
(
spawn_options
::
link_flag
))
&
~
static_cast
<
int
>
(
spawn_options
::
monitor_flag
));
&
~
(
static_cast
<
int
>
(
linked
)
|
static_cast
<
int
>
(
monitored
))));
}
/** @endcond */
...
...
cppa/system_messages.hpp
View file @
5e5ccb08
...
...
@@ -76,6 +76,20 @@ struct down_msg {
*/
struct
sync_timeout_msg
{
};
/**
* @brief Sent whenever a terminated actor receives a synchronous request.
*/
struct
sync_exited_msg
{
/**
* @brief The source of this message, i.e., the terminated actor.
*/
actor_addr
source
;
/**
* @brief The exit reason of the terminated actor.
*/
std
::
uint32_t
reason
;
};
/**
* @brief Signalizes a timeout event.
* @note This message is handled implicitly by the runtime system.
...
...
cppa/typed_actor.hpp
View file @
5e5ccb08
...
...
@@ -31,19 +31,26 @@
#ifndef CPPA_TYPED_ACTOR_HPP
#define CPPA_TYPED_ACTOR_HPP
#include "cppa/actor_addr.hpp"
#include "cppa/replies_to.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/typed_event_based_actor.hpp"
namespace
cppa
{
class
actor_addr
;
class
local_actor
;
struct
invalid_actor_addr_t
;
template
<
typename
...
Signatures
>
class
typed_actor
{
class
typed_actor
:
util
::
comparable
<
typed_actor
<
Signatures
...
>>
,
util
::
comparable
<
typed_actor
<
Signatures
...
>
,
actor_addr
>
,
util
::
comparable
<
typed_actor
<
Signatures
...
>
,
invalid_actor_addr_t
>
{
public:
friend
class
local_actor
;
typedef
typed_event_based_actor
<
Signatures
...
>
actor_type
;
public:
typedef
util
::
type_list
<
Signatures
...
>
signatures
;
...
...
@@ -69,6 +76,21 @@ class typed_actor {
set
(
other
);
}
/**
* @brief Queries the address of the stored actor.
*/
actor_addr
address
()
const
{
return
m_ptr
?
m_ptr
->
address
()
:
actor_addr
{};
}
inline
intptr_t
compare
(
const
actor_addr
&
rhs
)
const
{
return
address
().
compare
(
rhs
);
}
inline
intptr_t
compare
(
const
invalid_actor_addr_t
&
)
const
{
return
m_ptr
?
1
:
0
;
}
private:
template
<
class
ListA
,
class
ListB
>
...
...
cppa/typed_behavior_stack_based.hpp
View file @
5e5ccb08
...
...
@@ -102,7 +102,7 @@ class typed_behavior_stack_based : public extend<Base>::template
return
m_bhvr_stack
;
}
inline
optional
<
behavior
&>
sync_handler
(
message_id
msg_id
)
{
optional
<
behavior
&>
sync_handler
(
message_id
msg_id
)
override
{
return
m_bhvr_stack
.
sync_handler
(
msg_id
);
}
...
...
src/blocking_actor.cpp
View file @
5e5ccb08
...
...
@@ -37,31 +37,10 @@
namespace
cppa
{
void
blocking_actor
::
response_handle
::
await
(
behavior
&
bhvr
)
{
m_self
->
dequeue_response
(
bhvr
,
m_mid
);
}
void
blocking_actor
::
await_all_other_actors_done
()
{
get_actor_registry
()
->
await_running_count_equal
(
1
);
}
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_handle
blocking_actor
::
timed_sync_send_tuple
(
const
util
::
duration
&
rtime
,
const
actor
&
dest
,
any_tuple
what
)
{
return
{
timed_sync_send_tuple_impl
(
message_priority
::
normal
,
dest
,
rtime
,
std
::
move
(
what
)),
this
};
}
void
blocking_actor
::
quit
(
std
::
uint32_t
reason
)
{
planned_exit_reason
(
reason
);
throw
actor_exited
(
reason
);
...
...
src/continue_helper.cpp
0 → 100644
View file @
5e5ccb08
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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/continue_helper.hpp"
#include "cppa/event_based_actor.hpp"
namespace
cppa
{
continue_helper
::
continue_helper
(
message_id
mid
,
local_actor
*
self
)
:
m_mid
(
mid
),
m_self
(
self
)
{
}
continue_helper
&
continue_helper
::
continue_with
(
behavior
::
continuation_fun
f
)
{
auto
ref_opt
=
m_self
->
sync_handler
(
m_mid
);
if
(
ref_opt
)
{
behavior
cpy
=
*
ref_opt
;
*
ref_opt
=
cpy
.
add_continuation
(
std
::
move
(
f
));
}
else
{
CPPA_LOG_ERROR
(
"failed to add continuation"
);
}
return
*
this
;
}
}
// namespace cppa
src/event_based_actor.cpp
View file @
5e5ccb08
...
...
@@ -35,39 +35,8 @@
namespace
cppa
{
continue_helper
&
continue_helper
::
continue_with
(
behavior
::
continuation_fun
fun
)
{
auto
ref_opt
=
m_self
->
bhvr_stack
().
sync_handler
(
m_mid
);
if
(
ref_opt
)
{
behavior
cpy
=
*
ref_opt
;
*
ref_opt
=
cpy
.
add_continuation
(
std
::
move
(
fun
));
}
else
{
CPPA_LOG_ERROR
(
"failed to add continuation"
);
}
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_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_handle
event_based_actor
::
timed_sync_send_tuple
(
const
util
::
duration
&
rtime
,
const
actor
&
dest
,
any_tuple
what
)
{
return
{
timed_sync_send_tuple_impl
(
message_priority
::
normal
,
dest
,
rtime
,
std
::
move
(
what
)),
this
};
}
}
// namespace cppa
src/local_actor.cpp
View file @
5e5ccb08
...
...
@@ -147,18 +147,18 @@ void local_actor::send_tuple(message_priority prio, const channel& dest, any_tup
dest
.
enqueue
({
address
(),
dest
,
id
},
std
::
move
(
what
));
}
void
local_actor
::
send_tuple
(
const
channel
&
dest
,
any_tuple
what
)
{
send_tuple
(
message_priority
::
normal
,
dest
,
std
::
move
(
what
));
}
void
local_actor
::
send_exit
(
const
actor_addr
&
whom
,
std
::
uint32_t
reason
)
{
send
(
detail
::
raw_access
::
get
(
whom
),
exit_msg
{
address
(),
reason
});
}
void
local_actor
::
delayed_send_tuple
(
const
channel
&
dest
,
void
local_actor
::
delayed_send_tuple
(
message_priority
prio
,
const
channel
&
dest
,
const
util
::
duration
&
rel_time
,
cppa
::
any_tuple
msg
)
{
get_scheduler
()
->
delayed_send
({
address
(),
dest
},
rel_time
,
std
::
move
(
msg
));
message_id
mid
;
if
(
prio
==
message_priority
::
high
)
mid
=
mid
.
with_high_priority
();
get_scheduler
()
->
delayed_send
({
address
(),
dest
,
mid
},
rel_time
,
std
::
move
(
msg
));
}
response_promise
local_actor
::
make_response_promise
()
{
...
...
@@ -205,6 +205,15 @@ message_id local_actor::timed_sync_send_tuple_impl(message_priority mp,
return
rri
;
}
message_id
local_actor
::
sync_send_tuple_impl
(
message_priority
mp
,
const
actor
&
dest
,
any_tuple
&&
what
)
{
auto
nri
=
new_request_id
();
if
(
mp
==
message_priority
::
high
)
nri
=
nri
.
with_high_priority
();
dest
->
enqueue
({
address
(),
dest
,
nri
},
std
::
move
(
what
));
return
nri
.
response_id
();
}
void
anon_send_exit
(
const
actor_addr
&
whom
,
std
::
uint32_t
reason
)
{
auto
ptr
=
detail
::
raw_access
::
get
(
whom
);
ptr
->
enqueue
({
invalid_actor_addr
,
ptr
,
message_id
{}.
with_high_priority
()},
...
...
src/sync_request_bouncer.cpp
View file @
5e5ccb08
...
...
@@ -33,6 +33,7 @@
#include "cppa/message_id.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/system_messages.hpp"
#include "cppa/detail/raw_access.hpp"
#include "cppa/detail/sync_request_bouncer.hpp"
...
...
@@ -48,7 +49,7 @@ void sync_request_bouncer::operator()(const actor_addr& sender,
if
(
sender
&&
mid
.
is_request
())
{
auto
ptr
=
detail
::
raw_access
::
get
(
sender
);
ptr
->
enqueue
({
invalid_actor_addr
,
ptr
,
mid
.
response_id
()},
make_any_tuple
(
atom
(
"EXITED"
),
rsn
));
make_any_tuple
(
sync_exited_msg
{
sender
,
rsn
}
));
}
}
...
...
src/uniform_type_info_map.cpp
View file @
5e5ccb08
...
...
@@ -73,6 +73,7 @@ namespace cppa { namespace detail {
{
"cppa::io::accept_handle"
,
"@ac_hdl"
},
{
"cppa::io::connection_handle"
,
"@cn_hdl"
},
{
"cppa::message_header"
,
"@header"
},
{
"cppa::sync_exited_msg"
,
"@sync_exited"
},
{
"cppa::sync_timeout_msg"
,
"@sync_timeout"
},
{
"cppa::timeout_msg"
,
"@timeout"
},
{
"cppa::unit_t"
,
"@0"
},
...
...
@@ -391,7 +392,9 @@ inline void deserialize_impl(bool& val, deserializer* source) {
// exit_msg & down_msg have the same fields
template
<
typename
T
>
typename
std
::
enable_if
<
std
::
is_same
<
T
,
down_msg
>::
value
||
std
::
is_same
<
T
,
exit_msg
>::
value
std
::
is_same
<
T
,
down_msg
>::
value
||
std
::
is_same
<
T
,
exit_msg
>::
value
||
std
::
is_same
<
T
,
sync_exited_msg
>::
value
>::
type
serialize_impl
(
const
T
&
dm
,
serializer
*
sink
)
{
serialize_impl
(
dm
.
source
,
sink
);
...
...
@@ -401,7 +404,9 @@ serialize_impl(const T& dm, serializer* sink) {
// exit_msg & down_msg have the same fields
template
<
typename
T
>
typename
std
::
enable_if
<
std
::
is_same
<
T
,
down_msg
>::
value
||
std
::
is_same
<
T
,
exit_msg
>::
value
std
::
is_same
<
T
,
down_msg
>::
value
||
std
::
is_same
<
T
,
exit_msg
>::
value
||
std
::
is_same
<
T
,
sync_exited_msg
>::
value
>::
type
deserialize_impl
(
T
&
dm
,
deserializer
*
source
)
{
deserialize_impl
(
dm
.
source
,
source
);
...
...
@@ -782,6 +787,7 @@ class utim_impl : public uniform_type_info_map {
*
i
++
=
&
m_type_proc
;
// @proc
*
i
++
=
&
m_type_str
;
// @str
*
i
++
=
&
m_type_strmap
;
// @strmap
*
i
++
=
&
m_type_sync_exited
;
// @sync_exited
*
i
++
=
&
m_type_sync_timeout
;
// @sync_timeout
*
i
++
=
&
m_type_timeout
;
// @timeout
*
i
++
=
&
m_type_tuple
;
// @tuple
...
...
@@ -903,6 +909,7 @@ class utim_impl : public uniform_type_info_map {
// 10-19
uti_impl
<
any_tuple
>
m_type_tuple
;
uti_impl
<
util
::
duration
>
m_type_duration
;
uti_impl
<
sync_exited_msg
>
m_type_sync_exited
;
uti_impl
<
sync_timeout_msg
>
m_type_sync_timeout
;
uti_impl
<
timeout_msg
>
m_type_timeout
;
uti_impl
<
message_header
>
m_type_header
;
...
...
@@ -910,9 +917,9 @@ class utim_impl : public uniform_type_info_map {
uti_impl
<
atom_value
>
m_type_atom
;
uti_impl
<
std
::
string
>
m_type_str
;
uti_impl
<
std
::
u16string
>
m_type_u16str
;
uti_impl
<
std
::
u32string
>
m_type_u32str
;
// 20-29
uti_impl
<
std
::
u32string
>
m_type_u32str
;
default_uniform_type_info_impl
<
strmap
>
m_type_strmap
;
uti_impl
<
bool
>
m_type_bool
;
uti_impl
<
float
>
m_type_float
;
...
...
@@ -922,15 +929,15 @@ class utim_impl : public uniform_type_info_map {
int_tinfo
<
std
::
uint8_t
>
m_type_u8
;
int_tinfo
<
std
::
int16_t
>
m_type_i16
;
int_tinfo
<
std
::
uint16_t
>
m_type_u16
;
int_tinfo
<
std
::
int32_t
>
m_type_i32
;
// 30-32
// 30-33
int_tinfo
<
std
::
int32_t
>
m_type_i32
;
int_tinfo
<
std
::
uint32_t
>
m_type_u32
;
int_tinfo
<
std
::
int64_t
>
m_type_i64
;
int_tinfo
<
std
::
uint64_t
>
m_type_u64
;
// both containers are sorted by uniform name
std
::
array
<
pointer
,
3
3
>
m_builtin_types
;
std
::
array
<
pointer
,
3
4
>
m_builtin_types
;
std
::
vector
<
uniform_type_info
*>
m_user_types
;
mutable
util
::
shared_spinlock
m_lock
;
...
...
unit_testing/test_spawn.cpp
View file @
5e5ccb08
...
...
@@ -675,7 +675,7 @@ void test_spawn() {
}
);
// must skip remaining async message
self
->
receive_response
(
handle
)
(
handle
.
await
(
on_arg_match
>>
[
&
](
int
a
,
int
b
)
{
CPPA_CHECK_EQUAL
(
a
,
42
);
CPPA_CHECK_EQUAL
(
b
,
2
);
...
...
@@ -738,7 +738,7 @@ void test_spawn() {
CPPA_CHECKPOINT
();
self
->
sync_send
(
sync_testee
,
"!?"
).
await
(
on
(
atom
(
"EXITED"
),
any_vals
)
>>
CPPA_CHECKPOINT_CB
(),
on
<
sync_exited_msg
>
(
)
>>
CPPA_CHECKPOINT_CB
(),
others
()
>>
CPPA_UNEXPECTED_MSG_CB
(),
after
(
chrono
::
milliseconds
(
5
))
>>
CPPA_UNEXPECTED_TOUT_CB
()
);
...
...
unit_testing/test_sync_send.cpp
View file @
5e5ccb08
...
...
@@ -281,8 +281,8 @@ void test_sync_send() {
// test use case 3
self
->
spawn
<
monitored
+
blocking_api
>
([](
blocking_actor
*
self
)
{
// client
auto
s
=
self
->
spawn
<
server
,
linked
>
();
// server
auto
w
=
self
->
spawn
<
linked
>
([](
event_based_actor
*
self
)
{
// worker
auto
s
=
self
->
spawn
<
server
,
linked
>
();
// server
auto
w
=
self
->
spawn
<
linked
>
([](
event_based_actor
*
self
)
{
// worker
self
->
become
(
on
(
atom
(
"request"
))
>>
[]{
return
atom
(
"response"
);
});
});
// first 'idle', then 'request'
...
...
@@ -300,7 +300,7 @@ void test_sync_send() {
// first 'request', then 'idle'
auto
handle
=
self
->
sync_send
(
s
,
atom
(
"request"
));
send_as
(
w
,
s
,
atom
(
"idle"
));
self
->
receive_response
(
handle
)
(
handle
.
await
(
on
(
atom
(
"response"
))
>>
[
=
]
{
CPPA_CHECKPOINT
();
CPPA_CHECK_EQUAL
(
self
->
last_sender
(),
w
);
...
...
unit_testing/test_typed_spawn.cpp
View file @
5e5ccb08
...
...
@@ -41,6 +41,14 @@ bool operator==(const my_request& lhs, const my_request& rhs) {
return
lhs
.
a
==
rhs
.
a
&&
lhs
.
b
==
rhs
.
b
;
}
typed_behavior
<
replies_to
<
my_request
>::
with
<
bool
>>
typed_server
()
{
return
{
on_arg_match
>>
[](
const
my_request
&
req
)
{
return
req
.
a
==
req
.
b
;
}
};
}
/*
typed_actor_ptr<replies_to<my_request>::with<bool>>
spawn_typed_server() {
...
...
@@ -66,10 +74,31 @@ class typed_testee : public typed_actor<replies_to<my_request>::with<bool>> {
};
*/
void
test_typed_spawn
()
{
auto
ts
=
spawn_typed
(
typed_server
);
scoped_actor
self
;
self
->
send
(
ts
,
my_request
{
1
,
2
});
self
->
receive
(
on_arg_match
>>
[](
bool
value
)
{
CPPA_CHECK_EQUAL
(
value
,
false
);
}
);
self
->
send
(
ts
,
my_request
{
42
,
42
});
self
->
receive
(
on_arg_match
>>
[](
bool
value
)
{
CPPA_CHECK_EQUAL
(
value
,
true
);
}
);
self
->
send_exit
(
ts
,
exit_reason
::
user_shutdown
);
}
int
main
()
{
CPPA_TEST
(
test_typed_spawn
);
/*
announce
<
my_request
>
(
&
my_request
::
a
,
&
my_request
::
b
);
test_typed_spawn
();
await_all_actors_done
();
shutdown
();
/*
auto sptr = spawn_typed_server();
sync_send(sptr, my_request{2, 2}).await(
[](bool value) {
...
...
unit_testing/test_uniform_type.cpp
View file @
5e5ccb08
...
...
@@ -96,6 +96,7 @@ int main() {
"@down"
,
// down_msg
"@exit"
,
// exit_msg
"@timeout"
,
// timeout_msg
"@sync_exited"
,
// sync_exited_msg
"@sync_timeout"
,
// sync_timeout_msg
// default announced cppa tuples
"@<>+@atom"
,
// {atom_value}
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment