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
7b10813f
Commit
7b10813f
authored
Jan 20, 2020
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Rename policies for fan-out requests
parent
da9346e1
Changes
8
Hide whitespace changes
Inline
Side-by-side
Showing
8 changed files
with
69 additions
and
66 deletions
+69
-66
doc/tex/MessagePassing.tex
doc/tex/MessagePassing.tex
+4
-1
examples/message_passing/fan_out_request.cpp
examples/message_passing/fan_out_request.cpp
+3
-3
libcaf_core/CMakeLists.txt
libcaf_core/CMakeLists.txt
+2
-2
libcaf_core/caf/policy/select_all.hpp
libcaf_core/caf/policy/select_all.hpp
+24
-24
libcaf_core/caf/policy/select_any.hpp
libcaf_core/caf/policy/select_any.hpp
+8
-8
libcaf_core/test/mixin/requester.cpp
libcaf_core/test/mixin/requester.cpp
+3
-3
libcaf_core/test/policy/select_all.cpp
libcaf_core/test/policy/select_all.cpp
+13
-13
libcaf_core/test/policy/select_any.cpp
libcaf_core/test/policy/select_any.cpp
+12
-12
No files found.
doc/tex/MessagePassing.tex
View file @
7b10813f
...
...
@@ -192,7 +192,7 @@ Sending the same message to a group of workers is a common work flow in actor
applications. Usually, a manager maintains a set of workers. On request, the
manager fans-out the request to all of its workers and then collects the
results. The function
\lstinline
`fan
_
out
_
request` combined with the merge policy
\lstinline
`
fan
_
in
_
responses
` streamlines this exact use case.
\lstinline
`
select
_
all
` streamlines this exact use case.
In the following snippet, we have a matrix actor (
\lstinline
`self`) that stores
worker actors for each cell (each simply storing an integer). For computing the
...
...
@@ -204,6 +204,9 @@ results.
\cppexample
[86-98]
{
message
_
passing/fan
_
out
_
request
}
The policy
\lstinline
`select
_
any` models a second common use case: sending a
request to multiple receivers but only caring for the first arriving response.
\clearpage
\subsubsection
{
Error Handling in Requests
}
\label
{
error-response
}
...
...
examples/message_passing/fan_out_request.cpp
View file @
7b10813f
...
...
@@ -14,7 +14,7 @@
#include "caf/event_based_actor.hpp"
#include "caf/exec_main.hpp"
#include "caf/function_view.hpp"
#include "caf/policy/
fan_in_responses
.hpp"
#include "caf/policy/
select_all
.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/typed_actor.hpp"
...
...
@@ -87,7 +87,7 @@ matrix::behavior_type matrix_actor(matrix::stateful_pointer<matrix_state> self,
assert
(
row
<
rows
);
auto
rp
=
self
->
make_response_promise
<
double
>
();
auto
&
row_vec
=
self
->
state
.
rows
[
row
];
self
->
fan_out_request
<
policy
::
fan_in_responses
>
(
row_vec
,
infinite
,
get
)
self
->
fan_out_request
<
policy
::
select_all
>
(
row_vec
,
infinite
,
get
)
.
then
(
[
=
](
std
::
vector
<
int
>
xs
)
mutable
{
assert
(
xs
.
size
()
==
static_cast
<
size_t
>
(
columns
));
...
...
@@ -104,7 +104,7 @@ matrix::behavior_type matrix_actor(matrix::stateful_pointer<matrix_state> self,
for
(
int
row
=
0
;
row
<
rows
;
++
row
)
columns
.
emplace_back
(
rows_vec
[
row
][
column
]);
auto
rp
=
self
->
make_response_promise
<
double
>
();
self
->
fan_out_request
<
policy
::
fan_in_responses
>
(
columns
,
infinite
,
get
)
self
->
fan_out_request
<
policy
::
select_all
>
(
columns
,
infinite
,
get
)
.
then
(
[
=
](
std
::
vector
<
int
>
xs
)
mutable
{
assert
(
xs
.
size
()
==
static_cast
<
size_t
>
(
rows
));
...
...
libcaf_core/CMakeLists.txt
View file @
7b10813f
...
...
@@ -251,8 +251,8 @@ set(CAF_CORE_TEST_SOURCES
test/or_else.cpp
test/pipeline_streaming.cpp
test/policy/categorized.cpp
test/policy/
choose_response
.cpp
test/policy/
fan_in_responses
.cpp
test/policy/
select_all
.cpp
test/policy/
select_any
.cpp
test/request_timeout.cpp
test/result.cpp
test/rtti_pair.cpp
...
...
libcaf_core/caf/policy/
fan_in_responses
.hpp
→
libcaf_core/caf/policy/
select_all
.hpp
View file @
7b10813f
...
...
@@ -36,7 +36,7 @@
namespace
caf
::
detail
{
template
<
class
F
,
class
T
>
struct
fan_in_responses
_helper
{
struct
select_all
_helper
{
std
::
vector
<
T
>
results
;
std
::
shared_ptr
<
size_t
>
pending
;
F
f
;
...
...
@@ -51,7 +51,7 @@ struct fan_in_responses_helper {
}
template
<
class
Fun
>
fan_in_responses
_helper
(
size_t
pending
,
Fun
&&
f
)
select_all
_helper
(
size_t
pending
,
Fun
&&
f
)
:
pending
(
std
::
make_shared
<
size_t
>
(
pending
)),
f
(
std
::
forward
<
Fun
>
(
f
))
{
results
.
reserve
(
pending
);
}
...
...
@@ -62,7 +62,7 @@ struct fan_in_responses_helper {
};
template
<
class
F
,
class
...
Ts
>
struct
fan_in_responses
_tuple_helper
{
struct
select_all
_tuple_helper
{
using
value_type
=
std
::
tuple
<
Ts
...
>
;
std
::
vector
<
value_type
>
results
;
std
::
shared_ptr
<
size_t
>
pending
;
...
...
@@ -78,7 +78,7 @@ struct fan_in_responses_tuple_helper {
}
template
<
class
Fun
>
fan_in_responses
_tuple_helper
(
size_t
pending
,
Fun
&&
f
)
select_all
_tuple_helper
(
size_t
pending
,
Fun
&&
f
)
:
pending
(
std
::
make_shared
<
size_t
>
(
pending
)),
f
(
std
::
forward
<
Fun
>
(
f
))
{
results
.
reserve
(
pending
);
}
...
...
@@ -89,29 +89,28 @@ struct fan_in_responses_tuple_helper {
};
template
<
class
F
,
class
=
typename
detail
::
get_callable_trait
<
F
>
::
arg_types
>
struct
select_
fan_in_responses
_helper
;
struct
select_
select_all
_helper
;
template
<
class
F
,
class
...
Ts
>
struct
select_
fan_in_responses
_helper
<
struct
select_
select_all
_helper
<
F
,
detail
::
type_list
<
std
::
vector
<
std
::
tuple
<
Ts
...
>>>>
{
using
type
=
fan_in_responses
_tuple_helper
<
F
,
Ts
...
>
;
using
type
=
select_all
_tuple_helper
<
F
,
Ts
...
>
;
};
template
<
class
F
,
class
T
>
struct
select_
fan_in_responses
_helper
<
F
,
detail
::
type_list
<
std
::
vector
<
T
>>>
{
using
type
=
fan_in_responses
_helper
<
F
,
T
>
;
struct
select_
select_all
_helper
<
F
,
detail
::
type_list
<
std
::
vector
<
T
>>>
{
using
type
=
select_all
_helper
<
F
,
T
>
;
};
template
<
class
F
>
using
fan_in_responses_helper_t
=
typename
select_fan_in_responses_helper
<
F
>::
type
;
using
select_all_helper_t
=
typename
select_select_all_helper
<
F
>::
type
;
// TODO: Replace with a lambda when switching to C++17 (move g into lambda).
template
<
class
G
>
class
fan_in_responses
_error_handler
{
class
select_all
_error_handler
{
public:
template
<
class
Fun
>
fan_in_responses
_error_handler
(
Fun
&&
fun
,
std
::
shared_ptr
<
size_t
>
pending
)
select_all
_error_handler
(
Fun
&&
fun
,
std
::
shared_ptr
<
size_t
>
pending
)
:
handler
(
std
::
forward
<
Fun
>
(
fun
)),
pending
(
std
::
move
(
pending
))
{
// nop
}
...
...
@@ -133,12 +132,12 @@ private:
namespace
caf
::
policy
{
/// Enables a `response_handle` to fan-in
multiple respons
es into a single
/// result (a `vector`
of individual values) for the client
.
/// Enables a `response_handle` to fan-in
all responses messag
es into a single
/// result (a `vector`
that stores all received results)
.
/// @relates mixin::requester
/// @relates response_handle
template
<
class
ResponseType
>
class
fan_in_responses
{
class
select_all
{
public:
static
constexpr
bool
is_trivial
=
false
;
...
...
@@ -147,17 +146,18 @@ public:
using
message_id_list
=
std
::
vector
<
message_id
>
;
template
<
class
Fun
>
using
type_checker
=
detail
::
type_checker
<
response_type
,
detail
::
fan_in_responses_helper_t
<
detail
::
decay_t
<
Fun
>>>
;
using
type_checker
=
detail
::
type_checker
<
response_type
,
detail
::
select_all_helper_t
<
detail
::
decay_t
<
Fun
>>>
;
explicit
fan_in_responses
(
message_id_list
ids
)
:
ids_
(
std
::
move
(
ids
))
{
explicit
select_all
(
message_id_list
ids
)
:
ids_
(
std
::
move
(
ids
))
{
CAF_ASSERT
(
ids_
.
size
()
<=
static_cast
<
size_t
>
(
std
::
numeric_limits
<
int
>::
max
()));
}
fan_in_responses
(
fan_in_responses
&&
)
noexcept
=
default
;
select_all
(
select_all
&&
)
noexcept
=
default
;
fan_in_responses
&
operator
=
(
fan_in_responses
&&
)
noexcept
=
default
;
select_all
&
operator
=
(
select_all
&&
)
noexcept
=
default
;
template
<
class
Self
,
class
F
,
class
OnError
>
void
await
(
Self
*
self
,
F
&&
f
,
OnError
&&
g
)
const
{
...
...
@@ -178,7 +178,7 @@ public:
template
<
class
Self
,
class
F
,
class
G
>
void
receive
(
Self
*
self
,
F
&&
f
,
G
&&
g
)
const
{
CAF_LOG_TRACE
(
CAF_ARG
(
ids_
));
using
helper_type
=
detail
::
fan_in_responses
_helper_t
<
detail
::
decay_t
<
F
>>
;
using
helper_type
=
detail
::
select_all
_helper_t
<
detail
::
decay_t
<
F
>>
;
helper_type
helper
{
ids_
.
size
(),
std
::
forward
<
F
>
(
f
)};
auto
error_handler
=
[
&
](
error
&
err
)
{
if
(
*
helper
.
pending
>
0
)
{
...
...
@@ -202,8 +202,8 @@ private:
template
<
class
F
,
class
OnError
>
behavior
make_behavior
(
F
&&
f
,
OnError
&&
g
)
const
{
using
namespace
detail
;
using
helper_type
=
fan_in_responses
_helper_t
<
decay_t
<
F
>>
;
using
error_handler_type
=
fan_in_responses
_error_handler
<
decay_t
<
OnError
>>
;
using
helper_type
=
select_all
_helper_t
<
decay_t
<
F
>>
;
using
error_handler_type
=
select_all
_error_handler
<
decay_t
<
OnError
>>
;
helper_type
helper
{
ids_
.
size
(),
std
::
move
(
f
)};
error_handler_type
err_helper
{
std
::
forward
<
OnError
>
(
g
),
helper
.
pending
};
return
{
...
...
libcaf_core/caf/policy/
choose_response
.hpp
→
libcaf_core/caf/policy/
select_any
.hpp
View file @
7b10813f
...
...
@@ -32,10 +32,10 @@
namespace
caf
::
detail
{
template
<
class
F
,
class
=
typename
get_callable_trait
<
F
>
::
arg_types
>
struct
choose_response
_factory
;
struct
select_any
_factory
;
template
<
class
F
,
class
...
Ts
>
struct
choose_response
_factory
<
F
,
type_list
<
Ts
...
>>
{
struct
select_any
_factory
<
F
,
type_list
<
Ts
...
>>
{
template
<
class
Fun
>
static
auto
make
(
std
::
shared_ptr
<
size_t
>
pending
,
Fun
&&
fun
)
{
return
[
pending
,
f
{
std
::
forward
<
Fun
>
(
fun
)}](
Ts
...
xs
)
mutable
{
...
...
@@ -52,12 +52,12 @@ struct choose_response_factory<F, type_list<Ts...>> {
namespace
caf
::
policy
{
/// Enables a `response_handle` to
choose one response for any number of
///
response message
s.
/// Enables a `response_handle` to
pick the first arriving response, ignoring
///
all other result
s.
/// @relates mixin::requester
/// @relates response_handle
template
<
class
ResponseType
>
class
choose_response
{
class
select_any
{
public:
static
constexpr
bool
is_trivial
=
false
;
...
...
@@ -69,7 +69,7 @@ public:
using
type_checker
=
detail
::
type_checker
<
response_type
,
detail
::
decay_t
<
Fun
>>
;
explicit
choose_response
(
message_id_list
ids
)
:
ids_
(
std
::
move
(
ids
))
{
explicit
select_any
(
message_id_list
ids
)
:
ids_
(
std
::
move
(
ids
))
{
CAF_ASSERT
(
ids_
.
size
()
<=
static_cast
<
size_t
>
(
std
::
numeric_limits
<
int
>::
max
()));
}
...
...
@@ -93,7 +93,7 @@ public:
template
<
class
Self
,
class
F
,
class
G
>
void
receive
(
Self
*
self
,
F
&&
f
,
G
&&
g
)
const
{
CAF_LOG_TRACE
(
CAF_ARG
(
ids_
));
using
factory
=
detail
::
choose_response
_factory
<
std
::
decay_t
<
F
>>
;
using
factory
=
detail
::
select_any
_factory
<
std
::
decay_t
<
F
>>
;
auto
pending
=
std
::
make_shared
<
size_t
>
(
ids_
.
size
());
auto
fw
=
factory
::
make
(
pending
,
std
::
forward
<
F
>
(
f
));
auto
gw
=
make_error_handler
(
std
::
move
(
pending
),
std
::
forward
<
G
>
(
g
));
...
...
@@ -126,7 +126,7 @@ private:
template
<
class
F
,
class
OnError
>
behavior
make_behavior
(
F
&&
f
,
OnError
&&
g
)
const
{
using
factory
=
detail
::
choose_response
_factory
<
std
::
decay_t
<
F
>>
;
using
factory
=
detail
::
select_any
_factory
<
std
::
decay_t
<
F
>>
;
auto
pending
=
std
::
make_shared
<
size_t
>
(
ids_
.
size
());
auto
result_handler
=
factory
::
make
(
pending
,
std
::
forward
<
F
>
(
f
));
return
{
...
...
libcaf_core/test/mixin/requester.cpp
View file @
7b10813f
...
...
@@ -23,7 +23,7 @@
#include "caf/test/dsl.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/policy/
fan_in_responses
.hpp"
#include "caf/policy/
select_all
.hpp"
using
namespace
caf
;
...
...
@@ -152,7 +152,7 @@ CAF_TEST(delegated request with integer result) {
}
CAF_TEST
(
requesters
support
fan_out_request
)
{
using
policy
::
fan_in_responses
;
using
policy
::
select_all
;
std
::
vector
<
adding_server_type
>
workers
{
make_server
([](
int
x
,
int
y
)
{
return
x
+
y
;
}),
make_server
([](
int
x
,
int
y
)
{
return
x
+
y
;
}),
...
...
@@ -161,7 +161,7 @@ CAF_TEST(requesters support fan_out_request) {
run
();
auto
sum
=
std
::
make_shared
<
int
>
(
0
);
auto
client
=
sys
.
spawn
([
=
](
event_based_actor
*
self
)
{
self
->
fan_out_request
<
fan_in_responses
>
(
workers
,
infinite
,
1
,
2
)
self
->
fan_out_request
<
select_all
>
(
workers
,
infinite
,
1
,
2
)
.
then
([
=
](
std
::
vector
<
int
>
results
)
{
for
(
auto
result
:
results
)
CAF_CHECK_EQUAL
(
result
,
3
);
...
...
libcaf_core/test/policy/
fan_in_responses
.cpp
→
libcaf_core/test/policy/
select_all
.cpp
View file @
7b10813f
...
...
@@ -16,9 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE policy.
fan_in_responses
#define CAF_SUITE policy.
select_all
#include "caf/policy/
fan_in_responses
.hpp"
#include "caf/policy/
select_all
.hpp"
#include "caf/test/dsl.hpp"
...
...
@@ -28,7 +28,7 @@
#include "caf/event_based_actor.hpp"
#include "caf/sec.hpp"
using
caf
::
policy
::
fan_in_responses
;
using
caf
::
policy
::
select_all
;
using
namespace
caf
;
...
...
@@ -63,9 +63,9 @@ struct fixture : test_coordinator_fixture<> {
CAF_MESSAGE("subtest: " message); \
for (int subtest_dummy = 0; subtest_dummy < 1; ++subtest_dummy)
CAF_TEST_FIXTURE_SCOPE
(
fan_in_responses
_tests
,
fixture
)
CAF_TEST_FIXTURE_SCOPE
(
select_all
_tests
,
fixture
)
CAF_TEST
(
fan_in_responses
combines
two
integer
results
into
one
vector
)
{
CAF_TEST
(
select_all
combines
two
integer
results
into
one
vector
)
{
using
int_list
=
std
::
vector
<
int
>
;
auto
f
=
[](
int
x
,
int
y
)
{
return
x
+
y
;
};
auto
server1
=
make_server
(
f
);
...
...
@@ -74,7 +74,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
SUBTEST
(
"vector of int"
)
{
auto
r1
=
self
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
self
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
run
();
merge
.
receive
(
self
.
ptr
(),
...
...
@@ -88,7 +88,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
using
std
::
make_tuple
;
auto
r1
=
self
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
self
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
run
();
using
results_vector
=
std
::
vector
<
std
::
tuple
<
int
>>
;
merge
.
receive
(
...
...
@@ -106,7 +106,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
auto
client
=
sys
.
spawn
([
=
,
&
results
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
merge
.
then
(
client_ptr
,
[
&
results
](
int_list
xs
)
{
results
=
std
::
move
(
xs
);
},
make_error_handler
());
...
...
@@ -124,7 +124,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
auto
client
=
sys
.
spawn
([
=
,
&
results
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
merge
.
await
(
client_ptr
,
[
&
results
](
int_list
xs
)
{
results
=
std
::
move
(
xs
);
},
make_error_handler
());
...
...
@@ -141,7 +141,7 @@ CAF_TEST(fan_in_responses combines two integer results into one vector) {
}
}
CAF_TEST
(
fan_in_responses
calls
the
error
handler
at
most
once
)
{
CAF_TEST
(
select_all
calls
the
error
handler
at
most
once
)
{
using
int_list
=
std
::
vector
<
int
>
;
auto
f
=
[](
int
,
int
)
->
result
<
int
>
{
return
sec
::
invalid_argument
;
};
auto
server1
=
make_server
(
f
);
...
...
@@ -149,7 +149,7 @@ CAF_TEST(fan_in_responses calls the error handler at most once) {
SUBTEST
(
"request.receive"
)
{
auto
r1
=
self
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
self
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
run
();
size_t
errors
=
0
;
merge
.
receive
(
...
...
@@ -163,7 +163,7 @@ CAF_TEST(fan_in_responses calls the error handler at most once) {
auto
client
=
sys
.
spawn
([
=
,
&
errors
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
merge
.
then
(
client_ptr
,
[](
int_list
)
{
CAF_FAIL
(
"fan-in policy called the result handler"
);
},
...
...
@@ -181,7 +181,7 @@ CAF_TEST(fan_in_responses calls the error handler at most once) {
auto
client
=
sys
.
spawn
([
=
,
&
errors
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
fan_in_responses
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
select_all
<
detail
::
type_list
<
int
>>
merge
{{
r1
.
id
(),
r2
.
id
()}};
merge
.
await
(
client_ptr
,
[](
int_list
)
{
CAF_FAIL
(
"fan-in policy called the result handler"
);
},
...
...
libcaf_core/test/policy/
choose_response
.cpp
→
libcaf_core/test/policy/
select_any
.cpp
View file @
7b10813f
...
...
@@ -16,9 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE policy.
choose_response
#define CAF_SUITE policy.
select_any
#include "caf/policy/
choose_response
.hpp"
#include "caf/policy/
select_any
.hpp"
#include "caf/test/dsl.hpp"
...
...
@@ -26,7 +26,7 @@
#include "caf/event_based_actor.hpp"
#include "caf/sec.hpp"
using
caf
::
policy
::
choose_response
;
using
caf
::
policy
::
select_any
;
using
namespace
caf
;
...
...
@@ -61,9 +61,9 @@ struct fixture : test_coordinator_fixture<> {
CAF_MESSAGE("subtest: " message); \
for (int subtest_dummy = 0; subtest_dummy < 1; ++subtest_dummy)
CAF_TEST_FIXTURE_SCOPE
(
choose_response
_tests
,
fixture
)
CAF_TEST_FIXTURE_SCOPE
(
select_any
_tests
,
fixture
)
CAF_TEST
(
choose_response
picks
the
first
arriving
integer
)
{
CAF_TEST
(
select_any
picks
the
first
arriving
integer
)
{
auto
f
=
[](
int
x
,
int
y
)
{
return
x
+
y
;
};
auto
server1
=
make_server
(
f
);
auto
server2
=
make_server
(
f
);
...
...
@@ -71,7 +71,7 @@ CAF_TEST(choose_response picks the first arriving integer) {
SUBTEST
(
"single integer"
)
{
auto
r1
=
self
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
self
->
request
(
server2
,
infinite
,
2
,
3
);
choose_response
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
select_any
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
run
();
choose
.
receive
(
self
.
ptr
(),
[](
int
result
)
{
CAF_CHECK_EQUAL
(
result
,
3
);
},
...
...
@@ -83,7 +83,7 @@ CAF_TEST(choose_response picks the first arriving integer) {
auto
client
=
sys
.
spawn
([
=
,
&
result
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
choose_response
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
select_any
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
choose
.
then
(
client_ptr
,
[
&
result
](
int
x
)
{
result
=
x
;
},
make_error_handler
());
});
...
...
@@ -100,7 +100,7 @@ CAF_TEST(choose_response picks the first arriving integer) {
auto
client
=
sys
.
spawn
([
=
,
&
result
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
choose_response
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
select_any
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
choose
.
await
(
client_ptr
,
[
&
result
](
int
x
)
{
result
=
x
;
},
make_error_handler
());
});
...
...
@@ -116,14 +116,14 @@ CAF_TEST(choose_response picks the first arriving integer) {
}
}
CAF_TEST
(
choose_response
calls
the
error
handler
at
most
once
)
{
CAF_TEST
(
select_any
calls
the
error
handler
at
most
once
)
{
auto
f
=
[](
int
,
int
)
->
result
<
int
>
{
return
sec
::
invalid_argument
;
};
auto
server1
=
make_server
(
f
);
auto
server2
=
make_server
(
f
);
SUBTEST
(
"request.receive"
)
{
auto
r1
=
self
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
self
->
request
(
server2
,
infinite
,
2
,
3
);
choose_response
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
select_any
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
run
();
size_t
errors
=
0
;
choose
.
receive
(
...
...
@@ -137,7 +137,7 @@ CAF_TEST(choose_response calls the error handler at most once) {
auto
client
=
sys
.
spawn
([
=
,
&
errors
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
choose_response
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
select_any
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
choose
.
then
(
client_ptr
,
[](
int
)
{
CAF_FAIL
(
"fan-in policy called the result handler"
);
},
...
...
@@ -155,7 +155,7 @@ CAF_TEST(choose_response calls the error handler at most once) {
auto
client
=
sys
.
spawn
([
=
,
&
errors
](
event_based_actor
*
client_ptr
)
{
auto
r1
=
client_ptr
->
request
(
server1
,
infinite
,
1
,
2
);
auto
r2
=
client_ptr
->
request
(
server2
,
infinite
,
2
,
3
);
choose_response
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
select_any
<
detail
::
type_list
<
int
>>
choose
{{
r1
.
id
(),
r2
.
id
()}};
choose
.
await
(
client_ptr
,
[](
int
)
{
CAF_FAIL
(
"fan-in policy called the result handler"
);
},
...
...
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