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
6395e33e
Commit
6395e33e
authored
Jan 07, 2016
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Plain Diff
Merge branch 'Lingxi-Li-topic/actor-decorator'
parents
1bd98d28
a000469e
Changes
9
Hide whitespace changes
Inline
Side-by-side
Showing
9 changed files
with
708 additions
and
256 deletions
+708
-256
libcaf_core/caf/bound_actor.hpp
libcaf_core/caf/bound_actor.hpp
+14
-10
libcaf_core/caf/composed_actor.hpp
libcaf_core/caf/composed_actor.hpp
+22
-8
libcaf_core/caf/typed_actor.hpp
libcaf_core/caf/typed_actor.hpp
+11
-7
libcaf_core/src/actor.cpp
libcaf_core/src/actor.cpp
+5
-4
libcaf_core/src/bound_actor.cpp
libcaf_core/src/bound_actor.cpp
+74
-26
libcaf_core/src/composed_actor.cpp
libcaf_core/src/composed_actor.cpp
+80
-34
libcaf_core/test/bound_actor.cpp
libcaf_core/test/bound_actor.cpp
+272
-0
libcaf_core/test/composed_actor.cpp
libcaf_core/test/composed_actor.cpp
+230
-0
libcaf_core/test/typed_spawn.cpp
libcaf_core/test/typed_spawn.cpp
+0
-167
No files found.
libcaf_core/caf/bound_actor.hpp
View file @
6395e33e
...
@@ -23,25 +23,29 @@
...
@@ -23,25 +23,29 @@
#include "caf/message.hpp"
#include "caf/message.hpp"
#include "caf/actor_addr.hpp"
#include "caf/actor_addr.hpp"
#include "caf/attachable.hpp"
#include "caf/attachable.hpp"
#include "caf/
abstract
_actor.hpp"
#include "caf/
monitorable
_actor.hpp"
namespace
caf
{
namespace
caf
{
/// An actor decorator implementing `std::bind`-like compositions.
/// An actor decorator implementing `std::bind`-like compositions.
class
bound_actor
:
public
abstract_actor
{
/// Bound actors are hidden actors. A bound actor exits when its
/// decorated actor exits. The decorated actor has no dependency
/// on the bound actor by default, and exit of a bound actor has
/// no effect on the decorated actor. Bound actors are hosted on
/// the same actor system and node as decorated actors.
class
bound_actor
:
public
monitorable_actor
{
public:
public:
bound_actor
(
actor_system
*
sys
,
actor_addr
decorated
,
message
msg
);
bound_actor
(
actor_addr
decorated
,
message
msg
);
void
attach
(
attachable_ptr
ptr
)
override
;
size_t
detach
(
const
attachable
::
token
&
what
)
override
;
// non-system messages are processed and then forwarded;
// system messages are handled and consumed on the spot;
// in either case, the processing is done synchronously
void
enqueue
(
mailbox_element_ptr
what
,
execution_unit
*
host
)
override
;
void
enqueue
(
mailbox_element_ptr
what
,
execution_unit
*
host
)
override
;
protected:
bool
link_impl
(
linking_operation
op
,
const
actor_addr
&
other
)
override
;
private:
private:
void
handle_system_message
(
const
message
&
msg
,
execution_unit
*
host
);
static
bool
is_system_message
(
const
message
&
msg
);
actor_addr
decorated_
;
actor_addr
decorated_
;
message
merger_
;
message
merger_
;
};
};
...
...
libcaf_core/caf/composed_actor.hpp
View file @
6395e33e
...
@@ -20,25 +20,39 @@
...
@@ -20,25 +20,39 @@
#ifndef CAF_COMPOSED_ACTOR_HPP
#ifndef CAF_COMPOSED_ACTOR_HPP
#define CAF_COMPOSED_ACTOR_HPP
#define CAF_COMPOSED_ACTOR_HPP
#include "caf/local_actor.hpp"
#include "caf/actor_addr.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/monitorable_actor.hpp"
namespace
caf
{
namespace
caf
{
/// An actor decorator implementing "dot operator"-like compositions,
/// An actor decorator implementing "dot operator"-like compositions,
/// i.e., `f.g(x) = f(g(x))`.
/// i.e., `f.g(x) = f(g(x))`. Composed actors are hidden actors.
class
composed_actor
:
public
local_actor
{
/// A composed actor exits when either of its constituent actors exits;
/// Constituent actors have no dependency on the composed actor
/// by default, and exit of a composed actor has no effect on its
/// constituent actors. A composed actor is hosted on the same actor
/// system and node as `g`, the first actor on the forwarding chain.
class
composed_actor
:
public
monitorable_actor
{
public:
public:
composed_actor
(
actor_system
*
sys
,
actor_addr
first
,
actor_addr
second
)
;
using
message_types_set
=
std
::
set
<
std
::
string
>
;
void
initialize
()
override
;
composed_actor
(
actor_addr
f
,
actor_addr
g
,
message_types_set
msg_types
)
;
// non-system messages are processed and then forwarded;
// system messages are handled and consumed on the spot;
// in either case, the processing is done synchronously
void
enqueue
(
mailbox_element_ptr
what
,
execution_unit
*
host
)
override
;
void
enqueue
(
mailbox_element_ptr
what
,
execution_unit
*
host
)
override
;
message_types_set
message_types
()
const
override
;
private:
private:
bool
is_system_message
(
const
message
&
msg
);
void
handle_system_message
(
const
message
&
msg
,
execution_unit
*
host
);
static
bool
is_system_message
(
const
message
&
msg
);
actor_addr
first_
;
actor_addr
f_
;
actor_addr
second_
;
actor_addr
g_
;
message_types_set
msg_types_
;
};
};
}
// namespace caf
}
// namespace caf
...
...
libcaf_core/caf/typed_actor.hpp
View file @
6395e33e
...
@@ -27,6 +27,7 @@
...
@@ -27,6 +27,7 @@
#include "caf/replies_to.hpp"
#include "caf/replies_to.hpp"
#include "caf/bound_actor.hpp"
#include "caf/bound_actor.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/composed_actor.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/typed_response_promise.hpp"
...
@@ -214,7 +215,7 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
...
@@ -214,7 +215,7 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
bind
(
Ts
&&
...
xs
)
const
{
bind
(
Ts
&&
...
xs
)
const
{
if
(
!
ptr_
)
if
(
!
ptr_
)
return
invalid_actor
;
return
invalid_actor
;
auto
ptr
=
make_counted
<
bound_actor
>
(
&
ptr_
->
home_system
(),
ptr_
->
address
(),
auto
ptr
=
make_counted
<
bound_actor
>
(
ptr_
->
address
(),
make_message
(
xs
...));
make_message
(
xs
...));
return
{
ptr
.
release
(),
false
};
return
{
ptr
.
release
(),
false
};
}
}
...
@@ -287,18 +288,21 @@ bool operator!=(const typed_actor<Xs...>& x,
...
@@ -287,18 +288,21 @@ bool operator!=(const typed_actor<Xs...>& x,
template
<
class
...
Xs
,
class
...
Ys
>
template
<
class
...
Xs
,
class
...
Ys
>
typename
detail
::
mpi_composition
<
typename
detail
::
mpi_composition
<
typed_actor
,
typed_actor
,
detail
::
type_list
<
Y
s
...
>
,
detail
::
type_list
<
X
s
...
>
,
X
s
...
Y
s
...
>::
type
>::
type
operator
*
(
typed_actor
<
Xs
...
>
f
,
typed_actor
<
Ys
...
>
g
)
{
operator
*
(
typed_actor
<
Xs
...
>
f
,
typed_actor
<
Ys
...
>
g
)
{
using
result
=
using
result
=
typename
detail
::
mpi_composition
<
typename
detail
::
mpi_composition
<
typed_actor
,
typed_actor
,
detail
::
type_list
<
Y
s
...
>
,
detail
::
type_list
<
X
s
...
>
,
X
s
...
Y
s
...
>::
type
;
>::
type
;
return
actor_cast
<
result
>
(
actor_cast
<
actor
>
(
std
::
move
(
f
))
auto
ptr
=
make_counted
<
composed_actor
>
(
f
.
address
(),
*
actor_cast
<
actor
>
(
std
::
move
(
g
)));
g
.
address
(),
g
->
home_system
().
message_types
(
result
{}));
return
actor_cast
<
result
>
(
std
::
move
(
ptr
));
}
}
/// @relates typed_actor
/// @relates typed_actor
...
...
libcaf_core/src/actor.cpp
View file @
6395e33e
...
@@ -78,16 +78,17 @@ actor_id actor::id() const noexcept {
...
@@ -78,16 +78,17 @@ actor_id actor::id() const noexcept {
actor
actor
::
bind_impl
(
message
msg
)
const
{
actor
actor
::
bind_impl
(
message
msg
)
const
{
if
(
!
ptr_
)
if
(
!
ptr_
)
return
invalid_actor
;
return
invalid_actor
;
return
actor_cast
<
actor
>
(
make_counted
<
bound_actor
>
(
&
ptr_
->
home_system
(),
return
actor_cast
<
actor
>
(
make_counted
<
bound_actor
>
(
address
(),
address
(),
std
::
move
(
msg
)));
std
::
move
(
msg
)));
}
}
actor
operator
*
(
actor
f
,
actor
g
)
{
actor
operator
*
(
actor
f
,
actor
g
)
{
if
(
!
f
||
!
g
)
if
(
!
f
||
!
g
)
return
invalid_actor
;
return
invalid_actor
;
auto
ptr
=
make_counted
<
composed_actor
>
(
&
f
->
home_system
(),
auto
ptr
=
make_counted
<
composed_actor
>
(
f
.
address
(),
f
.
address
(),
g
.
address
());
g
.
address
(),
g
->
home_system
().
message_types
(
actor
{}));
return
actor_cast
<
actor
>
(
std
::
move
(
ptr
));
return
actor_cast
<
actor
>
(
std
::
move
(
ptr
));
}
}
...
...
libcaf_core/src/bound_actor.cpp
View file @
6395e33e
...
@@ -20,47 +20,95 @@
...
@@ -20,47 +20,95 @@
#include "caf/bound_actor.hpp"
#include "caf/bound_actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp"
#include "caf/default_attachable.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/system_messages.hpp"
#include "caf/system_messages.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/merged_tuple.hpp"
#include "caf/detail/merged_tuple.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
namespace
caf
{
namespace
caf
{
bound_actor
::
bound_actor
(
actor_system
*
sys
,
actor_addr
decorated
,
message
msg
)
bound_actor
::
bound_actor
(
actor_addr
decorated
,
message
msg
)
:
abstract_actor
(
sys
,
decorated
->
id
(),
decorated
->
node
(),
:
monitorable_actor
{
&
decorated
->
home_system
(),
is_abstract_actor_flag
|
is_actor_bind_decorator_flag
),
decorated
->
home_system
().
next_actor_id
(),
decorated_
(
std
::
move
(
decorated
)),
decorated
->
node
(),
merger_
(
std
::
move
(
msg
))
{
is_abstract_actor_flag
// nop
|
is_actor_bind_decorator_flag
},
decorated_
{
std
::
move
(
decorated
)
},
merger_
{
std
::
move
(
msg
)
}
{
// bound actor has dependency on the decorated actor by default;
// if the decorated actor is already dead upon establishing the
// dependency, the actor is spawned dead
decorated_
->
attach
(
default_attachable
::
make_monitor
(
address
()));
}
}
void
bound_actor
::
attach
(
attachable_ptr
ptr
)
{
void
bound_actor
::
enqueue
(
mailbox_element_ptr
what
,
decorated_
->
attach
(
std
::
move
(
ptr
));
execution_unit
*
host
)
{
}
if
(
!
what
)
return
;
// not even an empty message
size_t
bound_actor
::
detach
(
const
attachable
::
token
&
what
)
{
auto
reason
=
exit_reason_
.
load
();
return
decorated_
->
detach
(
what
);
if
(
reason
!=
exit_reason
::
not_exited
)
{
// actor has exited
auto
&
mid
=
what
->
mid
;
if
(
mid
.
is_request
())
{
// make sure that a request always gets a response;
// the exit reason reflects the first actor on the
// forwarding chain that is out of service
detail
::
sync_request_bouncer
rb
{
reason
};
rb
(
what
->
sender
,
mid
);
}
return
;
}
if
(
is_system_message
(
what
->
msg
))
{
// handle and consume the system message;
// the only effect that MAY result from handling a system message
// is to exit the actor if it hasn't exited already;
// `handle_system_message()` is thread-safe, and if the actor
// has already exited upon the invocation, nothing is done
handle_system_message
(
what
->
msg
,
host
);
}
else
{
// process and forward the non-system message
auto
&
msg
=
what
->
msg
;
message
tmp
{
detail
::
merged_tuple
::
make
(
merger_
,
std
::
move
(
msg
))
};
msg
.
swap
(
tmp
);
decorated_
->
enqueue
(
std
::
move
(
what
),
host
);
}
}
}
void
bound_actor
::
enqueue
(
mailbox_element_ptr
what
,
execution_unit
*
host
)
{
void
bound_actor
::
handle_system_message
(
const
message
&
msg
,
// ignore system messages
execution_unit
*
host
)
{
auto
&
msg
=
what
->
msg
;
// `monitorable_actor::cleanup()` is thread-safe, and if the actor
if
(
!
((
msg
.
size
()
==
1
&&
(
msg
.
match_element
<
exit_msg
>
(
0
)
// has already exited upon the invocation, nothing is done;
||
msg
.
match_element
<
down_msg
>
(
0
)))
// handles only `down_msg` from the decorated actor and `exit_msg` from anyone
||
(
msg
.
size
()
>
1
&&
msg
.
match_element
<
sys_atom
>
(
0
))))
{
if
(
msg
.
size
()
!=
1
)
// merge non-system messages
return
;
// neither case
message
tmp
{
detail
::
merged_tuple
::
make
(
merger_
,
std
::
move
(
msg
))};
if
(
msg
.
match_element
<
down_msg
>
(
0
))
{
msg
.
swap
(
tmp
);
auto
&
dm
=
msg
.
get_as
<
down_msg
>
(
0
);
CAF_ASSERT
(
dm
.
reason
!=
exit_reason
::
not_exited
);
if
(
dm
.
source
==
decorated_
)
{
// decorated actor has exited, so exits
// the bound actor with the same reason
monitorable_actor
::
cleanup
(
dm
.
reason
,
host
);
}
return
;
}
if
(
msg
.
match_element
<
exit_msg
>
(
0
))
{
auto
&
em
=
msg
.
get_as
<
exit_msg
>
(
0
);
CAF_ASSERT
(
em
.
reason
!=
exit_reason
::
not_exited
);
// exit message received, so exits with the same reason
monitorable_actor
::
cleanup
(
em
.
reason
,
host
);
return
;
}
}
decorated_
->
enqueue
(
std
::
move
(
what
),
host
);
// drop other cases
}
}
bool
bound_actor
::
link_impl
(
linking_operation
op
,
const
actor_addr
&
other
)
{
bool
bound_actor
::
is_system_message
(
const
message
&
msg
)
{
if
(
actor_cast
<
abstract_actor
*>
(
other
)
==
this
||
other
==
decorated_
)
return
(
msg
.
size
()
==
1
&&
(
msg
.
match_element
<
exit_msg
>
(
0
)
return
false
;
||
msg
.
match_element
<
down_msg
>
(
0
)))
return
decorated_
->
link_impl
(
op
,
other
);
||
(
msg
.
size
()
>
1
&&
msg
.
match_element
<
sys_atom
>
(
0
)
);
}
}
}
// namespace caf
}
// namespace caf
libcaf_core/src/composed_actor.cpp
View file @
6395e33e
...
@@ -19,53 +19,99 @@
...
@@ -19,53 +19,99 @@
#include "caf/composed_actor.hpp"
#include "caf/composed_actor.hpp"
namespace
caf
{
#include "caf/actor_system.hpp"
#include "caf/default_attachable.hpp"
composed_actor
::
composed_actor
(
actor_system
*
sys
,
#include "caf/detail/disposer.hpp"
actor_addr
first
,
actor_addr
second
)
#include "caf/detail/sync_request_bouncer.hpp"
:
local_actor
(
sys
,
first
->
id
(),
first
->
node
(),
is_abstract_actor_flag
|
is_actor_dot_decorator_flag
),
namespace
caf
{
first_
(
first
),
second_
(
second
)
{
// nop
}
void
composed_actor
::
initialize
()
{
composed_actor
::
composed_actor
(
actor_addr
f
,
actor_addr
g
,
trap_exit
(
true
);
message_types_set
msg_types
)
link_to
(
first_
);
:
monitorable_actor
{
&
g
->
home_system
(),
link_to
(
second_
);
g
->
home_system
().
next_actor_id
(),
do_become
(
g
->
node
(),
{
is_abstract_actor_flag
[
=
](
const
exit_msg
&
em
)
{
|
is_actor_dot_decorator_flag
},
if
(
em
.
source
==
first_
||
em
.
source
==
second_
)
f_
{
std
::
move
(
f
)
},
quit
(
em
.
reason
);
g_
{
std
::
move
(
g
)
},
else
if
(
em
.
reason
!=
exit_reason
::
normal
)
msg_types_
{
std
::
move
(
msg_types
)
}
{
quit
(
em
.
reason
);
// composed actor has dependency on constituent actors by default;
},
// if either constituent actor is already dead upon establishing
others
>>
[]
{
// the dependency, the actor is spawned dead
// drop
f_
->
attach
(
default_attachable
::
make_monitor
(
address
()));
}
if
(
g_
!=
f_
)
},
g_
->
attach
(
default_attachable
::
make_monitor
(
address
()));
true
);
}
}
void
composed_actor
::
enqueue
(
mailbox_element_ptr
what
,
execution_unit
*
host
)
{
void
composed_actor
::
enqueue
(
mailbox_element_ptr
what
,
execution_unit
*
host
)
{
if
(
!
what
)
if
(
!
what
)
return
;
// not even an empty message
auto
reason
=
exit_reason_
.
load
();
if
(
reason
!=
exit_reason
::
not_exited
)
{
// actor has exited
auto
&
mid
=
what
->
mid
;
if
(
mid
.
is_request
())
{
// make sure that a request always gets a response;
// the exit reason reflects the first actor on the
// forwarding chain that is out of service
detail
::
sync_request_bouncer
rb
{
reason
};
rb
(
what
->
sender
,
mid
);
}
return
;
return
;
}
if
(
is_system_message
(
what
->
msg
))
{
if
(
is_system_message
(
what
->
msg
))
{
local_actor
::
enqueue
(
std
::
move
(
what
),
host
);
// handle and consume the system message;
// the only effect that MAY result from handling a system message
// is to exit the actor if it hasn't exited already;
// `handle_system_message()` is thread-safe, and if the actor
// has already exited upon the invocation, nothing is done
handle_system_message
(
what
->
msg
,
host
);
}
else
{
// process and forward the non-system message;
// store `f` as the next stage in the forwarding chain
what
->
stages
.
push_back
(
f_
);
// forward modified message to `g`
g_
->
enqueue
(
std
::
move
(
what
),
host
);
}
}
composed_actor
::
message_types_set
composed_actor
::
message_types
()
const
{
return
msg_types_
;
}
void
composed_actor
::
handle_system_message
(
const
message
&
msg
,
execution_unit
*
host
)
{
// `monitorable_actor::cleanup()` is thread-safe, and if the actor
// has already exited upon the invocation, nothing is done;
// handles only `down_msg` from constituent actors and `exit_msg` from anyone
if
(
msg
.
size
()
!=
1
)
return
;
// neither case
if
(
msg
.
match_element
<
down_msg
>
(
0
))
{
auto
&
dm
=
msg
.
get_as
<
down_msg
>
(
0
);
CAF_ASSERT
(
dm
.
reason
!=
exit_reason
::
not_exited
);
if
(
dm
.
source
==
f_
||
dm
.
source
==
g_
)
{
// one of the constituent actors has exited, so
// exits the composed actor with the same reason
monitorable_actor
::
cleanup
(
dm
.
reason
,
host
);
}
return
;
}
if
(
msg
.
match_element
<
exit_msg
>
(
0
))
{
auto
&
em
=
msg
.
get_as
<
exit_msg
>
(
0
);
CAF_ASSERT
(
em
.
reason
!=
exit_reason
::
not_exited
);
// exit message received; exits with the same reason
monitorable_actor
::
cleanup
(
em
.
reason
,
host
);
return
;
return
;
}
}
// store second_ as the next stage in the forwarding chain
// drop other cases
what
->
stages
.
push_back
(
second_
);
// forward modified message to first_
first_
->
enqueue
(
std
::
move
(
what
),
host
);
}
}
bool
composed_actor
::
is_system_message
(
const
message
&
msg
)
{
bool
composed_actor
::
is_system_message
(
const
message
&
msg
)
{
return
(
msg
.
size
()
==
1
&&
(
msg
.
match_element
<
exit_msg
>
(
0
)
return
(
msg
.
size
()
==
1
&&
(
msg
.
match_element
<
exit_msg
>
(
0
)
||
msg
.
match_element
<
down_msg
>
(
0
)))
||
msg
.
match_element
<
down_msg
>
(
0
)))
||
(
msg
.
size
()
>
1
&&
msg
.
match_element
<
sys_atom
>
(
0
));
||
(
msg
.
size
()
>
1
&&
msg
.
match_element
<
sys_atom
>
(
0
));
}
}
...
...
libcaf_core/test/bound_actor.cpp
0 → 100644
View file @
6395e33e
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE bound_actor
#include "caf/test/unit_test.hpp"
#include "caf/all.hpp"
using
namespace
caf
;
namespace
{
behavior
dbl_bhvr
(
event_based_actor
*
self
)
{
return
{
[](
int
v
)
{
return
2
*
v
;
},
[
=
]
{
self
->
quit
();
}
};
}
struct
fixture
{
void
wait_until_exited
()
{
self
->
receive
(
[](
const
down_msg
&
)
{
CAF_CHECK
(
true
);
}
);
}
template
<
class
Actor
>
static
bool
exited
(
const
Actor
&
handle
)
{
auto
ptr
=
actor_cast
<
abstract_actor
*>
(
handle
);
auto
dptr
=
dynamic_cast
<
monitorable_actor
*>
(
ptr
);
CAF_REQUIRE
(
dptr
!=
nullptr
);
return
dptr
->
exited
();
}
actor_system
system
;
scoped_actor
self
{
system
,
true
};
};
}
// namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE
(
bound_actor_tests
,
fixture
)
CAF_TEST
(
identity
)
{
auto
dbl
=
system
.
spawn
(
dbl_bhvr
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
auto
bound
=
dbl
.
bind
(
1
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
CAF_CHECK
(
&
bound
->
home_system
()
==
&
dbl
->
home_system
());
CAF_CHECK
(
bound
->
node
()
==
dbl
->
node
());
CAF_CHECK
(
bound
!=
dbl
);
CAF_CHECK
(
bound
->
id
()
!=
dbl
->
id
());
anon_send_exit
(
bound
,
exit_reason
::
kill
);
anon_send_exit
(
dbl
,
exit_reason
::
kill
);
}
// bound actor spawned dead if decorated
// actor is already dead upon spawning
CAF_TEST
(
lifetime_1
)
{
auto
dbl
=
system
.
spawn
(
dbl_bhvr
);
self
->
monitor
(
dbl
);
anon_send_exit
(
dbl
,
exit_reason
::
kill
);
wait_until_exited
();
auto
bound
=
dbl
.
bind
(
1
);
CAF_CHECK
(
exited
(
bound
));
}
// bound actor exits when decorated actor exits
CAF_TEST
(
lifetime_2
)
{
auto
dbl
=
system
.
spawn
(
dbl_bhvr
);
auto
bound
=
dbl
.
bind
(
1
);
self
->
monitor
(
bound
);
anon_send
(
dbl
,
message
{});
wait_until_exited
();
}
// 1) ignores down message not from the decorated actor
// 2) exits by receiving an exit message
// 3) exit has no effect on decorated actor
CAF_TEST
(
lifetime_3
)
{
auto
dbl
=
system
.
spawn
(
dbl_bhvr
);
auto
bound
=
dbl
.
bind
(
1
);
anon_send
(
bound
,
down_msg
{
self
->
address
(),
exit_reason
::
kill
});
CAF_CHECK
(
!
exited
(
bound
));
self
->
monitor
(
bound
);
auto
em_sender
=
system
.
spawn
(
dbl_bhvr
);
em_sender
->
link_to
(
bound
->
address
());
anon_send_exit
(
em_sender
,
exit_reason
::
kill
);
wait_until_exited
();
self
->
request
(
dbl
,
1
).
receive
(
[](
int
v
)
{
CAF_CHECK
(
v
==
2
);
},
[](
error
)
{
CAF_CHECK
(
false
);
}
);
anon_send_exit
(
dbl
,
exit_reason
::
kill
);
}
CAF_TEST
(
request_response_promise
)
{
auto
dbl
=
system
.
spawn
(
dbl_bhvr
);
auto
bound
=
dbl
.
bind
(
1
);
anon_send_exit
(
bound
,
exit_reason
::
kill
);
CAF_CHECK
(
exited
(
bound
));
self
->
request
(
bound
,
message
{}).
receive
(
[](
int
)
{
CAF_CHECK
(
false
);
},
[](
error
err
)
{
CAF_CHECK
(
err
.
code
()
==
sec
::
request_receiver_down
);
}
);
anon_send_exit
(
dbl
,
exit_reason
::
kill
);
}
CAF_TEST
(
partial_currying
)
{
using
namespace
std
::
placeholders
;
auto
impl
=
[]()
->
behavior
{
return
{
[](
ok_atom
,
int
x
)
{
return
x
;
},
[](
ok_atom
,
double
x
)
{
return
x
;
}
};
};
auto
aut
=
system
.
spawn
(
impl
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
auto
bound
=
aut
.
bind
(
ok_atom
::
value
,
_1
);
CAF_CHECK
(
aut
.
id
()
!=
bound
.
id
());
CAF_CHECK
(
aut
.
node
()
==
bound
.
node
());
CAF_CHECK
(
aut
!=
bound
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
self
->
request
(
bound
,
2.0
).
receive
(
[](
double
y
)
{
CAF_CHECK
(
y
==
2.0
);
}
);
self
->
request
(
bound
,
10
).
receive
(
[](
int
y
)
{
CAF_CHECK
(
y
==
10
);
}
);
self
->
send_exit
(
aut
,
exit_reason
::
kill
);
}
CAF_TEST
(
full_currying
)
{
auto
dbl_actor
=
system
.
spawn
(
dbl_bhvr
);
auto
bound
=
dbl_actor
.
bind
(
1
);
self
->
request
(
bound
,
message
{}).
receive
(
[](
int
v
)
{
CAF_CHECK
(
v
==
2
);
},
[](
error
err
)
{
CAF_CHECK
(
false
);
}
);
anon_send_exit
(
bound
,
exit_reason
::
kill
);
anon_send_exit
(
dbl_actor
,
exit_reason
::
kill
);
}
CAF_TEST
(
type_safe_currying
)
{
using
namespace
std
::
placeholders
;
using
testee
=
typed_actor
<
replies_to
<
ok_atom
,
int
>::
with
<
int
>
,
replies_to
<
ok_atom
,
double
>::
with
<
double
>>
;
auto
impl
=
[]()
->
testee
::
behavior_type
{
return
{
[](
ok_atom
,
int
x
)
{
return
x
;
},
[](
ok_atom
,
double
x
)
{
return
x
;
}
};
};
auto
aut
=
system
.
spawn
(
impl
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
using
curried_signature
=
typed_actor
<
replies_to
<
int
>::
with
<
int
>
,
replies_to
<
double
>::
with
<
double
>>
;
auto
bound
=
aut
.
bind
(
ok_atom
::
value
,
_1
);
CAF_CHECK
(
aut
!=
bound
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
static_assert
(
std
::
is_same
<
decltype
(
bound
),
curried_signature
>::
value
,
"bind returned wrong actor handle"
);
self
->
request
(
bound
,
2.0
).
receive
(
[](
double
y
)
{
CAF_CHECK
(
y
==
2.0
);
}
);
self
->
request
(
bound
,
10
).
receive
(
[](
int
y
)
{
CAF_CHECK
(
y
==
10
);
}
);
self
->
send_exit
(
aut
,
exit_reason
::
kill
);
}
CAF_TEST
(
reordering
)
{
auto
impl
=
[]()
->
behavior
{
return
{
[](
int
x
,
double
y
)
{
return
x
*
y
;
}
};
};
auto
aut
=
system
.
spawn
(
impl
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
using
namespace
std
::
placeholders
;
auto
bound
=
aut
.
bind
(
_2
,
_1
);
CAF_CHECK
(
aut
!=
bound
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
self
->
request
(
bound
,
2.0
,
10
).
receive
(
[](
double
y
)
{
CAF_CHECK
(
y
==
20.0
);
}
);
self
->
send_exit
(
aut
,
exit_reason
::
kill
);
}
CAF_TEST
(
type_safe_reordering
)
{
using
testee
=
typed_actor
<
replies_to
<
int
,
double
>::
with
<
double
>>
;
auto
impl
=
[]()
->
testee
::
behavior_type
{
return
{
[](
int
x
,
double
y
)
{
return
x
*
y
;
}
};
};
auto
aut
=
system
.
spawn
(
impl
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
using
namespace
std
::
placeholders
;
using
swapped_signature
=
typed_actor
<
replies_to
<
double
,
int
>::
with
<
double
>>
;
auto
bound
=
aut
.
bind
(
_2
,
_1
);
CAF_CHECK
(
aut
!=
bound
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
static_assert
(
std
::
is_same
<
decltype
(
bound
),
swapped_signature
>::
value
,
"bind returned wrong actor handle"
);
self
->
request
(
bound
,
2.0
,
10
).
receive
(
[](
double
y
)
{
CAF_CHECK
(
y
==
20.0
);
}
);
self
->
send_exit
(
aut
,
exit_reason
::
kill
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_core/test/composed_actor.cpp
0 → 100644
View file @
6395e33e
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE composed_actor
#include "caf/test/unit_test.hpp"
#include "caf/all.hpp"
using
namespace
caf
;
namespace
{
behavior
dbl_bhvr
(
event_based_actor
*
self
)
{
return
{
[](
int
v
)
{
return
2
*
v
;
},
[
=
]
{
self
->
quit
();
}
};
}
using
first_stage
=
typed_actor
<
replies_to
<
int
>::
with
<
double
,
double
>>
;
using
second_stage
=
typed_actor
<
replies_to
<
double
,
double
>::
with
<
double
>>
;
first_stage
::
behavior_type
first_stage_impl
()
{
return
[](
int
i
)
{
return
std
::
make_tuple
(
i
*
2.0
,
i
*
4.0
);
};
};
second_stage
::
behavior_type
second_stage_impl
()
{
return
[](
double
x
,
double
y
)
{
return
x
*
y
;
};
}
struct
fixture
{
void
wait_until_exited
()
{
self
->
receive
(
[](
const
down_msg
&
)
{
CAF_CHECK
(
true
);
}
);
}
template
<
class
Actor
>
static
bool
exited
(
const
Actor
&
handle
)
{
auto
ptr
=
actor_cast
<
abstract_actor
*>
(
handle
);
auto
dptr
=
dynamic_cast
<
monitorable_actor
*>
(
ptr
);
CAF_REQUIRE
(
dptr
!=
nullptr
);
return
dptr
->
exited
();
}
actor_system
system
;
scoped_actor
self
{
system
,
true
};
};
}
// namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE
(
bound_actor_tests
,
fixture
)
CAF_TEST
(
identity
)
{
actor_system
system_of_g
;
actor_system
system_of_f
;
auto
g
=
system_of_g
.
spawn
(
first_stage_impl
);
auto
f
=
system_of_f
.
spawn
(
second_stage_impl
);
CAF_CHECK
(
system_of_g
.
registry
().
running
()
==
1
);
auto
composed
=
f
*
g
;
CAF_CHECK
(
system_of_g
.
registry
().
running
()
==
1
);
CAF_CHECK
(
&
composed
->
home_system
()
==
&
g
->
home_system
());
CAF_CHECK
(
composed
->
node
()
==
g
->
node
());
CAF_CHECK
(
composed
->
id
()
!=
g
->
id
());
CAF_CHECK
(
composed
!=
g
);
CAF_CHECK
(
composed
->
message_types
()
==
g
->
home_system
().
message_types
(
composed
));
anon_send_exit
(
composed
,
exit_reason
::
kill
);
anon_send_exit
(
f
,
exit_reason
::
kill
);
anon_send_exit
(
g
,
exit_reason
::
kill
);
}
// spawned dead if `g` is already dead upon spawning
CAF_TEST
(
lifetime_1a
)
{
auto
g
=
system
.
spawn
(
dbl_bhvr
);
auto
f
=
system
.
spawn
(
dbl_bhvr
);
self
->
monitor
(
g
);
anon_send_exit
(
g
,
exit_reason
::
kill
);
wait_until_exited
();
auto
fg
=
f
*
g
;
CAF_CHECK
(
exited
(
fg
));
anon_send_exit
(
f
,
exit_reason
::
kill
);
}
// spawned dead if `f` is already dead upon spawning
CAF_TEST
(
lifetime_1b
)
{
auto
g
=
system
.
spawn
(
dbl_bhvr
);
auto
f
=
system
.
spawn
(
dbl_bhvr
);
self
->
monitor
(
f
);
anon_send_exit
(
f
,
exit_reason
::
kill
);
wait_until_exited
();
auto
fg
=
f
*
g
;
CAF_CHECK
(
exited
(
fg
));
anon_send_exit
(
g
,
exit_reason
::
kill
);
}
// `f.g` exits when `g` exits
CAF_TEST
(
lifetime_2a
)
{
auto
g
=
system
.
spawn
(
dbl_bhvr
);
auto
f
=
system
.
spawn
(
dbl_bhvr
);
auto
fg
=
f
*
g
;
self
->
monitor
(
fg
);
anon_send
(
g
,
message
{});
wait_until_exited
();
anon_send_exit
(
f
,
exit_reason
::
kill
);
}
// `f.g` exits when `f` exits
CAF_TEST
(
lifetime_2b
)
{
auto
g
=
system
.
spawn
(
dbl_bhvr
);
auto
f
=
system
.
spawn
(
dbl_bhvr
);
auto
fg
=
f
*
g
;
self
->
monitor
(
fg
);
anon_send
(
f
,
message
{});
wait_until_exited
();
anon_send_exit
(
g
,
exit_reason
::
kill
);
}
// 1) ignores down message not from constituent actors
// 2) exits by receiving an exit message
// 3) exit has no effect on constituent actors
CAF_TEST
(
lifetime_3
)
{
auto
g
=
system
.
spawn
(
dbl_bhvr
);
auto
f
=
system
.
spawn
(
dbl_bhvr
);
auto
fg
=
f
*
g
;
self
->
monitor
(
fg
);
anon_send
(
fg
,
down_msg
{
self
->
address
(),
exit_reason
::
kill
});
CAF_CHECK
(
!
exited
(
fg
));
auto
em_sender
=
system
.
spawn
(
dbl_bhvr
);
em_sender
->
link_to
(
fg
.
address
());
anon_send_exit
(
em_sender
,
exit_reason
::
kill
);
wait_until_exited
();
self
->
request
(
f
,
1
).
receive
(
[](
int
v
)
{
CAF_CHECK
(
v
==
2
);
},
[](
error
)
{
CAF_CHECK
(
false
);
}
);
self
->
request
(
g
,
1
).
receive
(
[](
int
v
)
{
CAF_CHECK
(
v
==
2
);
},
[](
error
)
{
CAF_CHECK
(
false
);
}
);
anon_send_exit
(
f
,
exit_reason
::
kill
);
anon_send_exit
(
g
,
exit_reason
::
kill
);
}
CAF_TEST
(
request_response_promise
)
{
auto
g
=
system
.
spawn
(
dbl_bhvr
);
auto
f
=
system
.
spawn
(
dbl_bhvr
);
auto
fg
=
f
*
g
;
anon_send_exit
(
fg
,
exit_reason
::
kill
);
CAF_CHECK
(
exited
(
fg
));
self
->
request
(
fg
,
1
).
receive
(
[](
int
)
{
CAF_CHECK
(
false
);
},
[](
error
err
)
{
CAF_CHECK
(
err
.
code
()
==
sec
::
request_receiver_down
);
}
);
anon_send_exit
(
f
,
exit_reason
::
kill
);
anon_send_exit
(
g
,
exit_reason
::
kill
);
}
// single composition of distinct actors
CAF_TEST
(
dot_composition_1
)
{
auto
first
=
system
.
spawn
(
first_stage_impl
);
auto
second
=
system
.
spawn
(
second_stage_impl
);
auto
first_then_second
=
second
*
first
;
self
->
request
(
first_then_second
,
42
).
receive
(
[](
double
res
)
{
CAF_CHECK
(
res
==
(
42
*
2.0
)
*
(
42
*
4.0
));
}
);
anon_send_exit
(
first
,
exit_reason
::
kill
);
anon_send_exit
(
second
,
exit_reason
::
kill
);
}
// multiple self composition
CAF_TEST
(
dot_composition_2
)
{
auto
dbl_actor
=
system
.
spawn
(
dbl_bhvr
);
auto
dbl_x4_actor
=
dbl_actor
*
dbl_actor
*
dbl_actor
*
dbl_actor
;
self
->
request
(
dbl_x4_actor
,
1
).
receive
(
[](
int
v
)
{
CAF_CHECK
(
v
==
16
);
},
[](
error
)
{
CAF_CHECK
(
false
);
}
);
anon_send_exit
(
dbl_actor
,
exit_reason
::
kill
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_core/test/typed_spawn.cpp
View file @
6395e33e
...
@@ -521,173 +521,6 @@ CAF_TEST(check_signature) {
...
@@ -521,173 +521,6 @@ CAF_TEST(check_signature) {
);
);
}
}
namespace
{
using
first_stage
=
typed_actor
<
replies_to
<
int
>::
with
<
double
,
double
>>
;
using
second_stage
=
typed_actor
<
replies_to
<
double
,
double
>::
with
<
double
>>
;
first_stage
::
behavior_type
first_stage_impl
()
{
return
[](
int
i
)
{
return
std
::
make_tuple
(
i
*
2.0
,
i
*
4.0
);
};
};
second_stage
::
behavior_type
second_stage_impl
()
{
return
[](
double
x
,
double
y
)
{
return
x
*
y
;
};
}
}
// namespace <anonymous>
CAF_TEST
(
dot_composition
)
{
actor_system
system
;
auto
first
=
system
.
spawn
(
first_stage_impl
);
auto
second
=
system
.
spawn
(
second_stage_impl
);
auto
first_then_second
=
first
*
second
;
scoped_actor
self
{
system
};
self
->
request
(
first_then_second
,
42
).
receive
(
[](
double
res
)
{
CAF_CHECK
(
res
==
(
42
*
2.0
)
*
(
42
*
4.0
));
}
);
anon_send_exit
(
first
,
exit_reason
::
user_shutdown
);
anon_send_exit
(
second
,
exit_reason
::
user_shutdown
);
}
CAF_TEST
(
currying
)
{
using
namespace
std
::
placeholders
;
auto
impl
=
[]()
->
behavior
{
return
{
[](
ok_atom
,
int
x
)
{
return
x
;
},
[](
ok_atom
,
double
x
)
{
return
x
;
}
};
};
actor_system
system
;
auto
aut
=
system
.
spawn
(
impl
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
auto
bound
=
aut
.
bind
(
ok_atom
::
value
,
_1
);
CAF_CHECK
(
aut
.
id
()
==
bound
.
id
());
CAF_CHECK
(
aut
.
node
()
==
bound
.
node
());
CAF_CHECK
(
aut
==
bound
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
scoped_actor
self
{
system
};
CAF_CHECK
(
system
.
registry
().
running
()
==
2
);
self
->
request
(
bound
,
2.0
).
receive
(
[](
double
y
)
{
CAF_CHECK
(
y
==
2.0
);
}
);
self
->
request
(
bound
,
10
).
receive
(
[](
int
y
)
{
CAF_CHECK
(
y
==
10
);
}
);
self
->
send_exit
(
bound
,
exit_reason
::
kill
);
self
->
await_all_other_actors_done
();
}
CAF_TEST
(
type_safe_currying
)
{
using
namespace
std
::
placeholders
;
using
testee
=
typed_actor
<
replies_to
<
ok_atom
,
int
>::
with
<
int
>
,
replies_to
<
ok_atom
,
double
>::
with
<
double
>>
;
auto
impl
=
[]()
->
testee
::
behavior_type
{
return
{
[](
ok_atom
,
int
x
)
{
return
x
;
},
[](
ok_atom
,
double
x
)
{
return
x
;
}
};
};
actor_system
system
;
auto
aut
=
system
.
spawn
(
impl
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
using
curried_signature
=
typed_actor
<
replies_to
<
int
>::
with
<
int
>
,
replies_to
<
double
>::
with
<
double
>>
;
//auto bound = actor_bind(aut, ok_atom::value, _1);
auto
bound
=
aut
.
bind
(
ok_atom
::
value
,
_1
);
CAF_CHECK
(
aut
==
bound
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
static_assert
(
std
::
is_same
<
decltype
(
bound
),
curried_signature
>::
value
,
"bind returned wrong actor handle"
);
scoped_actor
self
{
system
};
CAF_CHECK
(
system
.
registry
().
running
()
==
2
);
self
->
request
(
bound
,
2.0
).
receive
(
[](
double
y
)
{
CAF_CHECK
(
y
==
2.0
);
}
);
self
->
request
(
bound
,
10
).
receive
(
[](
int
y
)
{
CAF_CHECK
(
y
==
10
);
}
);
self
->
send_exit
(
bound
,
exit_reason
::
kill
);
self
->
await_all_other_actors_done
();
}
CAF_TEST
(
reordering
)
{
auto
impl
=
[]()
->
behavior
{
return
{
[](
int
x
,
double
y
)
{
return
x
*
y
;
}
};
};
actor_system
system
;
auto
aut
=
system
.
spawn
(
impl
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
using
namespace
std
::
placeholders
;
auto
bound
=
aut
.
bind
(
_2
,
_1
);
CAF_CHECK
(
aut
==
bound
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
scoped_actor
self
{
system
};
CAF_CHECK
(
system
.
registry
().
running
()
==
2
);
self
->
request
(
bound
,
2.0
,
10
).
receive
(
[](
double
y
)
{
CAF_CHECK
(
y
==
20.0
);
}
);
self
->
send_exit
(
bound
,
exit_reason
::
kill
);
self
->
await_all_other_actors_done
();
}
CAF_TEST
(
type_safe_reordering
)
{
using
testee
=
typed_actor
<
replies_to
<
int
,
double
>::
with
<
double
>>
;
auto
impl
=
[]()
->
testee
::
behavior_type
{
return
{
[](
int
x
,
double
y
)
{
return
x
*
y
;
}
};
};
actor_system
system
;
auto
aut
=
system
.
spawn
(
impl
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
using
namespace
std
::
placeholders
;
using
swapped_signature
=
typed_actor
<
replies_to
<
double
,
int
>::
with
<
double
>>
;
auto
bound
=
aut
.
bind
(
_2
,
_1
);
CAF_CHECK
(
aut
==
bound
);
CAF_CHECK
(
system
.
registry
().
running
()
==
1
);
static_assert
(
std
::
is_same
<
decltype
(
bound
),
swapped_signature
>::
value
,
"bind returned wrong actor handle"
);
scoped_actor
self
{
system
};
CAF_CHECK
(
system
.
registry
().
running
()
==
2
);
self
->
request
(
bound
,
2.0
,
10
).
receive
(
[](
double
y
)
{
CAF_CHECK
(
y
==
20.0
);
}
);
self
->
send_exit
(
bound
,
exit_reason
::
kill
);
self
->
await_all_other_actors_done
();
}
CAF_TEST_FIXTURE_SCOPE_END
()
CAF_TEST_FIXTURE_SCOPE_END
()
#endif // CAF_WINDOWS
#endif // CAF_WINDOWS
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