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
21281a6c
Commit
21281a6c
authored
May 08, 2020
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Switch to _v syntax for atoms
parent
c584b10b
Changes
50
Expand all
Hide whitespace changes
Inline
Side-by-side
Showing
50 changed files
with
623 additions
and
896 deletions
+623
-896
libcaf_core/caf/actor_pool.hpp
libcaf_core/caf/actor_pool.hpp
+1
-1
libcaf_core/caf/all.hpp
libcaf_core/caf/all.hpp
+13
-16
libcaf_core/src/actor_ostream.cpp
libcaf_core/src/actor_ostream.cpp
+9
-13
libcaf_core/src/actor_system.cpp
libcaf_core/src/actor_system.cpp
+3
-3
libcaf_core/src/blocking_actor.cpp
libcaf_core/src/blocking_actor.cpp
+2
-4
libcaf_core/src/forwarding_actor_proxy.cpp
libcaf_core/src/forwarding_actor_proxy.cpp
+5
-5
libcaf_core/src/group_manager.cpp
libcaf_core/src/group_manager.cpp
+3
-3
libcaf_core/src/monitorable_actor.cpp
libcaf_core/src/monitorable_actor.cpp
+4
-4
libcaf_core/src/scheduled_actor.cpp
libcaf_core/src/scheduled_actor.cpp
+7
-7
libcaf_core/test/actor_clock.cpp
libcaf_core/test/actor_clock.cpp
+10
-10
libcaf_core/test/actor_lifetime.cpp
libcaf_core/test/actor_lifetime.cpp
+11
-20
libcaf_core/test/actor_pool.cpp
libcaf_core/test/actor_pool.cpp
+3
-3
libcaf_core/test/actor_registry.cpp
libcaf_core/test/actor_registry.cpp
+5
-3
libcaf_core/test/atom.cpp
libcaf_core/test/atom.cpp
+7
-12
libcaf_core/test/constructor_attach.cpp
libcaf_core/test/constructor_attach.cpp
+10
-20
libcaf_core/test/continuous_streaming.cpp
libcaf_core/test/continuous_streaming.cpp
+6
-6
libcaf_core/test/detached_actors.cpp
libcaf_core/test/detached_actors.cpp
+4
-4
libcaf_core/test/dynamic_spawn.cpp
libcaf_core/test/dynamic_spawn.cpp
+97
-173
libcaf_core/test/function_view.cpp
libcaf_core/test/function_view.cpp
+3
-3
libcaf_core/test/fused_downstream_manager.cpp
libcaf_core/test/fused_downstream_manager.cpp
+13
-22
libcaf_core/test/local_group.cpp
libcaf_core/test/local_group.cpp
+3
-3
libcaf_core/test/local_migration.cpp
libcaf_core/test/local_migration.cpp
+9
-9
libcaf_core/test/match.cpp
libcaf_core/test/match.cpp
+5
-5
libcaf_core/test/message.cpp
libcaf_core/test/message.cpp
+25
-41
libcaf_core/test/or_else.cpp
libcaf_core/test/or_else.cpp
+31
-47
libcaf_core/test/pipeline_streaming.cpp
libcaf_core/test/pipeline_streaming.cpp
+1
-1
libcaf_core/test/request_timeout.cpp
libcaf_core/test/request_timeout.cpp
+110
-146
libcaf_core/test/selective_streaming.cpp
libcaf_core/test/selective_streaming.cpp
+2
-2
libcaf_core/test/serial_reply.cpp
libcaf_core/test/serial_reply.cpp
+30
-45
libcaf_core/test/serialization.cpp
libcaf_core/test/serialization.cpp
+2
-3
libcaf_core/test/simple_timeout.cpp
libcaf_core/test/simple_timeout.cpp
+2
-2
libcaf_core/test/stateful_actor.cpp
libcaf_core/test/stateful_actor.cpp
+15
-35
libcaf_core/test/typed_response_promise.cpp
libcaf_core/test/typed_response_promise.cpp
+46
-63
libcaf_core/test/typed_spawn.cpp
libcaf_core/test/typed_spawn.cpp
+12
-36
libcaf_core/test/unit.cpp
libcaf_core/test/unit.cpp
+23
-26
libcaf_io/src/io/basp/instance.cpp
libcaf_io/src/io/basp/instance.cpp
+1
-1
libcaf_io/src/io/basp_broker.cpp
libcaf_io/src/io/basp_broker.cpp
+12
-13
libcaf_io/src/io/connection_helper.cpp
libcaf_io/src/io/connection_helper.cpp
+1
-1
libcaf_io/src/io/middleman.cpp
libcaf_io/src/io/middleman.cpp
+16
-18
libcaf_io/src/io/middleman_actor_impl.cpp
libcaf_io/src/io/middleman_actor_impl.cpp
+4
-4
libcaf_io/test/io/basp/message_queue.cpp
libcaf_io/test/io/basp/message_queue.cpp
+1
-1
libcaf_io/test/io/basp_broker.cpp
libcaf_io/test/io/basp_broker.cpp
+9
-10
libcaf_io/test/io/broker.cpp
libcaf_io/test/io/broker.cpp
+31
-31
libcaf_io/test/io/remote_actor.cpp
libcaf_io/test/io/remote_actor.cpp
+7
-12
libcaf_io/test/io/remote_group.cpp
libcaf_io/test/io/remote_group.cpp
+2
-2
libcaf_io/test/io/remote_spawn.cpp
libcaf_io/test/io/remote_spawn.cpp
+3
-3
libcaf_io/test/io/worker.cpp
libcaf_io/test/io/worker.cpp
+1
-1
libcaf_openssl/caf/openssl/unpublish.hpp
libcaf_openssl/caf/openssl/unpublish.hpp
+1
-1
libcaf_openssl/src/openssl/publish.cpp
libcaf_openssl/src/openssl/publish.cpp
+1
-1
libcaf_openssl/src/openssl/remote_actor.cpp
libcaf_openssl/src/openssl/remote_actor.cpp
+1
-1
No files found.
libcaf_core/caf/actor_pool.hpp
View file @
21281a6c
...
...
@@ -37,7 +37,7 @@ namespace caf {
/// around in an actor system to hide the actual set of workers.
///
/// After construction, new workers can be added via `{'SYS', 'PUT', actor}`
/// messages, e.g., `send(my_pool, sys_atom
::value, put_atom::value
, worker)`.
/// messages, e.g., `send(my_pool, sys_atom
_v, put_atom_v
, worker)`.
/// `{'SYS', 'DELETE', actor}` messages remove a specific worker from the set,
/// `{'SYS', 'DELETE'}` removes all workers, and `{'SYS', 'GET'}` returns a
/// `vector<actor>` containing all workers.
...
...
libcaf_core/caf/all.hpp
View file @
21281a6c
...
...
@@ -247,10 +247,10 @@
/// using result_atom = atom_constant<atom("result")>;
///
/// // send a message to a1
/// self->send(a1, hello_atom
::value
, "hello a1!");
/// self->send(a1, hello_atom
_v
, "hello a1!");
///
/// // send a message to a1, a2, and a3
/// auto msg = make_message(compute_atom
::value
, 1, 2, 3);
/// auto msg = make_message(compute_atom
_v
, 1, 2, 3);
/// self->send(a1, msg);
/// self->send(a2, msg);
/// self->send(a3, msg);
...
...
@@ -269,7 +269,7 @@
/// },
/// [](compute_atom, int i0, int i1, int i2) {
/// // send our result back to the sender of this messages
/// return make_message(result_atom
::value
, i0 + i1 + i2);
/// return make_message(result_atom
_v
, i0 + i1 + i2);
/// }
/// }
/// ~~
...
...
@@ -289,11 +289,11 @@
///
/// @section Atoms Atoms
///
/// Atoms are a nice way to add semantic informations to a message.
///
Assuming an actor wants to provide a "math sevice" for integers. It
///
could provide operations such as addition, subtraction, etc.
///
This operations all have two operands. Thus, the actor does not know
///
what operation the sender of a
message wanted by receiving just two integers.
/// Atoms are a nice way to add semantic informations to a message.
Assuming an
///
actor wants to provide a "math sevice" for integers. It could provide
///
operations such as addition, subtraction, etc. This operations all have two
///
operands. Thus, the actor does not know what operation the sender of a
/// message wanted by receiving just two integers.
///
/// Example actor:
/// ~~
...
...
@@ -371,14 +371,14 @@
/// ~~
/// scoped_actor self{...};
///
/// self->delayed_send(self, std::chrono::seconds(1), poll_atom
::value
);
/// self->delayed_send(self, std::chrono::seconds(1), poll_atom
_v
);
/// bool running = true;
/// self->receive_while([&](){ return running; }) (
/// // ...
/// [&](poll_atom) {
/// // ... poll something ...
/// // and do it again after 1sec
/// self->delayed_send(self, std::chrono::seconds(1), poll_atom
::value
);
/// self->delayed_send(self, std::chrono::seconds(1), poll_atom
_v
);
/// }
/// );
/// ~~
...
...
@@ -396,18 +396,15 @@
///
/// A few examples:
/// ~~
/// // sends a
n std::
string containing "hello actor!" to itself
/// // sends a
string containing "hello actor!" to itself
/// send(self, "hello actor!");
///
/// const char* cstring = "cstring";
/// // sends an st
d::st
ring containing "cstring" to itself
/// // sends an string containing "cstring" to itself
/// send(self, cstring);
///
/// // sends an
std::
u16string containing the UTF16 string "hello unicode world!"
/// // sends an u16string containing the UTF16 string "hello unicode world!"
/// send(self, u"hello unicode world!");
///
/// // x has the type caf::tuple<std::string, std::string>
/// auto x = make_message("hello", "tuple");
/// ~~
///
/// @defgroup ActorCreation Creating Actors
...
...
libcaf_core/src/actor_ostream.cpp
View file @
21281a6c
...
...
@@ -18,38 +18,35 @@
#include "caf/actor_ostream.hpp"
#include "caf/send.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/default_attachable.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/send.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
namespace
caf
{
actor_ostream
::
actor_ostream
(
local_actor
*
self
)
:
self_
(
self
->
id
()),
printer_
(
self
->
home_system
().
scheduler
().
printer
())
{
:
self_
(
self
->
id
()),
printer_
(
self
->
home_system
().
scheduler
().
printer
())
{
init
(
self
);
}
actor_ostream
::
actor_ostream
(
scoped_actor
&
self
)
:
self_
(
self
->
id
()),
printer_
(
self
->
home_system
().
scheduler
().
printer
())
{
:
self_
(
self
->
id
()),
printer_
(
self
->
home_system
().
scheduler
().
printer
())
{
init
(
actor_cast
<
abstract_actor
*>
(
self
));
}
actor_ostream
&
actor_ostream
::
write
(
std
::
string
arg
)
{
printer_
->
enqueue
(
make_mailbox_element
(
nullptr
,
make_message_id
(),
{},
add_atom
::
value
,
self_
,
std
::
move
(
arg
)),
add_atom_v
,
self_
,
std
::
move
(
arg
)),
nullptr
);
return
*
this
;
}
actor_ostream
&
actor_ostream
::
flush
()
{
printer_
->
enqueue
(
make_mailbox_element
(
nullptr
,
make_message_id
(),
{},
flush_atom
::
value
,
self_
),
flush_atom_v
,
self_
),
nullptr
);
return
*
this
;
}
...
...
@@ -59,16 +56,15 @@ void actor_ostream::redirect(abstract_actor* self, std::string fn, int flags) {
return
;
auto
pr
=
self
->
home_system
().
scheduler
().
printer
();
pr
->
enqueue
(
make_mailbox_element
(
nullptr
,
make_message_id
(),
{},
redirect_atom
::
value
,
self
->
id
(
),
std
::
move
(
fn
),
flags
),
redirect_atom_v
,
self
->
id
(),
std
::
move
(
fn
),
flags
),
nullptr
);
}
void
actor_ostream
::
redirect_all
(
actor_system
&
sys
,
std
::
string
fn
,
int
flags
)
{
auto
pr
=
sys
.
scheduler
().
printer
();
pr
->
enqueue
(
make_mailbox_element
(
nullptr
,
make_message_id
(),
{},
redirect_atom
::
value
,
std
::
move
(
fn
),
flags
),
redirect_atom_v
,
std
::
move
(
fn
),
flags
),
nullptr
);
}
...
...
libcaf_core/src/actor_system.cpp
View file @
21281a6c
...
...
@@ -82,13 +82,13 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) {
// we never put a nullptr in our map
auto
subscriber
=
actor_cast
<
actor
>
(
subscriber_ptr
);
if
(
subscriber
!=
self
->
current_sender
())
self
->
send
(
subscriber
,
update_atom
::
value
,
key
,
vp
.
first
);
self
->
send
(
subscriber
,
update_atom
_v
,
key
,
vp
.
first
);
}
// also iterate all subscribers for '*'
for
(
auto
&
subscriber
:
self
->
state
.
data
[
wildcard
].
second
)
if
(
subscriber
!=
self
->
current_sender
())
self
->
send
(
actor_cast
<
actor
>
(
subscriber
),
update_atom
::
value
,
key
,
vp
.
first
);
self
->
send
(
actor_cast
<
actor
>
(
subscriber
),
update_atom
_v
,
key
,
vp
.
first
);
},
// get a key/value pair
[
=
](
get_atom
,
std
::
string
&
key
)
->
message
{
...
...
libcaf_core/src/blocking_actor.cpp
View file @
21281a6c
...
...
@@ -316,13 +316,11 @@ size_t blocking_actor::attach_functor(const actor_addr& x) {
}
size_t
blocking_actor
::
attach_functor
(
const
strong_actor_ptr
&
ptr
)
{
using
wait_for_atom
=
atom_constant
<
atom
(
"waitFor"
)
>
;
if
(
!
ptr
)
return
0
;
actor
self
{
this
};
ptr
->
get
()
->
attach_functor
([
=
](
const
error
&
)
{
anon_send
(
self
,
wait_for_atom
::
value
);
});
ptr
->
get
()
->
attach_functor
(
[
=
](
const
error
&
)
{
anon_send
(
self
,
wait_for_atom_v
);
});
return
1
;
}
...
...
libcaf_core/src/forwarding_actor_proxy.cpp
View file @
21281a6c
...
...
@@ -30,11 +30,11 @@ namespace caf {
forwarding_actor_proxy
::
forwarding_actor_proxy
(
actor_config
&
cfg
,
actor
dest
)
:
actor_proxy
(
cfg
),
broker_
(
std
::
move
(
dest
))
{
anon_send
(
broker_
,
monitor_atom
::
value
,
ctrl
());
anon_send
(
broker_
,
monitor_atom
_v
,
ctrl
());
}
forwarding_actor_proxy
::~
forwarding_actor_proxy
()
{
anon_send
(
broker_
,
make_message
(
delete_atom
::
value
,
node
(),
id
()));
anon_send
(
broker_
,
make_message
(
delete_atom
_v
,
node
(),
id
()));
}
void
forwarding_actor_proxy
::
forward_msg
(
strong_actor_ptr
sender
,
...
...
@@ -48,7 +48,7 @@ void forwarding_actor_proxy::forward_msg(strong_actor_ptr sender,
shared_lock
<
detail
::
shared_spinlock
>
guard
(
broker_mtx_
);
if
(
broker_
)
broker_
->
enqueue
(
nullptr
,
make_message_id
(),
make_message
(
forward_atom
::
value
,
std
::
move
(
sender
),
make_message
(
forward_atom
_v
,
std
::
move
(
sender
),
fwd
!=
nullptr
?
*
fwd
:
tmp
,
strong_actor_ptr
{
ctrl
()},
mid
,
std
::
move
(
msg
)),
...
...
@@ -66,7 +66,7 @@ void forwarding_actor_proxy::enqueue(mailbox_element_ptr what,
bool
forwarding_actor_proxy
::
add_backlink
(
abstract_actor
*
x
)
{
if
(
monitorable_actor
::
add_backlink
(
x
))
{
forward_msg
(
ctrl
(),
make_message_id
(),
make_message
(
link_atom
::
value
,
x
->
ctrl
()));
make_message
(
link_atom
_v
,
x
->
ctrl
()));
return
true
;
}
return
false
;
...
...
@@ -75,7 +75,7 @@ bool forwarding_actor_proxy::add_backlink(abstract_actor* x) {
bool
forwarding_actor_proxy
::
remove_backlink
(
abstract_actor
*
x
)
{
if
(
monitorable_actor
::
remove_backlink
(
x
))
{
forward_msg
(
ctrl
(),
make_message_id
(),
make_message
(
unlink_atom
::
value
,
x
->
ctrl
()));
make_message
(
unlink_atom
_v
,
x
->
ctrl
()));
return
true
;
}
return
false
;
...
...
libcaf_core/src/group_manager.cpp
View file @
21281a6c
...
...
@@ -257,7 +257,7 @@ public:
if
(
res
.
first
)
{
// join remote source
if
(
res
.
second
==
1
)
anon_send
(
broker_
,
join_atom
::
value
,
proxy_broker_
);
anon_send
(
broker_
,
join_atom
_v
,
proxy_broker_
);
return
true
;
}
CAF_LOG_WARNING
(
"actor already joined group"
);
...
...
@@ -270,7 +270,7 @@ public:
if
(
res
.
first
&&
res
.
second
==
0
)
{
// leave the remote source,
// because there's no more subscriber on this node
anon_send
(
broker_
,
leave_atom
::
value
,
proxy_broker_
);
anon_send
(
broker_
,
leave_atom
_v
,
proxy_broker_
);
}
}
...
...
@@ -279,7 +279,7 @@ public:
CAF_LOG_TRACE
(
CAF_ARG
(
sender
)
<<
CAF_ARG
(
mid
)
<<
CAF_ARG
(
msg
));
// forward message to the broker
broker_
->
enqueue
(
std
::
move
(
sender
),
mid
,
make_message
(
forward_atom
::
value
,
std
::
move
(
msg
)),
eu
);
make_message
(
forward_atom
_v
,
std
::
move
(
msg
)),
eu
);
}
void
stop
()
override
{
...
...
libcaf_core/src/monitorable_actor.cpp
View file @
21281a6c
...
...
@@ -99,7 +99,7 @@ bool monitorable_actor::cleanup(error&& reason, execution_unit* host) {
if
(
getf
(
abstract_actor
::
has_used_aout_flag
))
{
auto
pr
=
home_system
().
scheduler
().
printer
();
pr
->
enqueue
(
make_mailbox_element
(
nullptr
,
make_message_id
(),
{},
delete_atom
::
value
,
id
()),
delete_atom_v
,
id
()),
nullptr
);
}
return
true
;
...
...
@@ -241,9 +241,9 @@ bool monitorable_actor::handle_system_message(mailbox_element& x,
err
=
sec
::
unsupported_sys_key
;
return
;
}
res
=
make_mailbox_element
(
ctrl
(),
x
.
mid
.
response_id
(),
{},
ok_atom
::
value
,
std
::
move
(
what
)
,
strong_actor_ptr
{
ctrl
()},
name
());
res
=
make_mailbox_element
(
ctrl
(),
x
.
mid
.
response_id
(),
{},
ok_atom_v
,
std
::
move
(
what
),
strong_actor_ptr
{
ctrl
()}
,
name
());
}
);
if
(
!
res
&&
!
err
)
...
...
libcaf_core/src/scheduled_actor.cpp
View file @
21281a6c
...
...
@@ -473,7 +473,7 @@ void scheduled_actor::quit(error x) {
uint64_t
scheduled_actor
::
set_receive_timeout
(
actor_clock
::
time_point
x
)
{
CAF_LOG_TRACE
(
x
);
setf
(
has_timeout_flag
);
return
set_timeout
(
receive_atom
::
value
,
x
);
return
set_timeout
(
receive_atom
_v
,
x
);
}
uint64_t
scheduled_actor
::
set_receive_timeout
()
{
...
...
@@ -488,7 +488,7 @@ uint64_t scheduled_actor::set_receive_timeout() {
if
(
d
.
is_zero
())
{
// immediately enqueue timeout message if duration == 0s
auto
id
=
++
timeout_id_
;
auto
type
=
receive_atom
::
value
;
auto
type
=
receive_atom
_v
;
eq_impl
(
make_message_id
(),
nullptr
,
context
(),
timeout_msg
{
type
,
id
});
return
id
;
}
...
...
@@ -525,7 +525,7 @@ uint64_t scheduled_actor::set_stream_timeout(actor_clock::time_point x) {
return
0
;
}
// Delegate call.
return
set_timeout
(
stream_atom
::
value
,
x
);
return
set_timeout
(
stream_atom
_v
,
x
);
}
// -- message processing -------------------------------------------------------
...
...
@@ -550,8 +550,8 @@ scheduled_actor::categorize(mailbox_element& x) {
auto
&
content
=
x
.
content
();
switch
(
content
.
type_token
())
{
case
make_type_token
<
atom_value
,
atom_value
,
std
:
:
string
>
()
:
if
(
content
.
get_as
<
atom_value
>
(
0
)
==
sys_atom
::
value
&&
content
.
get_as
<
atom_value
>
(
1
)
==
get_atom
::
value
)
{
if
(
content
.
get_as
<
atom_value
>
(
0
)
==
sys_atom
_v
&&
content
.
get_as
<
atom_value
>
(
1
)
==
get_atom
_v
)
{
auto
rp
=
make_response_promise
();
if
(
!
rp
.
pending
())
{
CAF_LOG_WARNING
(
"received anonymous ('get', 'sys', $key) message"
);
...
...
@@ -560,7 +560,7 @@ scheduled_actor::categorize(mailbox_element& x) {
auto
&
what
=
content
.
get_as
<
std
::
string
>
(
2
);
if
(
what
==
"info"
)
{
CAF_LOG_DEBUG
(
"reply to 'info' message"
);
rp
.
deliver
(
ok_atom
::
value
,
std
::
move
(
what
),
strong_actor_ptr
{
ctrl
()},
rp
.
deliver
(
ok_atom
_v
,
std
::
move
(
what
),
strong_actor_ptr
{
ctrl
()},
name
());
}
else
{
rp
.
deliver
(
make_error
(
sec
::
unsupported_sys_key
));
...
...
@@ -572,7 +572,7 @@ scheduled_actor::categorize(mailbox_element& x) {
CAF_ASSERT
(
x
.
mid
.
is_async
());
auto
&
tm
=
content
.
get_as
<
timeout_msg
>
(
0
);
auto
tid
=
tm
.
timeout_id
;
if
(
tm
.
type
==
receive_atom
::
value
)
{
if
(
tm
.
type
==
receive_atom
_v
)
{
CAF_LOG_DEBUG
(
"handle ordinary timeout message"
);
if
(
is_active_receive_timeout
(
tid
)
&&
!
bhvr_stack_
.
empty
())
bhvr_stack_
.
back
().
handle_timeout
();
...
...
libcaf_core/test/actor_clock.cpp
View file @
21281a6c
...
...
@@ -98,7 +98,7 @@ CAF_TEST_FIXTURE_SCOPE(timer_tests, fixture)
CAF_TEST
(
single_receive_timeout
)
{
// Have AUT call t.set_receive_timeout().
self
->
send
(
aut
,
ok_atom
::
value
);
self
->
send
(
aut
,
ok_atom
_v
);
expect
((
ok_atom
),
from
(
self
).
to
(
aut
).
with
(
_
));
CAF_CHECK_EQUAL
(
t
.
schedule
().
size
(),
1u
);
CAF_CHECK_EQUAL
(
t
.
actor_lookup
().
size
(),
1u
);
...
...
@@ -112,12 +112,12 @@ CAF_TEST(single_receive_timeout) {
CAF_TEST
(
override_receive_timeout
)
{
// Have AUT call t.set_receive_timeout().
self
->
send
(
aut
,
ok_atom
::
value
);
self
->
send
(
aut
,
ok_atom
_v
);
expect
((
ok_atom
),
from
(
self
).
to
(
aut
).
with
(
_
));
CAF_CHECK_EQUAL
(
t
.
schedule
().
size
(),
1u
);
CAF_CHECK_EQUAL
(
t
.
actor_lookup
().
size
(),
1u
);
// Have AUT call t.set_timeout() again.
self
->
send
(
aut
,
ok_atom
::
value
);
self
->
send
(
aut
,
ok_atom
_v
);
expect
((
ok_atom
),
from
(
self
).
to
(
aut
).
with
(
_
));
CAF_CHECK_EQUAL
(
t
.
schedule
().
size
(),
1u
);
CAF_CHECK_EQUAL
(
t
.
actor_lookup
().
size
(),
1u
);
...
...
@@ -131,14 +131,14 @@ CAF_TEST(override_receive_timeout) {
CAF_TEST
(
multi_timeout
)
{
// Have AUT call t.set_multi_timeout().
self
->
send
(
aut
,
add_atom
::
value
);
self
->
send
(
aut
,
add_atom
_v
);
expect
((
add_atom
),
from
(
self
).
to
(
aut
).
with
(
_
));
CAF_CHECK_EQUAL
(
t
.
schedule
().
size
(),
1u
);
CAF_CHECK_EQUAL
(
t
.
actor_lookup
().
size
(),
1u
);
// Advance time just a little bit.
t
.
advance_time
(
seconds
(
5
));
// Have AUT call t.set_multi_timeout() again.
self
->
send
(
aut
,
add_atom
::
value
);
self
->
send
(
aut
,
add_atom
_v
);
expect
((
add_atom
),
from
(
self
).
to
(
aut
).
with
(
_
));
CAF_CHECK_EQUAL
(
t
.
schedule
().
size
(),
2u
);
CAF_CHECK_EQUAL
(
t
.
actor_lookup
().
size
(),
2u
);
...
...
@@ -158,14 +158,14 @@ CAF_TEST(multi_timeout) {
CAF_TEST
(
mixed_receive_and_multi_timeouts
)
{
// Have AUT call t.set_receive_timeout().
self
->
send
(
aut
,
add_atom
::
value
);
self
->
send
(
aut
,
add_atom
_v
);
expect
((
add_atom
),
from
(
self
).
to
(
aut
).
with
(
_
));
CAF_CHECK_EQUAL
(
t
.
schedule
().
size
(),
1u
);
CAF_CHECK_EQUAL
(
t
.
actor_lookup
().
size
(),
1u
);
// Advance time just a little bit.
t
.
advance_time
(
seconds
(
5
));
// Have AUT call t.set_multi_timeout() again.
self
->
send
(
aut
,
ok_atom
::
value
);
self
->
send
(
aut
,
ok_atom
_v
);
expect
((
ok_atom
),
from
(
self
).
to
(
aut
).
with
(
_
));
CAF_CHECK_EQUAL
(
t
.
schedule
().
size
(),
2u
);
CAF_CHECK_EQUAL
(
t
.
actor_lookup
().
size
(),
2u
);
...
...
@@ -186,7 +186,7 @@ CAF_TEST(mixed_receive_and_multi_timeouts) {
CAF_TEST
(
single_request_timeout
)
{
// Have AUT call t.set_request_timeout().
self
->
send
(
aut
,
put_atom
::
value
);
self
->
send
(
aut
,
put_atom
_v
);
expect
((
put_atom
),
from
(
self
).
to
(
aut
).
with
(
_
));
CAF_CHECK_EQUAL
(
t
.
schedule
().
size
(),
1u
);
CAF_CHECK_EQUAL
(
t
.
actor_lookup
().
size
(),
1u
);
...
...
@@ -200,14 +200,14 @@ CAF_TEST(single_request_timeout) {
CAF_TEST
(
mixed_receive_and_request_timeouts
)
{
// Have AUT call t.set_receive_timeout().
self
->
send
(
aut
,
ok_atom
::
value
);
self
->
send
(
aut
,
ok_atom
_v
);
expect
((
ok_atom
),
from
(
self
).
to
(
aut
).
with
(
_
));
CAF_CHECK_EQUAL
(
t
.
schedule
().
size
(),
1u
);
CAF_CHECK_EQUAL
(
t
.
actor_lookup
().
size
(),
1u
);
// Cause the request timeout to arrive later.
t
.
advance_time
(
seconds
(
5
));
// Have AUT call t.set_request_timeout().
self
->
send
(
aut
,
put_atom
::
value
);
self
->
send
(
aut
,
put_atom
_v
);
expect
((
put_atom
),
from
(
self
).
to
(
aut
).
with
(
_
));
CAF_CHECK_EQUAL
(
t
.
schedule
().
size
(),
2u
);
CAF_CHECK_EQUAL
(
t
.
actor_lookup
().
size
(),
2u
);
...
...
libcaf_core/test/actor_lifetime.cpp
View file @
21281a6c
...
...
@@ -16,21 +16,16 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE actor_lifetime
#include "caf/test/unit_test.hpp"
#include <mutex>
#include "core-test.hpp"
#include <atomic>
#include <condition_variable>
#include <mutex>
#include "caf/all.hpp"
#include "caf/test/dsl.hpp"
using
check_atom
=
caf
::
atom_constant
<
caf
::
atom
(
"check"
)
>
;
using
namespace
caf
;
namespace
{
...
...
@@ -64,9 +59,7 @@ public:
behavior
make_behavior
()
override
{
return
{
[
=
](
int
x
)
{
return
x
;
}
[
=
](
int
x
)
{
return
x
;
},
};
}
};
...
...
@@ -78,7 +71,7 @@ behavior tester(event_based_actor* self, const actor& aut) {
// must be still alive at this point
CAF_CHECK_EQUAL
(
s_testees
.
load
(),
1
);
CAF_CHECK_EQUAL
(
msg
.
reason
,
exit_reason
::
user_shutdown
);
self
->
send
(
self
,
check_atom
::
value
);
self
->
send
(
self
,
ok_atom_v
);
});
self
->
link_to
(
aut
);
}
else
{
...
...
@@ -90,7 +83,7 @@ behavior tester(event_based_actor* self, const actor& aut) {
// another worker thread; by waiting some milliseconds, we make sure
// testee had enough time to return control to the scheduler
// which in turn destroys it by dropping the last remaining reference
self
->
send
(
self
,
check_atom
::
value
);
self
->
send
(
self
,
ok_atom_v
);
});
self
->
monitor
(
aut
);
}
...
...
@@ -101,7 +94,7 @@ behavior tester(event_based_actor* self, const actor& aut) {
s_cv
.
notify_one
();
}
return
{
[
self
](
chec
k_atom
)
{
[
self
](
o
k_atom
)
{
{
// make sure aut's dtor and on_exit() have been called
std
::
unique_lock
<
std
::
mutex
>
guard
{
s_mtx
};
while
(
!
s_testee_cleanup_done
.
load
())
...
...
@@ -110,15 +103,13 @@ behavior tester(event_based_actor* self, const actor& aut) {
CAF_CHECK_EQUAL
(
s_testees
.
load
(),
0
);
CAF_CHECK_EQUAL
(
s_pending_on_exits
.
load
(),
0
);
self
->
quit
();
}
}
,
};
}
struct
config
:
actor_system_config
{
config
()
{
set
(
"scheduler.policy"
,
atom
(
"testing"
)
);
set
(
"scheduler.policy"
,
"testing"
);
}
};
...
...
@@ -165,7 +156,7 @@ struct fixture {
}
// Run the exit_msg.
sched
.
run_once
();
//expect((exit_msg), from(tst_driver).to(tst_subject));
//
expect((exit_msg), from(tst_driver).to(tst_subject));
{
// Resume driver.
std
::
unique_lock
<
std
::
mutex
>
guard
{
s_mtx
};
s_testee_cleanup_done
=
true
;
...
...
@@ -184,7 +175,7 @@ struct fixture {
}
// namespace
CAF_TEST
(
destructor_call
)
{
{
// lifetime scope of actor syst
me
{
// lifetime scope of actor syst
em
actor_system_config
cfg
;
actor_system
system
{
cfg
};
system
.
spawn
<
testee
>
();
...
...
libcaf_core/test/actor_pool.cpp
View file @
21281a6c
...
...
@@ -88,7 +88,7 @@ CAF_TEST(round_robin_actor_pool) {
scoped_actor
self
{
system
};
auto
pool
=
actor_pool
::
make
(
&
context
,
5
,
spawn_worker
,
actor_pool
::
round_robin
());
self
->
send
(
pool
,
sys_atom
::
value
,
put_atom
::
value
,
spawn_worker
());
self
->
send
(
pool
,
sys_atom
_v
,
put_atom_v
,
spawn_worker
());
std
::
vector
<
actor
>
workers
;
for
(
int32_t
i
=
0
;
i
<
6
;
++
i
)
{
self
->
request
(
pool
,
infinite
,
i
,
i
)
...
...
@@ -103,7 +103,7 @@ CAF_TEST(round_robin_actor_pool) {
}
CAF_CHECK_EQUAL
(
workers
.
size
(),
6u
);
CAF_CHECK
(
std
::
unique
(
workers
.
begin
(),
workers
.
end
())
==
workers
.
end
());
self
->
request
(
pool
,
infinite
,
sys_atom
::
value
,
get_atom
::
value
)
self
->
request
(
pool
,
infinite
,
sys_atom
_v
,
get_atom_v
)
.
receive
(
[
&
](
std
::
vector
<
actor
>&
ws
)
{
std
::
sort
(
workers
.
begin
(),
workers
.
end
());
...
...
@@ -121,7 +121,7 @@ CAF_TEST(round_robin_actor_pool) {
bool
success
=
false
;
size_t
i
=
0
;
while
(
!
success
&&
++
i
<=
10
)
{
self
->
request
(
pool
,
infinite
,
sys_atom
::
value
,
get_atom
::
value
)
self
->
request
(
pool
,
infinite
,
sys_atom
_v
,
get_atom_v
)
.
receive
(
[
&
](
std
::
vector
<
actor
>&
ws
)
{
success
=
workers
.
size
()
==
ws
.
size
();
...
...
libcaf_core/test/actor_registry.cpp
View file @
21281a6c
...
...
@@ -36,6 +36,8 @@ behavior dummy() {
using
foo_atom
=
atom_constant
<
atom
(
"foo"
)
>
;
constexpr
auto
foo_atom_v
=
foo_atom
::
value
;
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
actor_registry_tests
,
test_coordinator_fixture
<>
)
...
...
@@ -43,12 +45,12 @@ CAF_TEST_FIXTURE_SCOPE(actor_registry_tests, test_coordinator_fixture<>)
CAF_TEST
(
erase
)
{
// CAF registers a few actors by itself.
auto
baseline
=
sys
.
registry
().
named_actors
().
size
();
sys
.
registry
().
put
(
foo_atom
::
value
,
sys
.
spawn
(
dummy
));
sys
.
registry
().
put
(
foo_atom
_v
,
sys
.
spawn
(
dummy
));
CAF_CHECK_EQUAL
(
sys
.
registry
().
named_actors
().
size
(),
baseline
+
1u
);
self
->
send
(
sys
.
registry
().
get
<
actor
>
(
foo_atom
::
value
),
42
);
self
->
send
(
sys
.
registry
().
get
<
actor
>
(
foo_atom
_v
),
42
);
run
();
expect
((
int
),
from
(
_
).
to
(
self
).
with
(
42
));
sys
.
registry
().
erase
(
foo_atom
::
value
);
sys
.
registry
().
erase
(
foo_atom
_v
);
CAF_CHECK_EQUAL
(
sys
.
registry
().
named_actors
().
size
(),
baseline
);
}
...
...
libcaf_core/test/atom.cpp
View file @
21281a6c
...
...
@@ -76,10 +76,10 @@ struct send_to_self {
CAF_TEST
(
receive_atoms
)
{
scoped_actor
self
{
system
};
send_to_self
f
{
self
.
ptr
()};
f
(
foo_atom
::
value
,
static_cast
<
uint32_t
>
(
42
));
f
(
abc_atom
::
value
,
def_atom
::
value
,
"cstring"
);
f
(
foo_atom
_v
,
static_cast
<
uint32_t
>
(
42
));
f
(
abc_atom
_v
,
def_atom_v
,
"cstring"
);
f
(
1.
f
);
f
(
a_atom
::
value
,
b_atom
::
value
,
c_atom
::
value
,
23.
f
);
f
(
a_atom
_v
,
b_atom_v
,
c_atom
::
value
,
23.
f
);
bool
matched_pattern
[
3
]
=
{
false
,
false
,
false
};
int
i
=
0
;
CAF_MESSAGE
(
"start receive loop"
);
...
...
@@ -106,7 +106,7 @@ CAF_TEST(receive_atoms) {
}
);
atom_value
x
=
atom
(
"abc"
);
atom_value
y
=
abc_atom
::
value
;
atom_value
y
=
abc_atom
_v
;
CAF_CHECK_EQUAL
(
x
,
y
);
auto
msg
=
make_message
(
atom
(
"abc"
));
self
->
send
(
self
,
msg
);
...
...
@@ -131,14 +131,9 @@ testee::behavior_type testee_impl(testee::pointer self) {
CAF_TEST
(
request_atom_constants
)
{
scoped_actor
self
{
system
};
auto
tst
=
system
.
spawn
(
testee_impl
);
self
->
request
(
tst
,
infinite
,
abc_atom
::
value
).
receive
(
[](
int
i
)
{
CAF_CHECK_EQUAL
(
i
,
42
);
},
[
&
](
error
&
err
)
{
CAF_FAIL
(
"err: "
<<
system
.
render
(
err
));
}
);
self
->
request
(
tst
,
infinite
,
abc_atom_v
)
.
receive
([](
int
i
)
{
CAF_CHECK_EQUAL
(
i
,
42
);
},
[
&
](
error
&
err
)
{
CAF_FAIL
(
"err: "
<<
system
.
render
(
err
));
});
}
CAF_TEST
(
runtime_conversion
)
{
...
...
libcaf_core/test/constructor_attach.cpp
View file @
21281a6c
...
...
@@ -16,33 +16,25 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE constructor_attach
#include "caf/test/unit_test.hpp"
#include "caf/all.hpp"
#include "core-test.hpp"
using
namespace
caf
;
namespace
{
using
die_atom
=
atom_constant
<
atom
(
"die"
)
>
;
using
done_atom
=
atom_constant
<
atom
(
"done"
)
>
;
class
testee
:
public
event_based_actor
{
public:
testee
(
actor_config
&
cfg
,
actor
buddy
)
:
event_based_actor
(
cfg
)
{
attach_functor
([
=
](
const
error
&
reason
)
{
send
(
buddy
,
done_atom
::
value
,
reason
);
});
attach_functor
([
=
](
const
error
&
rsn
)
{
send
(
buddy
,
ok_atom_v
,
rsn
);
});
}
behavior
make_behavior
()
override
{
return
{
[
=
](
die_atom
)
{
quit
(
exit_reason
::
user_shutdown
);
}
[
=
](
delete_atom
)
{
quit
(
exit_reason
::
user_shutdown
);
},
};
}
};
...
...
@@ -50,9 +42,9 @@ public:
class
spawner
:
public
event_based_actor
{
public:
spawner
(
actor_config
&
cfg
)
:
event_based_actor
(
cfg
),
downs_
(
0
),
testee_
(
spawn
<
testee
,
monitored
>
(
this
))
{
:
event_based_actor
(
cfg
),
downs_
(
0
),
testee_
(
spawn
<
testee
,
monitored
>
(
this
))
{
set_down_handler
([
=
](
down_msg
&
msg
)
{
CAF_CHECK_EQUAL
(
msg
.
reason
,
exit_reason
::
user_shutdown
);
CAF_CHECK_EQUAL
(
msg
.
source
,
testee_
.
address
());
...
...
@@ -63,15 +55,13 @@ public:
behavior
make_behavior
()
override
{
return
{
[
=
](
done
_atom
,
const
error
&
reason
)
{
[
=
](
ok
_atom
,
const
error
&
reason
)
{
CAF_CHECK_EQUAL
(
reason
,
exit_reason
::
user_shutdown
);
if
(
++
downs_
==
2
)
{
quit
(
reason
);
}
},
[
=
](
die_atom
x
)
{
return
delegate
(
testee_
,
x
);
}
[
=
](
delete_atom
x
)
{
return
delegate
(
testee_
,
x
);
},
};
}
...
...
@@ -89,5 +79,5 @@ private:
CAF_TEST
(
constructor_attach
)
{
actor_system_config
cfg
;
actor_system
system
{
cfg
};
anon_send
(
system
.
spawn
<
spawner
>
(),
d
ie_atom
::
value
);
anon_send
(
system
.
spawn
<
spawner
>
(),
d
elete_atom_v
);
}
libcaf_core/test/continuous_streaming.cpp
View file @
21281a6c
...
...
@@ -165,8 +165,8 @@ CAF_TEST(depth_3_pipeline_with_fork) {
auto
snk2
=
sys
.
spawn
(
sum_up
);
auto
&
st
=
deref
<
stream_multiplexer_actor
>
(
stg
).
state
;
CAF_MESSAGE
(
"connect sinks to the stage (fork)"
);
self
->
send
(
snk1
,
join_atom
::
value
,
stg
);
self
->
send
(
snk2
,
join_atom
::
value
,
stg
);
self
->
send
(
snk1
,
join_atom
_v
,
stg
);
self
->
send
(
snk2
,
join_atom
_v
,
stg
);
consume_messages
();
CAF_CHECK_EQUAL
(
st
.
stage
->
out
().
num_paths
(),
2u
);
CAF_MESSAGE
(
"connect source to the stage (fork)"
);
...
...
@@ -189,7 +189,7 @@ CAF_TEST(depth_3_pipeline_with_join) {
auto
snk
=
sys
.
spawn
(
sum_up
);
auto
&
st
=
deref
<
stream_multiplexer_actor
>
(
stg
).
state
;
CAF_MESSAGE
(
"connect sink to the stage"
);
self
->
send
(
snk
,
join_atom
::
value
,
stg
);
self
->
send
(
snk
,
join_atom
_v
,
stg
);
consume_messages
();
CAF_CHECK_EQUAL
(
st
.
stage
->
out
().
num_paths
(),
1u
);
CAF_MESSAGE
(
"connect sources to the stage (join)"
);
...
...
@@ -212,8 +212,8 @@ CAF_TEST(closing_downstreams_before_end_of_stream) {
auto
snk2
=
sys
.
spawn
(
sum_up
);
auto
&
st
=
deref
<
stream_multiplexer_actor
>
(
stg
).
state
;
CAF_MESSAGE
(
"connect sinks to the stage (fork)"
);
self
->
send
(
snk1
,
join_atom
::
value
,
stg
);
self
->
send
(
snk2
,
join_atom
::
value
,
stg
);
self
->
send
(
snk1
,
join_atom
_v
,
stg
);
self
->
send
(
snk2
,
join_atom
_v
,
stg
);
consume_messages
();
CAF_CHECK_EQUAL
(
st
.
stage
->
out
().
num_paths
(),
2u
);
CAF_MESSAGE
(
"connect source to the stage (fork)"
);
...
...
@@ -233,7 +233,7 @@ CAF_TEST(closing_downstreams_before_end_of_stream) {
CAF_REQUIRE_GREATER
(
next_pending
,
0
);
auto
sink1_result
=
sum
(
next_pending
-
1
);
CAF_MESSAGE
(
"gracefully close sink 1, next pending: "
<<
next_pending
);
self
->
send
(
stg
,
close_atom
::
value
,
0
);
self
->
send
(
stg
,
close_atom
_v
,
0
);
expect
((
atom_value
,
int32_t
),
from
(
self
).
to
(
stg
));
CAF_MESSAGE
(
"ship remaining elements"
);
run
();
...
...
libcaf_core/test/detached_actors.cpp
View file @
21281a6c
...
...
@@ -52,7 +52,7 @@ CAF_TEST(shutdown_with_delayed_send) {
CAF_MESSAGE
(
"does sys shut down after spawning a detached actor that used "
"delayed_send?"
);
auto
f
=
[](
event_based_actor
*
self
)
->
behavior
{
self
->
delayed_send
(
self
,
std
::
chrono
::
nanoseconds
(
1
),
ok_atom
::
value
);
self
->
delayed_send
(
self
,
std
::
chrono
::
nanoseconds
(
1
),
ok_atom
_v
);
return
{
[
=
](
ok_atom
)
{
self
->
quit
();
...
...
@@ -66,7 +66,7 @@ CAF_TEST(shutdown_with_unhandled_delayed_send) {
CAF_MESSAGE
(
"does sys shut down after spawning a detached actor that used "
"delayed_send but didn't bother waiting for it?"
);
auto
f
=
[](
event_based_actor
*
self
)
{
self
->
delayed_send
(
self
,
std
::
chrono
::
nanoseconds
(
1
),
ok_atom
::
value
);
self
->
delayed_send
(
self
,
std
::
chrono
::
nanoseconds
(
1
),
ok_atom
_v
);
};
sys
.
spawn
<
detached
>
(
f
);
}
...
...
@@ -88,10 +88,10 @@ CAF_TEST(shutdown_delayed_send_loop) {
CAF_MESSAGE
(
"does sys shut down after spawning a detached actor that used "
"a delayed send loop and was interrupted via exit message?"
);
auto
f
=
[](
event_based_actor
*
self
)
->
behavior
{
self
->
delayed_send
(
self
,
std
::
chrono
::
milliseconds
(
1
),
ok_atom
::
value
);
self
->
delayed_send
(
self
,
std
::
chrono
::
milliseconds
(
1
),
ok_atom
_v
);
return
{
[
=
](
ok_atom
)
{
self
->
delayed_send
(
self
,
std
::
chrono
::
milliseconds
(
1
),
ok_atom
::
value
);
self
->
delayed_send
(
self
,
std
::
chrono
::
milliseconds
(
1
),
ok_atom
_v
);
},
};
};
...
...
libcaf_core/test/dynamic_spawn.cpp
View file @
21281a6c
This diff is collapsed.
Click to expand it.
libcaf_core/test/function_view.cpp
View file @
21281a6c
...
...
@@ -132,9 +132,9 @@ CAF_TEST(tuple_res_function_view) {
CAF_TEST
(
cell_function_view
)
{
auto
f
=
make_function_view
(
system
.
spawn
(
simple_cell
));
CAF_CHECK_EQUAL
(
f
(
get_atom
::
value
),
0
);
f
(
put_atom
::
value
,
1024
);
CAF_CHECK_EQUAL
(
f
(
get_atom
::
value
),
1024
);
CAF_CHECK_EQUAL
(
f
(
get_atom
_v
),
0
);
f
(
put_atom
_v
,
1024
);
CAF_CHECK_EQUAL
(
f
(
get_atom
_v
),
1024
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_core/test/fused_downstream_manager.cpp
View file @
21281a6c
...
...
@@ -43,10 +43,6 @@ using int_downstream_manager = broadcast_downstream_manager<int>;
using
string_downstream_manager
=
broadcast_downstream_manager
<
string
>
;
using
ints_atom
=
atom_constant
<
atom
(
"ints"
)
>
;
using
strings_atom
=
atom_constant
<
atom
(
"strings"
)
>
;
template
<
class
T
>
void
push
(
std
::
deque
<
T
>&
xs
,
downstream
<
T
>&
out
,
size_t
num
)
{
auto
n
=
std
::
min
(
num
,
xs
.
size
());
...
...
@@ -107,7 +103,7 @@ TESTEE_STATE(sum_up) {
TESTEE
(
sum_up
)
{
using
intptr
=
int
*
;
return
{
[
=
](
stream
<
int32_t
>
&
in
)
{
[
=
](
stream
<
int32_t
>
in
)
{
return
attach_stream_sink
(
self
,
// input stream
...
...
@@ -123,7 +119,7 @@ TESTEE(sum_up) {
},
[
=
](
join_atom
,
actor
src
)
{
CAF_MESSAGE
(
self
->
name
()
<<
" joins a stream"
);
self
->
send
(
self
*
src
,
join_atom
::
value
,
ints_atom
::
value
);
self
->
send
(
self
*
src
,
join_atom
_v
,
int32_t
{
0
}
);
},
};
}
...
...
@@ -134,7 +130,7 @@ TESTEE_STATE(collect) {
TESTEE
(
collect
)
{
return
{
[
=
](
stream
<
string
>
&
in
)
{
[
=
](
stream
<
string
>
in
)
{
return
attach_stream_sink
(
self
,
// input stream
...
...
@@ -154,7 +150,7 @@ TESTEE(collect) {
},
[
=
](
join_atom
,
actor
src
)
{
CAF_MESSAGE
(
self
->
name
()
<<
" joins a stream"
);
self
->
send
(
self
*
src
,
join_atom
::
value
,
strings_atom
::
value
);
self
->
send
(
self
*
src
,
join_atom
_v
,
"dummy"
);
},
};
}
...
...
@@ -223,38 +219,33 @@ TESTEE_STATE(stream_multiplexer) {
TESTEE
(
stream_multiplexer
)
{
self
->
state
.
stage
=
make_counted
<
fused_stage
>
(
self
);
return
{
[
=
](
join_atom
,
int
s_atom
)
{
[
=
](
join_atom
,
int
32_t
)
{
auto
&
stg
=
self
->
state
.
stage
;
CAF_MESSAGE
(
"received 'join' request for integers"
);
auto
result
=
stg
->
add_unchecked_outbound_path
<
int
>
();
stg
->
out
().
assign
<
int_downstream_manager
>
(
result
);
return
result
;
},
[
=
](
join_atom
,
st
rings_atom
)
{
[
=
](
join_atom
,
st
d
::
string
)
{
auto
&
stg
=
self
->
state
.
stage
;
CAF_MESSAGE
(
"received 'join' request for strings"
);
auto
result
=
stg
->
add_unchecked_outbound_path
<
string
>
();
stg
->
out
().
assign
<
string_downstream_manager
>
(
result
);
return
result
;
},
[
=
](
const
stream
<
int32_t
>&
in
)
{
[
=
](
stream
<
int32_t
>
in
)
{
CAF_MESSAGE
(
"received handshake for integers"
);
CAF_MESSAGE
(
self
->
current_mailbox_element
()
->
content
());
return
self
->
state
.
stage
->
add_unchecked_inbound_path
(
in
);
},
[
=
](
const
stream
<
string
>&
in
)
{
[
=
](
stream
<
string
>
in
)
{
CAF_MESSAGE
(
"received handshake for strings"
);
return
self
->
state
.
stage
->
add_unchecked_inbound_path
(
in
);
},
};
}
struct
config
:
actor_system_config
{
config
()
{
add_message_types
<
id_block
::
core_test
>
();
}
};
using
fixture
=
test_coordinator_fixture
<
config
>
;
using
fixture
=
test_coordinator_fixture
<>
;
}
// namespace
...
...
@@ -262,7 +253,7 @@ using fixture = test_coordinator_fixture<config>;
CAF_TEST_FIXTURE_SCOPE
(
fused_downstream_manager_tests
,
fixture
)
CAF_TEST
(
depth_3_pipeline_with_fork
)
{
CAF_TEST
_DISABLED
(
depth_3_pipeline_with_fork
)
{
auto
src1
=
sys
.
spawn
(
int_file_reader
,
50u
);
auto
src2
=
sys
.
spawn
(
string_file_reader
,
50u
);
auto
stg
=
sys
.
spawn
(
stream_multiplexer
);
...
...
@@ -270,8 +261,8 @@ CAF_TEST(depth_3_pipeline_with_fork) {
auto
snk2
=
sys
.
spawn
(
collect
);
auto
&
st
=
deref
<
stream_multiplexer_actor
>
(
stg
).
state
;
CAF_MESSAGE
(
"connect sinks to the fused stage"
);
self
->
send
(
snk1
,
join_atom
::
value
,
stg
);
self
->
send
(
snk2
,
join_atom
::
value
,
stg
);
self
->
send
(
snk1
,
join_atom
_v
,
stg
);
self
->
send
(
snk2
,
join_atom
_v
,
stg
);
sched
.
run
();
CAF_CHECK_EQUAL
(
st
.
stage
->
out
().
num_paths
(),
2u
);
CAF_CHECK_EQUAL
(
st
.
stage
->
inbound_paths
().
size
(),
0u
);
...
...
libcaf_core/test/local_group.cpp
View file @
21281a6c
...
...
@@ -76,11 +76,11 @@ CAF_TEST(class_based_joined_at_spawn) {
});
// make sure all actors start at 0
for
(
auto
&
f
:
fs
)
CAF_CHECK_EQUAL
(
f
(
get_atom
::
value
),
0
);
CAF_CHECK_EQUAL
(
f
(
get_atom
_v
),
0
);
// send a put to all actors via the group and make sure they change state
self
->
send
(
grp
,
put_atom
::
value
,
42
);
self
->
send
(
grp
,
put_atom
_v
,
42
);
for
(
auto
&
f
:
fs
)
CAF_CHECK_EQUAL
(
f
(
get_atom
::
value
),
42
);
CAF_CHECK_EQUAL
(
f
(
get_atom
_v
),
42
);
// shutdown all actors
for
(
auto
&
x
:
xs
)
self
->
send_exit
(
x
,
exit_reason
::
user_shutdown
);
...
...
libcaf_core/test/local_migration.cpp
View file @
21281a6c
...
...
@@ -67,7 +67,7 @@ behavior pseudo_mm(event_based_actor* self, const actor& dest) {
return {
[=](migrate_atom, const std::string& name, std::vector<char>& buf) {
CAF_CHECK(name == "migratable_actor");
self->delegate(dest, sys_atom
::value, migrate_atom::value
,
self->delegate(dest, sys_atom
_v, migrate_atom_v
,
std::move(buf));
}
};
...
...
@@ -82,30 +82,30 @@ CAF_TEST(migrate_locally) {
auto mm1 = system.spawn(pseudo_mm, b);
{ // Lifetime scope of scoped_actor
scoped_actor self{system};
self->send(a, put_atom
::value
, 42);
self->send(a, put_atom
_v
, 42);
// migrate from a to b
self->request(a, infinite, sys_atom
::value
,
migrate_atom
::value
, mm1).receive(
self->request(a, infinite, sys_atom
_v
,
migrate_atom
_v
, mm1).receive(
[&](ok_atom, const actor_addr& dest) {
CAF_CHECK(dest == b);
}
);
self->request(a, infinite, get_atom
::value
).receive(
self->request(a, infinite, get_atom
_v
).receive(
[&](int result) {
CAF_CHECK(result == 42);
CAF_CHECK(self->current_sender() == b.address());
}
);
auto mm2 = system.spawn(pseudo_mm, a);
self->send(b, put_atom
::value
, 23);
self->send(b, put_atom
_v
, 23);
// migrate back from b to a
self->request(b, infinite, sys_atom
::value
,
migrate_atom
::value
, mm2).receive(
self->request(b, infinite, sys_atom
_v
,
migrate_atom
_v
, mm2).receive(
[&](ok_atom, const actor_addr& dest) {
CAF_CHECK(dest == a);
}
);
self->request(b, infinite, get_atom
::value
).receive(
self->request(b, infinite, get_atom
_v
).receive(
[&](int result) {
CAF_CHECK(result == 23);
CAF_CHECK(self->current_sender() == a.address());
...
...
libcaf_core/test/match.cpp
View file @
21281a6c
...
...
@@ -117,16 +117,16 @@ CAF_TEST(atom_constants) {
invoked
[
1
]
=
true
;
}
};
CAF_CHECK_EQUAL
(
invoke
(
expr
,
atom_value
{
ok_atom
::
value
}),
-
1
);
CAF_CHECK_EQUAL
(
invoke
(
expr
,
atom_value
{
hi_atom
::
value
}),
0
);
CAF_CHECK_EQUAL
(
invoke
(
expr
,
atom_value
{
ho_atom
::
value
}),
1
);
CAF_CHECK_EQUAL
(
invoke
(
expr
,
atom_value
{
ok_atom
_v
}),
-
1
);
CAF_CHECK_EQUAL
(
invoke
(
expr
,
atom_value
{
hi_atom
_v
}),
0
);
CAF_CHECK_EQUAL
(
invoke
(
expr
,
atom_value
{
ho_atom
_v
}),
1
);
}
CAF_TEST
(
manual_matching
)
{
using
foo_atom
=
atom_constant
<
atom
(
"foo"
)
>
;
using
bar_atom
=
atom_constant
<
atom
(
"bar"
)
>
;
auto
msg1
=
make_message
(
foo_atom
::
value
,
42
);
auto
msg2
=
make_message
(
bar_atom
::
value
,
42
);
auto
msg1
=
make_message
(
foo_atom
_v
,
42
);
auto
msg2
=
make_message
(
bar_atom
_v
,
42
);
CAF_MESSAGE
(
"check individual message elements"
);
CAF_CHECK
((
msg1
.
match_element
<
int
>
(
1
)));
CAF_CHECK
((
msg2
.
match_element
<
int
>
(
1
)));
...
...
libcaf_core/test/message.cpp
View file @
21281a6c
...
...
@@ -38,12 +38,8 @@ using std::vector;
using
namespace
caf
;
CAF_TEST
(
apply
)
{
auto
f1
=
[]
{
CAF_ERROR
(
"f1 invoked!"
);
};
auto
f2
=
[](
int
i
)
{
CAF_CHECK_EQUAL
(
i
,
42
);
};
auto
f1
=
[]
{
CAF_ERROR
(
"f1 invoked!"
);
};
auto
f2
=
[](
int
i
)
{
CAF_CHECK_EQUAL
(
i
,
42
);
};
auto
m
=
make_message
(
42
);
m
.
apply
(
f1
);
m
.
apply
(
f2
);
...
...
@@ -51,14 +47,12 @@ CAF_TEST(apply) {
CAF_TEST
(
drop
)
{
auto
m1
=
make_message
(
1
,
2
,
3
,
4
,
5
);
std
::
vector
<
message
>
messages
{
m1
,
make_message
(
2
,
3
,
4
,
5
),
make_message
(
3
,
4
,
5
),
make_message
(
4
,
5
),
make_message
(
5
),
message
{}
};
std
::
vector
<
message
>
messages
{
m1
,
make_message
(
2
,
3
,
4
,
5
),
make_message
(
3
,
4
,
5
),
make_message
(
4
,
5
),
make_message
(
5
),
message
{}};
for
(
size_t
i
=
0
;
i
<
messages
.
size
();
++
i
)
{
CAF_CHECK_EQUAL
(
to_string
(
m1
.
drop
(
i
)),
to_string
(
messages
[
i
]));
}
...
...
@@ -78,10 +72,7 @@ CAF_TEST(extract1) {
auto
m5
=
make_message
(
1.0
,
2.0
,
3.0
,
1
,
2
);
auto
m6
=
make_message
(
1
,
2
,
1.0
,
2.0
,
3.0
,
1
,
2
);
auto
m7
=
make_message
(
1.0
,
1
,
2
,
3
,
4
,
2.0
,
3.0
);
message_handler
f
{
[](
int
,
int
)
{
},
[](
float
,
float
)
{
}
};
message_handler
f
{[](
int
,
int
)
{},
[](
float
,
float
)
{}};
auto
m1s
=
to_string
(
m1
);
CAF_CHECK_EQUAL
(
to_string
(
m2
.
extract
(
f
)),
m1s
);
CAF_CHECK_EQUAL
(
to_string
(
m3
.
extract
(
f
)),
m1s
);
...
...
@@ -95,10 +86,7 @@ CAF_TEST(extract2) {
auto
m1
=
make_message
(
1
);
CAF_CHECK
(
m1
.
extract
([](
int
)
{}).
empty
());
auto
m2
=
make_message
(
1.0
,
2
,
3
,
4.0
);
auto
m3
=
m2
.
extract
({
[](
int
,
int
)
{
},
[](
double
,
double
)
{
}
});
auto
m3
=
m2
.
extract
({[](
int
,
int
)
{},
[](
double
,
double
)
{}});
// check for false positives through collapsing
CAF_CHECK_EQUAL
(
to_string
(
m3
),
to_string
(
make_message
(
1.0
,
4.0
)));
}
...
...
@@ -107,12 +95,11 @@ CAF_TEST(extract_opts) {
auto
f
=
[](
std
::
vector
<
std
::
string
>
xs
,
std
::
vector
<
std
::
string
>
remainder
)
{
std
::
string
filename
;
size_t
log_level
=
0
;
auto
res
=
message_builder
(
xs
.
begin
(),
xs
.
end
()).
extract_opts
({
{
"version,v"
,
"print version"
},
{
"log-level,l"
,
"set the log level"
,
log_level
},
{
"file,f"
,
"set output file"
,
filename
},
{
"whatever"
,
"do whatever"
}
});
auto
res
=
message_builder
(
xs
.
begin
(),
xs
.
end
())
.
extract_opts
({{
"version,v"
,
"print version"
},
{
"log-level,l"
,
"set the log level"
,
log_level
},
{
"file,f"
,
"set output file"
,
filename
},
{
"whatever"
,
"do whatever"
}});
CAF_CHECK_EQUAL
(
res
.
opts
.
count
(
"file"
),
1u
);
CAF_CHECK
(
res
.
remainder
.
size
()
==
remainder
.
size
());
for
(
size_t
i
=
0
;
i
<
res
.
remainder
.
size
()
&&
i
<
remainder
.
size
();
++
i
)
{
...
...
@@ -150,37 +137,33 @@ CAF_TEST(extract_opts) {
auto
msg
=
make_message
(
"-f"
,
"42"
,
"-b"
,
"1337"
);
auto
foo
=
0
;
auto
bar
=
0
;
auto
r
=
msg
.
extract_opts
({
{
"foo,f"
,
"foo desc"
,
foo
}
});
auto
r
=
msg
.
extract_opts
({{
"foo,f"
,
"foo desc"
,
foo
}});
CAF_CHECK
(
r
.
opts
.
count
(
"foo"
)
>
0
);
CAF_CHECK_EQUAL
(
foo
,
42
);
CAF_CHECK_EQUAL
(
bar
,
0
);
CAF_CHECK
(
!
r
.
error
.
empty
());
// -b is an unknown option
CAF_CHECK
(
!
r
.
remainder
.
empty
()
&&
to_string
(
r
.
remainder
)
==
to_string
(
make_message
(
"-b"
,
"1337"
)));
r
=
r
.
remainder
.
extract_opts
({
{
"bar,b"
,
"bar desc"
,
bar
}
});
r
=
r
.
remainder
.
extract_opts
({{
"bar,b"
,
"bar desc"
,
bar
}});
CAF_CHECK
(
r
.
opts
.
count
(
"bar"
)
>
0
);
CAF_CHECK_EQUAL
(
bar
,
1337
);
CAF_CHECK
(
r
.
error
.
empty
());
}
CAF_TEST
(
type_token
)
{
auto
m1
=
make_message
(
get_atom
::
value
);
auto
m1
=
make_message
(
get_atom
_v
);
CAF_CHECK_EQUAL
(
m1
.
type_token
(),
make_type_token
<
get_atom
>
());
}
CAF_TEST
(
concat
)
{
auto
m1
=
make_message
(
get_atom
::
value
);
auto
m1
=
make_message
(
get_atom
_v
);
auto
m2
=
make_message
(
uint32_t
{
1
});
auto
m3
=
message
::
concat
(
m1
,
m2
);
CAF_CHECK_EQUAL
(
to_string
(
m3
),
to_string
(
m1
+
m2
));
CAF_CHECK_EQUAL
(
to_string
(
m3
),
"('get', 1)"
);
auto
m4
=
make_message
(
get_atom
::
value
,
uint32_t
{
1
},
get_atom
::
value
,
uint32_t
{
1
});
CAF_CHECK_EQUAL
(
to_string
(
message
::
concat
(
m3
,
message
{},
m1
,
m2
)),
to_string
(
m4
));
auto
m4
=
make_message
(
get_atom
_v
,
uint32_t
{
1
},
get_atom_v
,
uint32_t
{
1
});
CAF_CHECK_EQUAL
(
to_string
(
message
::
concat
(
m3
,
message
{},
m1
,
m2
)),
to_string
(
m4
));
}
template
<
class
...
Ts
>
...
...
@@ -216,8 +199,9 @@ CAF_TEST(strings_to_string) {
CAF_CHECK_EQUAL
(
to_string
(
msg2
),
R"__((["one", "two", "three"]))__"
);
auto
msg3
=
make_message
(
svec
{
"one"
,
"two"
},
"three"
,
"four"
,
svec
{
"five"
,
"six"
,
"seven"
});
CAF_CHECK
(
to_string
(
msg3
)
==
R"__((["one", "two"], "three", "four", ["five", "six", "seven"]))__"
);
CAF_CHECK
(
to_string
(
msg3
)
==
R"__((["one", "two"], "three", "four", ["five", "six", "seven"]))__"
);
auto
msg4
=
make_message
(
R"(this is a "test")"
);
CAF_CHECK_EQUAL
(
to_string
(
msg4
),
"(
\"
this is a
\\\"
test
\\\"\"
)"
);
}
...
...
libcaf_core/test/or_else.cpp
View file @
21281a6c
...
...
@@ -16,34 +16,34 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE or_else
#include "caf/test/unit_test.hpp"
#include "core-test.hpp"
#include "caf/all.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
#define ERROR_HANDLER [&](error& err) { CAF_FAIL(err); }
using
namespace
caf
;
namespace
{
using
a_atom
=
atom_constant
<
atom
(
"a"
)
>
;
using
b_atom
=
atom_constant
<
atom
(
"b"
)
>
;
using
c_atom
=
atom_constant
<
atom
(
"c"
)
>
;
message_handler
handle_a
()
{
return
[](
a_atom
)
{
return
1
;
};
return
{
[](
int8_t
)
{
return
"a"
;
},
};
}
message_handler
handle_b
()
{
return
[](
b_atom
)
{
return
2
;
};
return
{
[](
int16_t
)
{
return
"b"
;
},
};
}
message_handler
handle_c
()
{
return
[](
c_atom
)
{
return
3
;
};
return
{
[](
int32_t
)
{
return
"c"
;
},
};
}
struct
fixture
{
...
...
@@ -56,26 +56,16 @@ struct fixture {
void
run_testee
(
const
actor
&
testee
)
{
scoped_actor
self
{
system
};
self
->
request
(
testee
,
infinite
,
a_atom
::
value
).
receive
(
[](
int
i
)
{
CAF_CHECK_EQUAL
(
i
,
1
);
},
ERROR_HANDLER
);
self
->
request
(
testee
,
infinite
,
b_atom
::
value
).
receive
(
[](
int
i
)
{
CAF_CHECK_EQUAL
(
i
,
2
);
},
ERROR_HANDLER
);
self
->
request
(
testee
,
infinite
,
c_atom
::
value
).
receive
(
[](
int
i
)
{
CAF_CHECK_EQUAL
(
i
,
3
);
},
ERROR_HANDLER
);
self
->
request
(
testee
,
infinite
,
int8_t
{
1
})
.
receive
([](
const
std
::
string
&
str
)
{
CAF_CHECK_EQUAL
(
str
,
"a"
);
},
ERROR_HANDLER
);
self
->
request
(
testee
,
infinite
,
int16_t
{
1
})
.
receive
([](
const
std
::
string
&
str
)
{
CAF_CHECK_EQUAL
(
str
,
"b"
);
},
ERROR_HANDLER
);
self
->
request
(
testee
,
infinite
,
int32_t
{
1
})
.
receive
([](
const
std
::
string
&
str
)
{
CAF_CHECK_EQUAL
(
str
,
"c"
);
},
ERROR_HANDLER
);
self
->
send_exit
(
testee
,
exit_reason
::
user_shutdown
);
self
->
await_all_other_actors_done
();
}
};
...
...
@@ -84,28 +74,22 @@ struct fixture {
CAF_TEST_FIXTURE_SCOPE
(
atom_tests
,
fixture
)
CAF_TEST
(
composition1
)
{
run_testee
(
system
.
spawn
([
=
]
{
return
handle_a
().
or_else
(
handle_b
()).
or_else
(
handle_c
());
})
);
run_testee
(
system
.
spawn
(
[
=
]
{
return
handle_a
().
or_else
(
handle_b
()).
or_else
(
handle_c
());
}));
}
CAF_TEST
(
composition2
)
{
run_testee
(
system
.
spawn
([
=
]
{
return
handle_a
().
or_else
(
handle_b
()).
or_else
([](
c_atom
)
{
return
3
;
});
})
);
run_testee
(
system
.
spawn
([
=
]
{
return
handle_a
().
or_else
(
handle_b
()).
or_else
([](
int32_t
)
{
return
"c"
;
});
}));
}
CAF_TEST
(
composition3
)
{
run_testee
(
system
.
spawn
([
=
]
{
return
message_handler
{[](
a_atom
)
{
return
1
;
}}.
or_else
(
handle_b
()).
or_else
(
handle_c
());
})
);
run_testee
(
system
.
spawn
([
=
]
{
return
message_handler
{[](
int8_t
)
{
return
"a"
;
}}
.
or_else
(
handle_b
())
.
or_else
(
handle_c
());
}));
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_core/test/pipeline_streaming.cpp
View file @
21281a6c
...
...
@@ -318,7 +318,7 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) {
disallow
((
upstream_msg
::
ack_open
),
from
(
snk
).
to
(
src
));
disallow
((
upstream_msg
::
forced_drop
),
from
(
_
).
to
(
src
));
CAF_MESSAGE
(
"send 'ok' to trigger sink to handle open_stream_msg"
);
self
->
send
(
snk
,
ok_atom
::
value
);
self
->
send
(
snk
,
ok_atom
_v
);
expect
((
ok_atom
),
from
(
self
).
to
(
snk
));
expect
((
open_stream_msg
),
from
(
self
).
to
(
snk
));
expect
((
upstream_msg
::
ack_open
),
from
(
snk
).
to
(
src
));
...
...
libcaf_core/test/request_timeout.cpp
View file @
21281a6c
This diff is collapsed.
Click to expand it.
libcaf_core/test/selective_streaming.cpp
View file @
21281a6c
...
...
@@ -218,8 +218,8 @@ CAF_TEST(forking) {
auto
snk2
=
sys
.
spawn
(
log_consumer
);
sched
.
run
();
self
->
send
(
stg
*
src
,
level
::
all
);
self
->
send
(
snk1
*
stg
,
join_atom
::
value
,
level
::
trace
);
self
->
send
(
snk2
*
stg
,
join_atom
::
value
,
level
::
error
);
self
->
send
(
snk1
*
stg
,
join_atom
_v
,
level
::
trace
);
self
->
send
(
snk2
*
stg
,
join_atom
_v
,
level
::
error
);
sched
.
run
();
auto
&
st
=
deref
<
log_dispatcher_actor
>
(
stg
).
state
;
run_until
([
&
]
{
...
...
libcaf_core/test/serial_reply.cpp
View file @
21281a6c
...
...
@@ -41,11 +41,9 @@ CAF_TEST(test_serial_reply) {
actor_system
system
{
cfg
};
auto
mirror_behavior
=
[
=
](
event_based_actor
*
self
)
->
behavior
{
self
->
set_default_handler
(
reflect
);
return
{
[]
{
// nop
}
};
return
{[]
{
// nop
}};
};
auto
master
=
system
.
spawn
([
=
](
event_based_actor
*
self
)
{
CAF_MESSAGE
(
"ID of master: "
<<
self
->
id
());
...
...
@@ -55,48 +53,35 @@ CAF_TEST(test_serial_reply) {
auto
c2
=
self
->
spawn
<
linked
>
(
mirror_behavior
);
auto
c3
=
self
->
spawn
<
linked
>
(
mirror_behavior
);
auto
c4
=
self
->
spawn
<
linked
>
(
mirror_behavior
);
self
->
become
(
[
=
](
hi_atom
)
mutable
{
auto
rp
=
self
->
make_response_promise
();
CAF_MESSAGE
(
"received 'hi there'"
);
self
->
request
(
c0
,
infinite
,
sub0_atom
::
value
).
then
(
[
=
](
sub0_atom
)
mutable
{
CAF_MESSAGE
(
"received 'sub0'"
);
self
->
request
(
c1
,
infinite
,
sub1_atom
::
value
).
then
(
[
=
](
sub1_atom
)
mutable
{
CAF_MESSAGE
(
"received 'sub1'"
);
self
->
request
(
c2
,
infinite
,
sub2_atom
::
value
).
then
(
[
=
](
sub2_atom
)
mutable
{
CAF_MESSAGE
(
"received 'sub2'"
);
self
->
request
(
c3
,
infinite
,
sub3_atom
::
value
).
then
(
[
=
](
sub3_atom
)
mutable
{
CAF_MESSAGE
(
"received 'sub3'"
);
self
->
request
(
c4
,
infinite
,
sub4_atom
::
value
).
then
(
[
=
](
sub4_atom
)
mutable
{
CAF_MESSAGE
(
"received 'sub4'"
);
rp
.
deliver
(
ho_atom
::
value
);
}
);
}
);
}
);
}
);
}
);
}
);
self
->
become
([
=
](
hi_atom
)
mutable
{
auto
rp
=
self
->
make_response_promise
();
CAF_MESSAGE
(
"received 'hi there'"
);
self
->
request
(
c0
,
infinite
,
sub0_atom_v
).
then
([
=
](
sub0_atom
)
mutable
{
CAF_MESSAGE
(
"received 'sub0'"
);
self
->
request
(
c1
,
infinite
,
sub1_atom_v
).
then
([
=
](
sub1_atom
)
mutable
{
CAF_MESSAGE
(
"received 'sub1'"
);
self
->
request
(
c2
,
infinite
,
sub2_atom_v
).
then
([
=
](
sub2_atom
)
mutable
{
CAF_MESSAGE
(
"received 'sub2'"
);
self
->
request
(
c3
,
infinite
,
sub3_atom_v
)
.
then
([
=
](
sub3_atom
)
mutable
{
CAF_MESSAGE
(
"received 'sub3'"
);
self
->
request
(
c4
,
infinite
,
sub4_atom_v
)
.
then
([
=
](
sub4_atom
)
mutable
{
CAF_MESSAGE
(
"received 'sub4'"
);
rp
.
deliver
(
ho_atom_v
);
});
});
});
});
});
});
});
scoped_actor
self
{
system
};
CAF_MESSAGE
(
"ID of main: "
<<
self
->
id
());
self
->
request
(
master
,
infinite
,
hi_atom
::
value
).
receive
(
[](
ho_atom
)
{
CAF_MESSAGE
(
"received 'ho'"
);
},
[
&
](
const
error
&
err
)
{
CAF_ERROR
(
"Error: "
<<
self
->
system
().
render
(
err
));
}
);
self
->
request
(
master
,
infinite
,
hi_atom_v
)
.
receive
([](
ho_atom
)
{
CAF_MESSAGE
(
"received 'ho'"
);
},
[
&
](
const
error
&
err
)
{
CAF_ERROR
(
"Error: "
<<
self
->
system
().
render
(
err
));
});
CAF_REQUIRE
(
self
->
mailbox
().
size
()
==
0
);
}
libcaf_core/test/serialization.cpp
View file @
21281a6c
...
...
@@ -371,9 +371,8 @@ SERIALIZATION_TEST(atoms) {
auto
foo
=
atom
(
"foo"
);
CAF_CHECK_EQUAL
(
foo
,
roundtrip
(
foo
));
CAF_CHECK_EQUAL
(
foo
,
msg_roundtrip
(
foo
));
using
bar_atom
=
atom_constant
<
atom
(
"bar"
)
>
;
CAF_CHECK_EQUAL
(
bar_atom
::
value
,
roundtrip
(
atom
(
"bar"
)));
CAF_CHECK_EQUAL
(
bar_atom
::
value
,
msg_roundtrip
(
atom
(
"bar"
)));
CAF_CHECK_EQUAL
(
atom
(
"bar"
),
roundtrip
(
atom
(
"bar"
)));
CAF_CHECK_EQUAL
(
atom
(
"bar"
),
msg_roundtrip
(
atom
(
"bar"
)));
}
SERIALIZATION_TEST
(
raw_arrays
)
{
...
...
libcaf_core/test/simple_timeout.cpp
View file @
21281a6c
...
...
@@ -41,7 +41,7 @@ struct timer_state {
};
timer
::
behavior_type
timer_impl
(
timer
::
stateful_pointer
<
timer_state
>
self
)
{
self
->
delayed_send
(
self
,
ms
(
100
),
reset_atom
::
value
);
self
->
delayed_send
(
self
,
ms
(
100
),
reset_atom
_v
);
return
{
[
=
](
reset_atom
)
{
CAF_MESSAGE
(
"timer reset"
);
...
...
@@ -57,7 +57,7 @@ timer::behavior_type timer_impl(timer::stateful_pointer<timer_state> self) {
timer
::
behavior_type
timer_impl2
(
timer
::
pointer
self
)
{
auto
had_reset
=
std
::
make_shared
<
bool
>
(
false
);
delayed_anon_send
(
self
,
ms
(
100
),
reset_atom
::
value
);
delayed_anon_send
(
self
,
ms
(
100
),
reset_atom
_v
);
return
{
[
=
](
reset_atom
)
{
CAF_MESSAGE
(
"timer reset"
);
...
...
libcaf_core/test/stateful_actor.cpp
View file @
21281a6c
...
...
@@ -23,16 +23,15 @@
#include "caf/all.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
#define ERROR_HANDLER [&](error& err) { CAF_FAIL(system.render(err)); }
using
namespace
std
;
using
namespace
caf
;
namespace
{
using
typed_adder_actor
=
typed_actor
<
reacts_to
<
add_atom
,
int
>
,
replies_to
<
get_atom
>::
with
<
int
>>
;
using
typed_adder_actor
=
typed_actor
<
reacts_to
<
add_atom
,
int
>
,
replies_to
<
get_atom
>::
with
<
int
>>
;
struct
counter
{
int
value
=
0
;
...
...
@@ -40,14 +39,8 @@ struct counter {
};
behavior
adder
(
stateful_actor
<
counter
>*
self
)
{
return
{
[
=
](
add_atom
,
int
x
)
{
self
->
state
.
value
+=
x
;
},
[
=
](
get_atom
)
{
return
self
->
state
.
value
;
}
};
return
{[
=
](
add_atom
,
int
x
)
{
self
->
state
.
value
+=
x
;
},
[
=
](
get_atom
)
{
return
self
->
state
.
value
;
}};
}
class
adder_class
:
public
stateful_actor
<
counter
>
{
...
...
@@ -63,14 +56,8 @@ public:
typed_adder_actor
::
behavior_type
typed_adder
(
typed_adder_actor
::
stateful_pointer
<
counter
>
self
)
{
return
{
[
=
](
add_atom
,
int
x
)
{
self
->
state
.
value
+=
x
;
},
[
=
](
get_atom
)
{
return
self
->
state
.
value
;
}
};
return
{[
=
](
add_atom
,
int
x
)
{
self
->
state
.
value
+=
x
;
},
[
=
](
get_atom
)
{
return
self
->
state
.
value
;
}};
}
class
typed_adder_class
:
public
typed_adder_actor
::
stateful_base
<
counter
>
{
...
...
@@ -97,15 +84,11 @@ struct fixture {
template
<
class
ActorUnderTest
>
void
test_adder
(
ActorUnderTest
aut
)
{
scoped_actor
self
{
system
};
self
->
send
(
aut
,
add_atom
::
value
,
7
);
self
->
send
(
aut
,
add_atom
::
value
,
4
);
self
->
send
(
aut
,
add_atom
::
value
,
9
);
self
->
request
(
aut
,
infinite
,
get_atom
::
value
).
receive
(
[](
int
x
)
{
CAF_CHECK_EQUAL
(
x
,
20
);
},
ERROR_HANDLER
);
self
->
send
(
aut
,
add_atom_v
,
7
);
self
->
send
(
aut
,
add_atom_v
,
4
);
self
->
send
(
aut
,
add_atom_v
,
9
);
self
->
request
(
aut
,
infinite
,
get_atom_v
)
.
receive
([](
int
x
)
{
CAF_CHECK_EQUAL
(
x
,
20
);
},
ERROR_HANDLER
);
}
template
<
class
State
>
...
...
@@ -117,12 +100,9 @@ struct fixture {
};
});
scoped_actor
self
{
system
};
self
->
request
(
aut
,
infinite
,
get_atom
::
value
).
receive
(
[
&
](
const
string
&
str
)
{
CAF_CHECK_EQUAL
(
str
,
expected
);
},
ERROR_HANDLER
);
self
->
request
(
aut
,
infinite
,
get_atom_v
)
.
receive
([
&
](
const
string
&
str
)
{
CAF_CHECK_EQUAL
(
str
,
expected
);
},
ERROR_HANDLER
);
}
};
...
...
libcaf_core/test/typed_response_promise.cpp
View file @
21281a6c
...
...
@@ -29,21 +29,20 @@ using namespace caf;
namespace
{
using
foo_actor
=
typed_actor
<
replies_to
<
int
>::
with
<
int
>
,
replies_to
<
get_atom
,
int
>::
with
<
int
>
,
replies_to
<
get_atom
,
int
,
int
>::
with
<
int
,
int
>
,
replies_to
<
get_atom
,
double
>::
with
<
double
>
,
replies_to
<
get_atom
,
double
,
double
>
::
with
<
double
,
double
>
,
reacts_to
<
put_atom
,
int
,
int
>
,
reacts_to
<
put_atom
,
int
,
int
,
int
>>
;
using
foo_actor
=
typed_actor
<
replies_to
<
int
>::
with
<
int
>
,
replies_to
<
get_atom
,
int
>::
with
<
int
>
,
replies_to
<
get_atom
,
int
,
int
>::
with
<
int
,
int
>
,
replies_to
<
get_atom
,
double
>::
with
<
double
>
,
replies_to
<
get_atom
,
double
,
double
>::
with
<
double
,
double
>
,
reacts_to
<
put_atom
,
int
,
int
>
,
reacts_to
<
put_atom
,
int
,
int
,
int
>>
;
using
foo_promise
=
typed_response_promise
<
int
>
;
using
foo2_promise
=
typed_response_promise
<
int
,
int
>
;
using
foo3_promise
=
typed_response_promise
<
double
>
;
using
get1_helper
=
typed_actor
<
replies_to
<
int
,
int
>::
with
<
put_atom
,
int
,
int
>>
;
using
get2_helper
=
typed_actor
<
replies_to
<
int
,
int
,
int
>::
with
<
put_atom
,
int
,
int
,
int
>>
;
using
get2_helper
=
typed_actor
<
replies_to
<
int
,
int
,
int
>::
with
<
put_atom
,
int
,
int
,
int
>>
;
class
foo_actor_impl
:
public
foo_actor
::
base
{
public:
...
...
@@ -54,17 +53,15 @@ public:
behavior_type
make_behavior
()
override
{
return
{
[
=
](
int
x
)
->
foo_promise
{
auto
resp
=
response
(
x
*
2
);
CAF_CHECK
(
!
resp
.
pending
());
return
resp
.
deliver
(
x
*
4
);
// has no effect
auto
resp
=
response
(
x
*
2
);
CAF_CHECK
(
!
resp
.
pending
());
return
resp
.
deliver
(
x
*
4
);
// has no effect
},
[
=
](
get_atom
,
int
x
)
->
foo_promise
{
auto
calculator
=
spawn
([]()
->
get1_helper
::
behavior_type
{
return
{
[](
int
promise_id
,
int
value
)
->
result
<
put_atom
,
int
,
int
>
{
return
{
put_atom
::
value
,
promise_id
,
value
*
2
};
}
};
return
{[](
int
promise_id
,
int
value
)
->
result
<
put_atom
,
int
,
int
>
{
return
{
put_atom_v
,
promise_id
,
value
*
2
};
}};
});
send
(
calculator
,
next_id_
,
x
);
auto
&
entry
=
promises_
[
next_id_
++
];
...
...
@@ -73,11 +70,10 @@ public:
},
[
=
](
get_atom
,
int
x
,
int
y
)
->
foo2_promise
{
auto
calculator
=
spawn
([]()
->
get2_helper
::
behavior_type
{
return
{
[](
int
promise_id
,
int
v0
,
int
v1
)
->
result
<
put_atom
,
int
,
int
,
int
>
{
return
{
put_atom
::
value
,
promise_id
,
v0
*
2
,
v1
*
2
};
}
};
return
{[](
int
promise_id
,
int
v0
,
int
v1
)
->
result
<
put_atom
,
int
,
int
,
int
>
{
return
{
put_atom_v
,
promise_id
,
v0
*
2
,
v1
*
2
};
}};
});
send
(
calculator
,
next_id_
,
x
,
y
);
auto
&
entry
=
promises2_
[
next_id_
++
];
...
...
@@ -96,9 +92,7 @@ public:
auto
resp
=
make_response_promise
<
double
>
();
return
resp
.
deliver
(
make_error
(
sec
::
unexpected_message
));
},
[
=
](
get_atom
,
double
x
,
double
y
)
{
return
response
(
x
*
2
,
y
*
2
);
},
[
=
](
get_atom
,
double
x
,
double
y
)
{
return
response
(
x
*
2
,
y
*
2
);
},
[
=
](
put_atom
,
int
promise_id
,
int
x
)
{
auto
i
=
promises_
.
find
(
promise_id
);
if
(
i
==
promises_
.
end
())
...
...
@@ -112,7 +106,7 @@ public:
return
;
i
->
second
.
deliver
(
x
,
y
);
promises2_
.
erase
(
i
);
}
}
,
};
}
...
...
@@ -124,9 +118,7 @@ private:
struct
fixture
{
fixture
()
:
system
(
cfg
),
self
(
system
,
true
),
foo
(
system
.
spawn
<
foo_actor_impl
>
())
{
:
system
(
cfg
),
self
(
system
,
true
),
foo
(
system
.
spawn
<
foo_actor_impl
>
())
{
// nop
}
...
...
@@ -145,9 +137,9 @@ CAF_TEST(typed_response_promise) {
CAF_MESSAGE
(
"trigger 'invalid response promise' error"
);
resp
.
deliver
(
1
);
// delivers on an invalid promise has no effect
auto
f
=
make_function_view
(
foo
);
CAF_CHECK_EQUAL
(
f
(
get_atom
::
value
,
42
),
84
);
CAF_CHECK_EQUAL
(
f
(
get_atom
::
value
,
42
,
52
),
std
::
make_tuple
(
84
,
104
));
CAF_CHECK_EQUAL
(
f
(
get_atom
::
value
,
3.14
,
3.14
),
std
::
make_tuple
(
6.28
,
6.28
));
CAF_CHECK_EQUAL
(
f
(
get_atom
_v
,
42
),
84
);
CAF_CHECK_EQUAL
(
f
(
get_atom
_v
,
42
,
52
),
std
::
make_tuple
(
84
,
104
));
CAF_CHECK_EQUAL
(
f
(
get_atom
_v
,
3.14
,
3.14
),
std
::
make_tuple
(
6.28
,
6.28
));
}
CAF_TEST
(
typed_response_promise_chained
)
{
...
...
@@ -158,39 +150,30 @@ CAF_TEST(typed_response_promise_chained) {
// verify that only requests get an error response message
CAF_TEST
(
error_response_message
)
{
auto
f
=
make_function_view
(
foo
);
CAF_CHECK_EQUAL
(
f
(
get_atom
::
value
,
3.14
),
sec
::
unexpected_message
);
self
->
send
(
foo
,
get_atom
::
value
,
42
);
self
->
receive
(
[](
int
x
)
{
CAF_CHECK_EQUAL
(
x
,
84
);
},
[](
double
x
)
{
CAF_ERROR
(
"unexpected ordinary response message received: "
<<
x
);
}
);
self
->
send
(
foo
,
get_atom
::
value
,
3.14
);
self
->
receive
(
[
&
](
error
&
err
)
{
CAF_CHECK_EQUAL
(
err
,
sec
::
unexpected_message
);
self
->
send
(
self
,
message
{});
}
);
CAF_CHECK_EQUAL
(
f
(
get_atom_v
,
3.14
),
sec
::
unexpected_message
);
self
->
send
(
foo
,
get_atom_v
,
42
);
self
->
receive
([](
int
x
)
{
CAF_CHECK_EQUAL
(
x
,
84
);
},
[](
double
x
)
{
CAF_ERROR
(
"unexpected ordinary response message received: "
<<
x
);
});
self
->
send
(
foo
,
get_atom_v
,
3.14
);
self
->
receive
([
&
](
error
&
err
)
{
CAF_CHECK_EQUAL
(
err
,
sec
::
unexpected_message
);
self
->
send
(
self
,
message
{});
});
}
// verify that delivering to a satisfied promise has no effect
CAF_TEST
(
satisfied_promise
)
{
self
->
send
(
foo
,
1
);
self
->
send
(
foo
,
get_atom
::
value
,
3.14
,
3.14
);
self
->
send
(
foo
,
get_atom
_v
,
3.14
,
3.14
);
int
i
=
0
;
self
->
receive_for
(
i
,
2
)
(
[](
int
x
)
{
CAF_CHECK_EQUAL
(
x
,
1
*
2
);
},
[](
double
x
,
double
y
)
{
CAF_CHECK_EQUAL
(
x
,
3.14
*
2
);
CAF_CHECK_EQUAL
(
y
,
3.14
*
2
);
}
);
self
->
receive_for
(
i
,
2
)([](
int
x
)
{
CAF_CHECK_EQUAL
(
x
,
1
*
2
);
},
[](
double
x
,
double
y
)
{
CAF_CHECK_EQUAL
(
x
,
3.14
*
2
);
CAF_CHECK_EQUAL
(
y
,
3.14
*
2
);
});
}
CAF_TEST
(
delegating_promises
)
{
...
...
@@ -199,13 +182,13 @@ CAF_TEST(delegating_promises) {
std
::
vector
<
task
>
tasks
;
};
using
bar_actor
=
typed_actor
<
replies_to
<
int
>::
with
<
int
>
,
reacts_to
<
ok_atom
>>
;
auto
bar_fun
=
[](
bar_actor
::
stateful_pointer
<
state
>
self
,
foo_actor
worker
)
->
bar_actor
::
behavior_type
{
auto
bar_fun
=
[](
bar_actor
::
stateful_pointer
<
state
>
self
,
foo_actor
worker
)
->
bar_actor
::
behavior_type
{
return
{
[
=
](
int
x
)
->
typed_response_promise
<
int
>
{
auto
&
tasks
=
self
->
state
.
tasks
;
tasks
.
emplace_back
(
self
->
make_response_promise
<
int
>
(),
x
);
self
->
send
(
self
,
ok_atom
::
value
);
self
->
send
(
self
,
ok_atom
_v
);
return
tasks
.
back
().
first
;
},
[
=
](
ok_atom
)
{
...
...
@@ -215,7 +198,7 @@ CAF_TEST(delegating_promises) {
task
.
first
.
delegate
(
worker
,
task
.
second
);
tasks
.
pop_back
();
}
}
}
,
};
};
auto
f
=
make_function_view
(
system
.
spawn
(
bar_fun
,
foo
));
...
...
libcaf_core/test/typed_spawn.cpp
View file @
21281a6c
...
...
@@ -23,7 +23,7 @@
# define CAF_SUITE typed_spawn
# include "c
af/test/unit_
test.hpp"
# include "c
ore-
test.hpp"
# include "caf/string_algorithms.hpp"
...
...
@@ -33,13 +33,10 @@
namespace
{
struct
get_state_msg
;
struct
my_request
;
using
server_type
=
caf
::
typed_actor
<
caf
::
replies_to
<
my_request
>::
with
<
bool
>>
;
using
event_testee_type
=
caf
::
typed_actor
<
caf
::
replies_to
<
get_state_
msg
>::
with
<
std
::
string
>
,
=
caf
::
typed_actor
<
caf
::
replies_to
<
get_state_
atom
>::
with
<
std
::
string
>
,
caf
::
replies_to
<
std
::
string
>::
with
<
void
>
,
caf
::
replies_to
<
float
>::
with
<
void
>
,
caf
::
replies_to
<
int32_t
>::
with
<
int32_t
>>
;
...
...
@@ -53,23 +50,14 @@ using float_actor = caf::typed_actor<caf::reacts_to<float>>;
}
// namespace
CAF_BEGIN_TYPE_ID_BLOCK
(
typed_spawn
,
first_custom_type_id
)
CAF_ADD_TYPE_ID
(
typed_spawn
,
(
event_testee_type
))
CAF_ADD_TYPE_ID
(
typed_spawn
,
(
float_actor
))
CAF_ADD_TYPE_ID
(
typed_spawn
,
(
get_state_msg
))
CAF_ADD_TYPE_ID
(
typed_spawn
,
(
int_actor
))
CAF_ADD_TYPE_ID
(
typed_spawn
,
(
my_request
))
CAF_ADD_TYPE_ID
(
typed_spawn
,
(
string_actor
))
CAF_END_TYPE_ID_BLOCK
(
typed_spawn
)
using
std
::
string
;
using
namespace
caf
;
using
passed_atom
=
caf
::
atom_constant
<
caf
::
atom
(
"passed"
)
>
;
constexpr
auto
passed_atom_v
=
passed_atom
::
value
;
namespace
{
enum
class
mock_errc
:
uint8_t
{
...
...
@@ -103,16 +91,6 @@ static_assert(std::is_convertible<dummy5, dummy4>::value,
* simple request/response test *
******************************************************************************/
struct
my_request
{
int32_t
a
;
int32_t
b
;
};
template
<
class
Inspector
>
typename
Inspector
::
result_type
inspect
(
Inspector
&
f
,
my_request
&
x
)
{
return
f
(
x
.
a
,
x
.
b
);
}
server_type
::
behavior_type
typed_server1
()
{
return
{
[](
const
my_request
&
req
)
{
return
req
.
a
==
req
.
b
;
},
...
...
@@ -141,7 +119,7 @@ void client(event_based_actor* self, const actor& parent,
CAF_CHECK_EQUAL
(
val1
,
true
);
self
->
request
(
serv
,
infinite
,
my_request
{
10
,
20
}).
then
([
=
](
bool
val2
)
{
CAF_CHECK_EQUAL
(
val2
,
false
);
self
->
send
(
parent
,
passed_atom
::
value
);
self
->
send
(
parent
,
passed_atom
_v
);
});
});
}
...
...
@@ -150,8 +128,6 @@ void client(event_based_actor* self, const actor& parent,
* test skipping of messages intentionally + using become() *
******************************************************************************/
struct
get_state_msg
{};
class
event_testee
:
public
event_testee_type
::
base
{
public:
event_testee
(
actor_config
&
cfg
)
:
event_testee_type
::
base
(
cfg
)
{
...
...
@@ -160,7 +136,7 @@ public:
behavior_type
wait4string
()
{
return
{
[
=
](
const
get_state_msg
&
)
{
return
"wait4string"
;
},
[
=
](
get_state_atom
)
{
return
"wait4string"
;
},
[
=
](
const
string
&
)
{
become
(
wait4int
());
},
[
=
](
float
)
{
return
skip
();
},
[
=
](
int
)
{
return
skip
();
},
...
...
@@ -169,7 +145,7 @@ public:
behavior_type
wait4int
()
{
return
{
[
=
](
const
get_state_msg
&
)
{
return
"wait4int"
;
},
[
=
](
get_state_atom
)
{
return
"wait4int"
;
},
[
=
](
int
)
->
int
{
become
(
wait4float
());
return
42
;
...
...
@@ -181,7 +157,7 @@ public:
behavior_type
wait4float
()
{
return
{
[
=
](
const
get_state_msg
&
)
{
return
"wait4float"
;
},
[
=
](
get_state_atom
)
{
return
"wait4float"
;
},
[
=
](
float
)
{
become
(
wait4string
());
},
[
=
](
const
string
&
)
{
return
skip
();
},
[
=
](
int
)
{
return
skip
();
},
...
...
@@ -307,7 +283,7 @@ struct fixture {
scoped_actor
self
;
static
actor_system_config
&
init
(
actor_system_config
&
cfg
)
{
cfg
.
add_message_types
<
id_block
::
typed_spawn
>
();
cfg
.
add_message_types
<
id_block
::
core_test
>
();
cfg
.
parse
(
test
::
engine
::
argc
(),
test
::
engine
::
argv
());
return
cfg
;
}
...
...
@@ -370,13 +346,13 @@ CAF_TEST(event_testee_series) {
self
->
send
(
et
,
.3
f
);
self
->
send
(
et
,
"hello again event testee!"
);
self
->
send
(
et
,
"goodbye event testee!"
);
typed_actor
<
replies_to
<
get_state_
msg
>::
with
<
string
>>
sub_et
=
et
;
std
::
set
<
string
>
iface
{
"caf::replies_to<get_state_
msg
>::with<@str>"
,
typed_actor
<
replies_to
<
get_state_
atom
>::
with
<
string
>>
sub_et
=
et
;
std
::
set
<
string
>
iface
{
"caf::replies_to<get_state_
atom
>::with<@str>"
,
"caf::replies_to<@str>::with<void>"
,
"caf::replies_to<float>::with<void>"
,
"caf::replies_to<@i32>::with<@i32>"
};
CAF_CHECK_EQUAL
(
join
(
sub_et
->
message_types
(),
","
),
join
(
iface
,
","
));
self
->
send
(
sub_et
,
get_state_
msg
{}
);
self
->
send
(
sub_et
,
get_state_
atom_v
);
// we expect three 42s
int
i
=
0
;
self
->
receive_for
(
i
,
3
)([](
int
value
)
{
CAF_CHECK_EQUAL
(
value
,
42
);
});
...
...
libcaf_core/test/unit.cpp
View file @
21281a6c
...
...
@@ -17,34 +17,29 @@
******************************************************************************/
#define CAF_SUITE unit
#include "caf/test/unit_test.hpp"
#include "caf/unit.hpp"
#include "core-test.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/atom.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/unit.hpp"
using
namespace
caf
;
using
unit_res_atom
=
atom_constant
<
atom
(
"unitRes"
)
>
;
using
void_res_atom
=
atom_constant
<
atom
(
"voidRes"
)
>
;
using
unit_raw_atom
=
atom_constant
<
atom
(
"unitRaw"
)
>
;
using
void_raw_atom
=
atom_constant
<
atom
(
"voidRaw"
)
>
;
using
typed_unit_atom
=
atom_constant
<
atom
(
"typedUnit"
)
>
;
behavior
testee
(
event_based_actor
*
self
)
{
return
{
[]
(
unit_res_atom
)
->
result
<
unit_t
>
{
return
unit
;
},
[]
(
void_res_atom
)
->
result
<
void
>
{
return
{};
},
[]
(
unit_raw_atom
)
->
unit_t
{
return
unit
;
},
[]
(
void_raw_atom
)
->
void
{
},
[
=
](
typed_unit
_atom
)
->
result
<
unit_t
>
{
[]
(
add_atom
)
->
result
<
unit_t
>
{
return
unit
;
},
[]
(
get_atom
)
->
result
<
void
>
{
return
{};
},
[]
(
put_atom
)
->
unit_t
{
return
unit
;
},
[]
(
resolve_atom
)
->
void
{
},
[
=
](
update
_atom
)
->
result
<
unit_t
>
{
auto
rp
=
self
->
make_response_promise
<
unit_t
>
();
rp
.
deliver
(
unit
);
return
rp
;
}
}
,
};
}
...
...
@@ -53,17 +48,19 @@ CAF_TEST(unit_results) {
actor_system
sys
{
cfg
};
scoped_actor
self
{
sys
};
auto
aut
=
sys
.
spawn
(
testee
);
atom_value
as
[]
=
{
unit_res_atom
::
value
,
void_res_atom
::
value
,
unit_raw_atom
::
value
,
void_raw_atom
::
value
,
typed_unit_atom
::
value
};
message
as
[]
=
{
make_message
(
add_atom_v
),
make_message
(
get_atom_v
),
make_message
(
put_atom_v
),
make_message
(
resolve_atom_v
),
make_message
(
update_atom_v
),
};
for
(
auto
a
:
as
)
{
self
->
request
(
aut
,
infinite
,
a
)
.
receive
(
[
&
]
{
CAF_MESSAGE
(
"actor under test correctly replied to "
<<
to_string
(
a
));
},
[
&
]
(
const
error
&
)
{
CAF_FAIL
(
"actor under test failed at input "
<<
to_string
(
a
));
}
);
self
->
request
(
aut
,
infinite
,
a
)
.
receive
(
[
&
]
{
CAF_MESSAGE
(
"actor under test correctly replied to "
<<
to_string
(
a
));
},
[
&
](
const
error
&
)
{
CAF_FAIL
(
"actor under test failed at input "
<<
to_string
(
a
));
}
);
}
}
libcaf_io/src/io/basp/instance.cpp
View file @
21281a6c
...
...
@@ -462,7 +462,7 @@ connection_state instance::handle(execution_unit* ctx, connection_handle hdl,
// Delay this message to make sure we don't skip in-flight messages.
auto
msg_id
=
queue_
.
new_id
();
auto
ptr
=
make_mailbox_element
(
nullptr
,
make_message_id
(),
{},
delete_atom
::
value
,
source_node
,
delete_atom
_v
,
source_node
,
hdr
.
source_actor
,
std
::
move
(
fail_state
));
queue_
.
push
(
callee_
.
current_execution_unit
(),
msg_id
,
...
...
libcaf_io/src/io/basp_broker.cpp
View file @
21281a6c
...
...
@@ -111,7 +111,7 @@ behavior basp_broker::make_behavior() {
auto
port
=
res
->
second
;
auto
addrs
=
network
::
interfaces
::
list_addresses
(
false
);
auto
config_server
=
system
().
registry
().
get
(
atom
(
"ConfigServ"
));
send
(
actor_cast
<
actor
>
(
config_server
),
put_atom
::
value
,
send
(
actor_cast
<
actor
>
(
config_server
),
put_atom
_v
,
"basp.default-connectivity-tcp"
,
make_message
(
port
,
std
::
move
(
addrs
)));
}
...
...
@@ -125,7 +125,7 @@ behavior basp_broker::make_behavior() {
auto
first_tick
=
now
+
heartbeat_interval
;
auto
connection_timeout
=
get_or
(
config
(),
"middleman.connection-timeout"
,
defaults
::
middleman
::
connection_timeout
);
scheduled_send
(
this
,
first_tick
,
tick_atom
::
value
,
first_tick
,
scheduled_send
(
this
,
first_tick
,
tick_atom
_v
,
first_tick
,
heartbeat_interval
,
connection_timeout
);
}
return
behavior
{
...
...
@@ -230,8 +230,8 @@ behavior basp_broker::make_behavior() {
auto
&
q
=
instance
.
queue
();
auto
msg_id
=
q
.
new_id
();
q
.
push
(
context
(),
msg_id
,
ctrl
(),
make_mailbox_element
(
nullptr
,
make_message_id
(),
{},
delete_atom
::
value
,
msg
.
handle
));
make_mailbox_element
(
nullptr
,
make_message_id
(),
{},
delete_atom_v
,
msg
.
handle
));
},
// received from the message handler above for connection_closed_msg
[
=
](
delete_atom
,
connection_handle
hdl
)
{
...
...
@@ -244,8 +244,8 @@ behavior basp_broker::make_behavior() {
auto
&
q
=
instance
.
queue
();
auto
msg_id
=
q
.
new_id
();
q
.
push
(
context
(),
msg_id
,
ctrl
(),
make_mailbox_element
(
nullptr
,
make_message_id
(),
{},
delete_atom
::
value
,
msg
.
handle
));
make_mailbox_element
(
nullptr
,
make_message_id
(),
{},
delete_atom_v
,
msg
.
handle
));
},
// received from the message handler above for acceptor_closed_msg
[
=
](
delete_atom
,
accept_handle
hdl
)
{
...
...
@@ -342,7 +342,7 @@ behavior basp_broker::make_behavior() {
auto
now
=
clock
().
now
();
if
(
now
<
scheduled
)
{
CAF_LOG_WARNING
(
"received tick before its time, reschedule"
);
scheduled_send
(
this
,
scheduled
,
tick_atom
::
value
,
scheduled
,
scheduled_send
(
this
,
scheduled
,
tick_atom
_v
,
scheduled
,
heartbeat_interval
,
connection_timeout
);
return
;
}
...
...
@@ -377,7 +377,7 @@ behavior basp_broker::make_behavior() {
}
}
// Schedule next tick.
scheduled_send
(
this
,
next_tick
,
tick_atom
::
value
,
next_tick
,
scheduled_send
(
this
,
next_tick
,
tick_atom
_v
,
next_tick
,
heartbeat_interval
,
connection_timeout
);
}};
}
...
...
@@ -535,8 +535,8 @@ void basp_broker::learned_new_node(const node_id& nid) {
tself
->
become
([
=
](
spawn_atom
,
std
::
string
&
type
,
message
&
args
)
->
delegated
<
strong_actor_ptr
,
std
::
set
<
std
::
string
>>
{
CAF_LOG_TRACE
(
CAF_ARG
(
type
)
<<
CAF_ARG
(
args
));
tself
->
delegate
(
actor_cast
<
actor
>
(
std
::
move
(
config_serv
)),
get_atom
::
value
,
std
::
move
(
type
),
std
::
move
(
args
));
tself
->
delegate
(
actor_cast
<
actor
>
(
std
::
move
(
config_serv
)),
get_atom_v
,
std
::
move
(
type
),
std
::
move
(
args
));
return
{};
});
},
...
...
@@ -554,8 +554,7 @@ void basp_broker::learned_new_node(const node_id& nid) {
if
(
!
instance
.
dispatch
(
context
(),
tmp_ptr
,
stages
,
nid
,
static_cast
<
uint64_t
>
(
atom
(
"SpawnServ"
)),
basp
::
header
::
named_receiver_flag
,
make_message_id
(),
make_message
(
sys_atom
::
value
,
get_atom
::
value
,
"info"
)))
{
make_message
(
sys_atom_v
,
get_atom_v
,
"info"
)))
{
CAF_LOG_ERROR
(
"learned_new_node called, but no route to remote node"
<<
CAF_ARG
(
nid
));
}
...
...
@@ -586,7 +585,7 @@ void basp_broker::learned_new_node_indirectly(const node_id& nid) {
if
(
!
instance
.
dispatch
(
context
(),
sender
,
fwd_stack
,
nid
,
static_cast
<
uint64_t
>
(
atom
(
"ConfigServ"
)),
basp
::
header
::
named_receiver_flag
,
make_message_id
(),
make_message
(
get_atom
::
value
,
make_message
(
get_atom
_v
,
"basp.default-connectivity-tcp"
)))
{
CAF_LOG_ERROR
(
"learned_new_node_indirectly called, but no route to nid"
);
}
...
...
libcaf_io/src/io/connection_helper.cpp
View file @
21281a6c
...
...
@@ -59,7 +59,7 @@ behavior connection_helper(stateful_actor<connection_helper_state>* self,
// gotcha! send scribe to our BASP broker
// to initiate handshake etc.
CAF_LOG_INFO
(
"connected directly:"
<<
CAF_ARG
(
addr
));
self
->
send
(
b
,
connect_atom
::
value
,
*
hdl
,
port
);
self
->
send
(
b
,
connect_atom
_v
,
*
hdl
,
port
);
return
;
}
}
...
...
libcaf_io/src/io/middleman.cpp
View file @
21281a6c
...
...
@@ -100,14 +100,12 @@ middleman::middleman(actor_system& sys) : system_(sys) {
// nop
}
expected
<
strong_actor_ptr
>
middleman
::
remote_spawn_impl
(
const
node_id
&
nid
,
std
::
string
&
name
,
message
&
args
,
std
::
set
<
std
::
string
>
s
,
duration
timeout
)
{
expected
<
strong_actor_ptr
>
middleman
::
remote_spawn_impl
(
const
node_id
&
nid
,
std
::
string
&
name
,
message
&
args
,
std
::
set
<
std
::
string
>
s
,
duration
timeout
)
{
auto
f
=
make_function_view
(
actor_handle
(),
timeout
);
return
f
(
spawn_atom
::
value
,
nid
,
std
::
move
(
name
),
std
::
move
(
args
),
std
::
move
(
s
));
return
f
(
spawn_atom_v
,
nid
,
std
::
move
(
name
),
std
::
move
(
args
),
std
::
move
(
s
));
}
expected
<
uint16_t
>
middleman
::
open
(
uint16_t
port
,
const
char
*
in
,
bool
reuse
)
{
...
...
@@ -115,17 +113,17 @@ expected<uint16_t> middleman::open(uint16_t port, const char* in, bool reuse) {
if
(
in
!=
nullptr
)
str
=
in
;
auto
f
=
make_function_view
(
actor_handle
());
return
f
(
open_atom
::
value
,
port
,
std
::
move
(
str
),
reuse
);
return
f
(
open_atom
_v
,
port
,
std
::
move
(
str
),
reuse
);
}
expected
<
void
>
middleman
::
close
(
uint16_t
port
)
{
auto
f
=
make_function_view
(
actor_handle
());
return
f
(
close_atom
::
value
,
port
);
return
f
(
close_atom
_v
,
port
);
}
expected
<
node_id
>
middleman
::
connect
(
std
::
string
host
,
uint16_t
port
)
{
auto
f
=
make_function_view
(
actor_handle
());
auto
res
=
f
(
connect_atom
::
value
,
std
::
move
(
host
),
port
);
auto
res
=
f
(
connect_atom
_v
,
std
::
move
(
host
),
port
);
if
(
!
res
)
return
std
::
move
(
res
.
error
());
return
std
::
get
<
0
>
(
*
res
);
...
...
@@ -141,7 +139,7 @@ expected<uint16_t> middleman::publish(const strong_actor_ptr& whom,
if
(
cstr
!=
nullptr
)
in
=
cstr
;
auto
f
=
make_function_view
(
actor_handle
());
return
f
(
publish_atom
::
value
,
port
,
std
::
move
(
whom
),
std
::
move
(
sigs
),
in
,
ru
);
return
f
(
publish_atom
_v
,
port
,
std
::
move
(
whom
),
std
::
move
(
sigs
),
in
,
ru
);
}
expected
<
uint16_t
>
middleman
::
publish_local_groups
(
uint16_t
port
,
...
...
@@ -167,7 +165,7 @@ expected<uint16_t> middleman::publish_local_groups(uint16_t port,
expected
<
void
>
middleman
::
unpublish
(
const
actor_addr
&
whom
,
uint16_t
port
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
whom
)
<<
CAF_ARG
(
port
));
auto
f
=
make_function_view
(
actor_handle
());
return
f
(
unpublish_atom
::
value
,
whom
,
port
);
return
f
(
unpublish_atom
_v
,
whom
,
port
);
}
expected
<
strong_actor_ptr
>
middleman
::
remote_actor
(
std
::
set
<
std
::
string
>
ifs
,
...
...
@@ -175,7 +173,7 @@ expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs,
uint16_t
port
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
ifs
)
<<
CAF_ARG
(
host
)
<<
CAF_ARG
(
port
));
auto
f
=
make_function_view
(
actor_handle
());
auto
res
=
f
(
connect_atom
::
value
,
std
::
move
(
host
),
port
);
auto
res
=
f
(
connect_atom
_v
,
std
::
move
(
host
),
port
);
if
(
!
res
)
return
std
::
move
(
res
.
error
());
strong_actor_ptr
ptr
=
std
::
move
(
std
::
get
<
1
>
(
*
res
));
...
...
@@ -217,11 +215,11 @@ expected<group> middleman::remote_group(const std::string& group_identifier,
/// terminate the actor after both requests finish.
self
->
unbecome
();
auto
rp
=
self
->
make_response_promise
();
self
->
request
(
mm
,
infinite
,
connect_atom
::
value
,
host
,
port
)
self
->
request
(
mm
,
infinite
,
connect_atom
_v
,
host
,
port
)
.
then
([
=
](
const
node_id
&
,
strong_actor_ptr
&
ptr
,
const
std
::
set
<
std
::
string
>&
)
mutable
{
auto
hdl
=
actor_cast
<
actor
>
(
ptr
);
self
->
request
(
hdl
,
infinite
,
get_atom
::
value
,
group_identifier
)
self
->
request
(
hdl
,
infinite
,
get_atom
_v
,
group_identifier
)
.
then
([
=
](
group
&
grp
)
mutable
{
rp
.
deliver
(
std
::
move
(
grp
));
});
});
},
...
...
@@ -232,7 +230,7 @@ expected<group> middleman::remote_group(const std::string& group_identifier,
scoped_actor
self
{
system
(),
true
};
auto
worker
=
self
->
spawn
<
lazy_init
+
monitored
>
(
two_step_lookup
,
actor_handle
());
self
->
send
(
worker
,
get_atom
::
value
);
self
->
send
(
worker
,
get_atom
_v
);
self
->
receive
([
&
](
group
&
grp
)
{
result
=
std
::
move
(
grp
);
},
[
&
](
error
&
err
)
{
result
=
std
::
move
(
err
);
},
[
&
](
down_msg
&
dm
)
{
result
=
std
::
move
(
dm
.
reason
);
});
...
...
@@ -246,8 +244,8 @@ strong_actor_ptr middleman::remote_lookup(atom_value name, const node_id& nid) {
auto
basp
=
named_broker
<
basp_broker
>
(
atom
(
"BASP"
));
strong_actor_ptr
result
;
scoped_actor
self
{
system
(),
true
};
self
->
send
(
basp
,
forward_atom
::
value
,
nid
,
atom
(
"ConfigServ"
),
make_message
(
get_atom
::
value
,
name
));
self
->
send
(
basp
,
forward_atom
_v
,
nid
,
atom
(
"ConfigServ"
),
make_message
(
get_atom
_v
,
name
));
self
->
receive
([
&
](
strong_actor_ptr
&
addr
)
{
result
=
std
::
move
(
addr
);
},
after
(
std
::
chrono
::
minutes
(
5
))
>>
[]
{
...
...
libcaf_io/src/io/middleman_actor_impl.cpp
View file @
21281a6c
...
...
@@ -114,7 +114,7 @@ auto middleman_actor_impl::make_behavior() -> behavior_type {
auto
&
ptr
=
*
r
;
std
::
vector
<
response_promise
>
tmp
{
std
::
move
(
rp
)};
pending_
.
emplace
(
key
,
std
::
move
(
tmp
));
request
(
broker_
,
infinite
,
connect_atom
::
value
,
std
::
move
(
ptr
),
port
)
request
(
broker_
,
infinite
,
connect_atom
_v
,
std
::
move
(
ptr
),
port
)
.
then
(
[
=
](
node_id
&
nid
,
strong_actor_ptr
&
addr
,
mpi_set
&
sigs
)
{
auto
i
=
pending_
.
find
(
key
);
...
...
@@ -153,7 +153,7 @@ auto middleman_actor_impl::make_behavior() -> behavior_type {
[
=
](
spawn_atom
atm
,
node_id
&
nid
,
std
::
string
&
str
,
message
&
msg
,
std
::
set
<
std
::
string
>&
ifs
)
->
delegated
<
strong_actor_ptr
>
{
CAF_LOG_TRACE
(
""
);
delegate
(
broker_
,
forward_atom
::
value
,
nid
,
atom
(
"SpawnServ"
),
delegate
(
broker_
,
forward_atom
_v
,
nid
,
atom
(
"SpawnServ"
),
make_message
(
atm
,
std
::
move
(
str
),
std
::
move
(
msg
),
std
::
move
(
ifs
)));
return
{};
...
...
@@ -181,7 +181,7 @@ middleman_actor_impl::put(uint16_t port, strong_actor_ptr& whom, mpi_set& sigs,
return
std
::
move
(
res
.
error
());
auto
&
ptr
=
*
res
;
actual_port
=
ptr
->
port
();
anon_send
(
broker_
,
publish_atom
::
value
,
std
::
move
(
ptr
),
actual_port
,
anon_send
(
broker_
,
publish_atom
_v
,
std
::
move
(
ptr
),
actual_port
,
std
::
move
(
whom
),
std
::
move
(
sigs
));
return
actual_port
;
}
...
...
@@ -200,7 +200,7 @@ middleman_actor_impl::put_udp(uint16_t port, strong_actor_ptr& whom,
return
std
::
move
(
res
.
error
());
auto
&
ptr
=
*
res
;
actual_port
=
ptr
->
local_port
();
anon_send
(
broker_
,
publish_udp_atom
::
value
,
std
::
move
(
ptr
),
actual_port
,
anon_send
(
broker_
,
publish_udp_atom
_v
,
std
::
move
(
ptr
),
actual_port
,
std
::
move
(
whom
),
std
::
move
(
sigs
));
return
actual_port
;
}
...
...
libcaf_io/test/io/basp/message_queue.cpp
View file @
21281a6c
...
...
@@ -53,7 +53,7 @@ struct fixture : test_coordinator_fixture<> {
void
push
(
int
msg_id
)
{
queue
.
push
(
nullptr
,
static_cast
<
uint64_t
>
(
msg_id
),
testee
,
make_mailbox_element
(
self
->
ctrl
(),
make_message_id
(),
{},
ok_atom
::
value
,
msg_id
));
ok_atom
_v
,
msg_id
));
}
};
...
...
libcaf_io/test/io/basp_broker.cpp
View file @
21281a6c
...
...
@@ -286,7 +286,7 @@ public:
default_operation_data
,
any_vals
,
static_cast
<
uint64_t
>
(
spawn_serv_atom
),
std
::
vector
<
strong_actor_ptr
>
{},
make_message
(
sys_atom
::
value
,
get_atom
::
value
,
"info"
));
make_message
(
sys_atom
_v
,
get_atom_v
,
"info"
));
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
auto
path
=
tbl
().
lookup
(
n
.
id
);
...
...
@@ -439,8 +439,8 @@ public:
using
sig_t
=
std
::
set
<
std
::
string
>
;
scoped_actor
tmp
{
sys
};
sig_t
sigs
;
tmp
->
send
(
mma
,
publish_atom
::
value
,
port
,
actor_cast
<
strong_actor_ptr
>
(
whom
),
std
::
move
(
sigs
),
""
,
false
);
tmp
->
send
(
mma
,
publish_atom
_v
,
port
,
actor_cast
<
strong_actor_ptr
>
(
whom
)
,
std
::
move
(
sigs
),
""
,
false
);
expect
((
atom_value
,
uint16_t
,
strong_actor_ptr
,
sig_t
,
std
::
string
,
bool
),
from
(
tmp
).
to
(
mma
));
expect
((
uint16_t
),
from
(
mma
).
to
(
tmp
).
with
(
port
));
...
...
@@ -501,7 +501,7 @@ CAF_TEST(remote_address_and_port) {
connect_node
(
mars
());
auto
mm
=
sys
.
middleman
().
actor_handle
();
CAF_MESSAGE
(
"ask MM about node ID of Mars"
);
self
()
->
send
(
mm
,
get_atom
::
value
,
mars
().
id
);
self
()
->
send
(
mm
,
get_atom
_v
,
mars
().
id
);
do
{
mpx
()
->
exec_runnable
();
}
while
(
self
()
->
mailbox
().
empty
());
...
...
@@ -575,8 +575,7 @@ CAF_TEST(remote_actor_and_send) {
CAF_REQUIRE
(
mpx
()
->
has_pending_scribe
(
lo
,
4242
));
auto
mm1
=
sys
.
middleman
().
actor_handle
();
actor
result
;
auto
f
=
self
()
->
request
(
mm1
,
infinite
,
connect_atom
::
value
,
lo
,
uint16_t
{
4242
});
auto
f
=
self
()
->
request
(
mm1
,
infinite
,
connect_atom_v
,
lo
,
uint16_t
{
4242
});
// wait until BASP broker has received and processed the connect message
while
(
!
aut
()
->
valid
(
jupiter
().
connection
))
mpx
()
->
exec_runnable
();
...
...
@@ -597,7 +596,7 @@ CAF_TEST(remote_actor_and_send) {
default_operation_data
,
any_vals
,
static_cast
<
uint64_t
>
(
spawn_serv_atom
),
std
::
vector
<
strong_actor_ptr
>
{},
make_message
(
sys_atom
::
value
,
get_atom
::
value
,
"info"
))
make_message
(
sys_atom
_v
,
get_atom_v
,
"info"
))
.
receive
(
jupiter
().
connection
,
basp
::
message_type
::
monitor_message
,
no_flags
,
any_vals
,
no_operation_data
,
invalid_actor_id
,
jupiter
().
dummy_actor
->
id
(),
this_node
(),
jupiter
().
id
);
...
...
@@ -694,7 +693,7 @@ CAF_TEST(indirect_connections) {
default_operation_data
,
any_vals
,
static_cast
<
uint64_t
>
(
spawn_serv_atom
),
this_node
(),
jupiter
().
id
,
std
::
vector
<
strong_actor_ptr
>
{},
make_message
(
sys_atom
::
value
,
get_atom
::
value
,
"info"
));
make_message
(
sys_atom
_v
,
get_atom_v
,
"info"
));
CAF_MESSAGE
(
"expect announce_proxy message at Mars from Earth to Jupiter"
);
mx
.
receive
(
mars
().
connection
,
basp
::
message_type
::
monitor_message
,
no_flags
,
any_vals
,
no_operation_data
,
invalid_actor_id
,
...
...
@@ -746,14 +745,14 @@ CAF_TEST(automatic_connection) {
default_operation_data
,
any_vals
,
static_cast
<
uint64_t
>
(
spawn_serv_atom
),
this_node
(),
jupiter
().
id
,
std
::
vector
<
strong_actor_ptr
>
{},
make_message
(
sys_atom
::
value
,
get_atom
::
value
,
"info"
))
make_message
(
sys_atom
_v
,
get_atom_v
,
"info"
))
.
receive
(
mars
().
connection
,
basp
::
message_type
::
routed_message
,
basp
::
header
::
named_receiver_flag
,
any_vals
,
default_operation_data
,
any_vals
,
// actor ID of an actor spawned by the BASP broker
static_cast
<
uint64_t
>
(
config_serv_atom
),
this_node
(),
jupiter
().
id
,
std
::
vector
<
strong_actor_ptr
>
{},
make_message
(
get_atom
::
value
,
"basp.default-connectivity-tcp"
))
make_message
(
get_atom
_v
,
"basp.default-connectivity-tcp"
))
.
receive
(
mars
().
connection
,
basp
::
message_type
::
monitor_message
,
no_flags
,
any_vals
,
no_operation_data
,
invalid_actor_id
,
jupiter
().
dummy_actor
->
id
(),
this_node
(),
jupiter
().
id
);
...
...
libcaf_io/test/io/broker.cpp
View file @
21281a6c
...
...
@@ -22,6 +22,7 @@
#include "caf/test/io_dsl.hpp"
#include <cstdint>
#include <iostream>
#include <memory>
...
...
@@ -33,11 +34,6 @@ using namespace caf::io;
namespace
{
using
ping_atom
=
caf
::
atom_constant
<
caf
::
atom
(
"ping"
)
>
;
using
pong_atom
=
caf
::
atom_constant
<
caf
::
atom
(
"pong"
)
>
;
using
publish_atom
=
caf
::
atom_constant
<
caf
::
atom
(
"publish"
)
>
;
using
kickoff_atom
=
caf
::
atom_constant
<
caf
::
atom
(
"kickoff"
)
>
;
struct
suite_state
{
int
pings
=
0
;
int
pongs
=
0
;
...
...
@@ -48,17 +44,20 @@ using suite_state_ptr = std::shared_ptr<suite_state>;
behavior
ping
(
event_based_actor
*
self
,
suite_state_ptr
ssp
)
{
return
{
[
=
](
kickoff
_atom
,
const
actor
&
pong
)
{
CAF_MESSAGE
(
"received `
kickoff
_atom`"
);
[
=
](
ok
_atom
,
const
actor
&
pong
)
{
CAF_MESSAGE
(
"received `
ok
_atom`"
);
++
ssp
->
pings
;
self
->
send
(
pong
,
ping_atom
::
value
);
self
->
become
([
=
](
pong_atom
)
{
CAF_MESSAGE
(
"ping: received pong"
);
self
->
send
(
pong
,
ping_atom
::
value
);
if
(
++
ssp
->
pings
==
10
)
{
self
->
quit
();
CAF_MESSAGE
(
"ping is done"
);
}
self
->
send
(
pong
,
ping_atom_v
);
self
->
become
({
[
=
](
pong_atom
)
{
CAF_MESSAGE
(
"ping: received pong"
);
self
->
send
(
pong
,
ping_atom_v
);
if
(
++
ssp
->
pings
==
10
)
{
self
->
quit
();
CAF_MESSAGE
(
"ping is done"
);
}
},
[
=
](
ping_atom
)
{
CAF_FAIL
(
"ping received a ping message"
);
},
});
},
};
...
...
@@ -66,13 +65,13 @@ behavior ping(event_based_actor* self, suite_state_ptr ssp) {
behavior
pong
(
event_based_actor
*
self
,
suite_state_ptr
ssp
)
{
return
{
[
=
](
ping_atom
)
->
atom_value
{
[
=
](
ping_atom
)
->
pong_atom
{
CAF_MESSAGE
(
"pong: received ping"
);
if
(
++
ssp
->
pongs
==
10
)
{
self
->
quit
();
CAF_MESSAGE
(
"pong is done"
);
}
return
pong_atom
::
value
;
return
pong_atom
_v
;
},
};
}
...
...
@@ -88,10 +87,10 @@ behavior peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
});
// Assume exactly one connection.
CAF_REQUIRE
(
self
->
connections
().
size
()
==
1
);
self
->
configure_read
(
hdl
,
receive_policy
::
exactly
(
sizeof
(
atom_value
)));
auto
write
=
[
=
](
atom_value
typ
e
)
{
self
->
configure_read
(
hdl
,
receive_policy
::
exactly
(
sizeof
(
type_id_t
)));
auto
write
=
[
=
](
atom_value
valu
e
)
{
auto
&
buf
=
self
->
wr_buf
(
hdl
);
auto
first
=
reinterpret_cast
<
char
*>
(
&
typ
e
);
auto
first
=
reinterpret_cast
<
char
*>
(
&
valu
e
);
buf
.
insert
(
buf
.
end
(),
first
,
first
+
sizeof
(
atom_value
));
self
->
flush
(
hdl
);
};
...
...
@@ -102,12 +101,13 @@ behavior peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
},
[
=
](
const
new_data_msg
&
msg
)
{
CAF_MESSAGE
(
"received new_data_msg"
);
atom_value
type
;
memcpy
(
&
type
,
msg
.
buf
.
data
(),
sizeof
(
atom_value
));
self
->
send
(
buddy
,
type
);
CAF_REQUIRE_EQUAL
(
msg
.
buf
.
size
(),
sizeof
(
atom_value
));
atom_value
value
;
memcpy
(
&
value
,
msg
.
buf
.
data
(),
sizeof
(
atom_value
));
self
->
send
(
buddy
,
value
);
},
[
=
](
ping_atom
)
{
write
(
ping_atom
::
value
);
},
[
=
](
pong_atom
)
{
write
(
pong_atom
::
value
);
},
[
=
](
ping_atom
)
{
write
(
ping_atom
_v
);
},
[
=
](
pong_atom
)
{
write
(
pong_atom
_v
);
},
};
}
...
...
@@ -119,11 +119,11 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) {
self
->
fork
(
peer_fun
,
msg
.
handle
,
buddy
);
self
->
quit
();
},
[
=
](
publish_atom
)
->
expected
<
uint16_t
>
{
auto
res
=
self
->
add_tcp_doorman
(
8080
);
if
(
!
res
)
[
=
](
publish_atom
)
->
result
<
uint16_t
>
{
if
(
auto
res
=
self
->
add_tcp_doorman
(
8080
))
return
res
->
second
;
else
return
std
::
move
(
res
.
error
());
return
res
->
second
;
},
};
}
...
...
@@ -153,13 +153,13 @@ CAF_TEST(test broker to broker communication) {
auto
ssp
=
std
::
make_shared
<
suite_state
>
();
auto
server
=
mars
.
mm
.
spawn_broker
(
peer_acceptor_fun
,
mars
.
sys
.
spawn
(
pong
,
ssp
));
mars
.
self
->
send
(
server
,
publish_atom
::
value
);
mars
.
self
->
send
(
server
,
publish_atom
_v
);
run
();
expect_on
(
mars
,
(
uint16_t
),
from
(
server
).
to
(
mars
.
self
).
with
(
8080
));
CAF_MESSAGE
(
"spawn ping and client on earth"
);
auto
pinger
=
earth
.
sys
.
spawn
(
ping
,
ssp
);
auto
client
=
unbox
(
earth
.
mm
.
spawn_client
(
peer_fun
,
"mars"
,
8080
,
pinger
));
anon_send
(
pinger
,
kickoff_atom
::
value
,
client
);
anon_send
(
pinger
,
ok_atom_v
,
client
);
run
();
CAF_CHECK_EQUAL
(
ssp
->
pings
,
10
);
CAF_CHECK_EQUAL
(
ssp
->
pongs
,
10
);
...
...
libcaf_io/test/io/remote_actor.cpp
View file @
21281a6c
...
...
@@ -32,11 +32,6 @@ using namespace caf;
namespace
{
using
ping_atom
=
caf
::
atom_constant
<
caf
::
atom
(
"ping"
)
>
;
using
pong_atom
=
caf
::
atom_constant
<
caf
::
atom
(
"pong"
)
>
;
using
publish_atom
=
caf
::
atom_constant
<
caf
::
atom
(
"publish"
)
>
;
using
kickoff_atom
=
caf
::
atom_constant
<
caf
::
atom
(
"kickoff"
)
>
;
struct
suite_state
{
int
pings
=
0
;
int
pongs
=
0
;
...
...
@@ -48,13 +43,13 @@ using suite_state_ptr = std::shared_ptr<suite_state>;
behavior
ping
(
event_based_actor
*
self
,
suite_state_ptr
ssp
)
{
return
{
[
=
](
kickoff
_atom
,
const
actor
&
pong
)
{
CAF_MESSAGE
(
"received `
kickoff
_atom`"
);
[
=
](
ok
_atom
,
const
actor
&
pong
)
{
CAF_MESSAGE
(
"received `
ok
_atom`"
);
++
ssp
->
pings
;
self
->
send
(
pong
,
ping_atom
::
value
);
self
->
send
(
pong
,
ping_atom
_v
);
self
->
become
([
=
](
pong_atom
)
{
CAF_MESSAGE
(
"ping: received pong"
);
self
->
send
(
pong
,
ping_atom
::
value
);
self
->
send
(
pong
,
ping_atom
_v
);
if
(
++
ssp
->
pings
==
10
)
{
self
->
quit
();
CAF_MESSAGE
(
"ping is done"
);
...
...
@@ -66,13 +61,13 @@ behavior ping(event_based_actor* self, suite_state_ptr ssp) {
behavior
pong
(
event_based_actor
*
self
,
suite_state_ptr
ssp
)
{
return
{
[
=
](
ping_atom
)
->
atom_value
{
[
=
](
ping_atom
)
->
pong_atom
{
CAF_MESSAGE
(
"pong: received ping"
);
if
(
++
ssp
->
pongs
==
10
)
{
self
->
quit
();
CAF_MESSAGE
(
"pong is done"
);
}
return
pong_atom
::
value
;
return
pong_atom
_v
;
},
};
}
...
...
@@ -130,7 +125,7 @@ CAF_TEST(ping_pong) {
auto
port
=
mars
.
publish
(
mars
.
sys
.
spawn
(
pong
,
ssp
),
8080
);
CAF_CHECK_EQUAL
(
port
,
8080u
);
auto
remote_pong
=
earth
.
remote_actor
(
"mars"
,
8080
);
anon_send
(
earth
.
sys
.
spawn
(
ping
,
ssp
),
kickoff_atom
::
value
,
remote_pong
);
anon_send
(
earth
.
sys
.
spawn
(
ping
,
ssp
),
ok_atom_v
,
remote_pong
);
run
();
CAF_CHECK_EQUAL
(
ssp
->
pings
,
10
);
CAF_CHECK_EQUAL
(
ssp
->
pongs
,
10
);
...
...
libcaf_io/test/io/remote_group.cpp
View file @
21281a6c
...
...
@@ -107,7 +107,7 @@ CAF_TEST_DISABLED(message transmission) {
{
received_messages
=
0u
;
scoped_actor
self
{
mars
.
sys
};
self
->
send
(
mars_grp
,
ok_atom
::
value
);
self
->
send
(
mars_grp
,
ok_atom
_v
);
exec_all
();
CAF_CHECK_EQUAL
(
received_messages
,
10u
);
}
...
...
@@ -115,7 +115,7 @@ CAF_TEST_DISABLED(message transmission) {
{
received_messages
=
0u
;
scoped_actor
self
{
earth
.
sys
};
self
->
send
(
earth_grp
,
ok_atom
::
value
);
self
->
send
(
earth_grp
,
ok_atom
_v
);
exec_all
();
CAF_CHECK_EQUAL
(
received_messages
,
10u
);
}
...
...
libcaf_io/test/io/remote_spawn.cpp
View file @
21281a6c
...
...
@@ -99,10 +99,10 @@ CAF_TEST(nodes can spawn actors remotely) {
calc
=
earth
.
mm
.
remote_spawn
<
calculator
>
(
nid
,
"typed_calculator"
,
make_message
());
CAF_MESSAGE
(
"remotely spawned actors respond to messages"
);
earth
.
self
->
send
(
*
calc
,
add_atom
::
value
,
10
,
20
);
earth
.
self
->
send
(
*
calc
,
add_atom
_v
,
10
,
20
);
run
();
expect_on
(
earth
,
(
int
),
from
(
*
calc
).
to
(
earth
.
self
).
with
(
30
));
earth
.
self
->
send
(
*
calc
,
sub_atom
::
value
,
10
,
20
);
earth
.
self
->
send
(
*
calc
,
sub_atom
_v
,
10
,
20
);
run
();
expect_on
(
earth
,
(
int
),
from
(
*
calc
).
to
(
earth
.
self
).
with
(
-
10
));
anon_send_exit
(
*
calc
,
exit_reason
::
user_shutdown
);
...
...
@@ -111,7 +111,7 @@ CAF_TEST(nodes can spawn actors remotely) {
auto
dyn_calc
=
earth
.
mm
.
remote_spawn
<
actor
>
(
nid
,
"calculator-class"
,
make_message
());
CAF_REQUIRE
(
dyn_calc
);
earth
.
self
->
send
(
*
dyn_calc
,
add_atom
::
value
,
10
,
20
);
earth
.
self
->
send
(
*
dyn_calc
,
add_atom
_v
,
10
,
20
);
run
();
expect_on
(
earth
,
(
int
),
from
(
*
dyn_calc
).
to
(
earth
.
self
).
with
(
30
));
anon_send_exit
(
*
dyn_calc
,
exit_reason
::
user_shutdown
);
...
...
libcaf_io/test/io/worker.cpp
View file @
21281a6c
...
...
@@ -108,7 +108,7 @@ CAF_TEST(deliver serialized message) {
std
::
vector
<
char
>
payload
;
std
::
vector
<
strong_actor_ptr
>
stages
;
binary_serializer
sink
{
sys
,
payload
};
if
(
auto
err
=
sink
(
stages
,
make_message
(
ok_atom
::
value
)))
if
(
auto
err
=
sink
(
stages
,
make_message
(
ok_atom
_v
)))
CAF_FAIL
(
"unable to serialize message: "
<<
sys
.
render
(
err
));
io
::
basp
::
header
hdr
{
io
::
basp
::
message_type
::
direct_message
,
0
,
...
...
libcaf_openssl/caf/openssl/unpublish.hpp
View file @
21281a6c
...
...
@@ -42,7 +42,7 @@ expected<void> unpublish(const Handle& whom, uint16_t port = 0) {
return
sec
::
invalid_argument
;
auto
&
sys
=
whom
.
home_system
();
auto
f
=
make_function_view
(
sys
.
openssl_manager
().
actor_handle
());
return
f
(
unpublish_atom
::
value
,
port
);
return
f
(
unpublish_atom
_v
,
port
);
}
}
// namespace openssl
...
...
libcaf_openssl/src/openssl/publish.cpp
View file @
21281a6c
...
...
@@ -40,7 +40,7 @@ expected<uint16_t> publish(actor_system& sys, const strong_actor_ptr& whom,
if
(
cstr
!=
nullptr
)
in
=
cstr
;
auto
f
=
make_function_view
(
sys
.
openssl_manager
().
actor_handle
());
return
f
(
publish_atom
::
value
,
port
,
std
::
move
(
whom
),
std
::
move
(
sigs
),
return
f
(
publish_atom
_v
,
port
,
std
::
move
(
whom
),
std
::
move
(
sigs
),
std
::
move
(
in
),
ru
);
}
...
...
libcaf_openssl/src/openssl/remote_actor.cpp
View file @
21281a6c
...
...
@@ -39,7 +39,7 @@ expected<strong_actor_ptr> remote_actor(actor_system& sys,
CAF_LOG_TRACE
(
CAF_ARG
(
mpi
)
<<
CAF_ARG
(
host
)
<<
CAF_ARG
(
port
));
expected
<
strong_actor_ptr
>
res
{
strong_actor_ptr
{
nullptr
}};
auto
f
=
make_function_view
(
sys
.
openssl_manager
().
actor_handle
());
auto
x
=
f
(
connect_atom
::
value
,
std
::
move
(
host
),
port
);
auto
x
=
f
(
connect_atom
_v
,
std
::
move
(
host
),
port
);
if
(
!
x
)
return
std
::
move
(
x
.
error
());
auto
&
tup
=
*
x
;
...
...
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