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
e22067f8
Commit
e22067f8
authored
Mar 17, 2020
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Allow actors to monitor CAF nodes
parent
383f3c28
Changes
15
Hide whitespace changes
Inline
Side-by-side
Showing
15 changed files
with
349 additions
and
8 deletions
+349
-8
CHANGELOG.md
CHANGELOG.md
+7
-0
libcaf_core/caf/actor_system.hpp
libcaf_core/caf/actor_system.hpp
+27
-0
libcaf_core/caf/local_actor.hpp
libcaf_core/caf/local_actor.hpp
+9
-1
libcaf_core/caf/scheduled_actor.hpp
libcaf_core/caf/scheduled_actor.hpp
+24
-0
libcaf_core/caf/system_messages.hpp
libcaf_core/caf/system_messages.hpp
+28
-0
libcaf_core/src/actor_system.cpp
libcaf_core/src/actor_system.cpp
+22
-0
libcaf_core/src/local_actor.cpp
libcaf_core/src/local_actor.cpp
+8
-0
libcaf_core/src/scheduled_actor.cpp
libcaf_core/src/scheduled_actor.cpp
+12
-0
libcaf_io/CMakeLists.txt
libcaf_io/CMakeLists.txt
+1
-0
libcaf_io/caf/io/basp_broker.hpp
libcaf_io/caf/io/basp_broker.hpp
+7
-0
libcaf_io/caf/io/middleman.hpp
libcaf_io/caf/io/middleman.hpp
+5
-1
libcaf_io/src/io/basp_broker.cpp
libcaf_io/src/io/basp_broker.cpp
+61
-1
libcaf_io/src/io/middleman.cpp
libcaf_io/src/io/middleman.cpp
+10
-0
libcaf_io/test/io/monitor.cpp
libcaf_io/test/io/monitor.cpp
+116
-0
libcaf_test/caf/test/io_dsl.hpp
libcaf_test/caf/test/io_dsl.hpp
+12
-5
No files found.
CHANGELOG.md
View file @
e22067f8
...
...
@@ -37,6 +37,9 @@ is based on [Keep a Changelog](https://keepachangelog.com).
-
The class
`exit_msg`
finally got its missing
`operator==`
(#1039).
-
The class
`node_id`
received an overload for
`parse`
to allow users to convert
the output of
`to_string`
back to the original ID (#1058).
-
Actors can now
`monitor`
and
`demonitor`
CAF nodes (#1042). Monitoring a CAF
node causes the actor system to send a
`node_down_msg`
to the observer when
losing connection to the monitored node.
### Changed
...
...
@@ -82,6 +85,10 @@ is based on [Keep a Changelog](https://keepachangelog.com).
deadlocks when calling blocking functions in message handlers. This function
now behaves as expected (#1016).
-
Exceptions while handling requests now trigger error messages (#1055).
-
The member function
`demonitor`
falsely refused typed actor handles. Actors
could monitor typed actors but not demonitoring it again. This member function
is now a template that accepts any actor handle in the same way
`monitor`
already did.
## [0.17.5] - Unreleased
...
...
libcaf_core/caf/actor_system.hpp
View file @
e22067f8
...
...
@@ -167,6 +167,22 @@ public:
using
module_array
=
std
::
array
<
module_ptr
,
module
::
num_ids
>
;
/// An (optional) component of the actor system with networking capabilities.
class
CAF_CORE_EXPORT
networking_module
:
public
module
{
public:
~
networking_module
()
override
;
/// Causes the module to send a `node_down_msg` to `observer` if this system
/// loses connection to `node`.
virtual
void
monitor
(
const
node_id
&
node
,
const
actor_addr
&
observer
)
=
0
;
/// Causes the module remove one entry for `observer` from the list of
/// actors that receive a `node_down_msg` if this system loses connection to
/// `node`. Each call to `monitor` requires one call to `demonitor` in order
/// to unsubscribe the `observer` completely.
virtual
void
demonitor
(
const
node_id
&
node
,
const
actor_addr
&
observer
)
=
0
;
};
/// @warning The system stores a reference to `cfg`, which means the
/// config object must outlive the actor system.
explicit
actor_system
(
actor_system_config
&
cfg
);
...
...
@@ -279,6 +295,17 @@ public:
/// Blocks this caller until all actors are done.
void
await_all_actors_done
()
const
;
/// Send a `node_down_msg` to `observer` if this system loses connection to
/// `node`.
/// @note Calling this function *n* times causes the system to send
/// `node_down_msg` *n* times. In order to not receive the messages, the
/// observer must call `demonitor` *n* times.
void
monitor
(
const
node_id
&
node
,
const
actor_addr
&
observer
);
/// Removes `observer` from the list of actors that receive a `node_down_msg`
/// if this system loses connection to `node`.
void
demonitor
(
const
node_id
&
node
,
const
actor_addr
&
observer
);
/// Called by `spawn` when used to create a class-based actor to
/// apply automatic conversions to `xs` before spawning the actor.
/// Should not be called by users of the library directly.
...
...
libcaf_core/caf/local_actor.hpp
View file @
e22067f8
...
...
@@ -288,6 +288,10 @@ public:
current_element_
=
ptr
;
}
/// Adds a unidirectional `monitor` to `node`.
/// @note Each call to `monitor` creates a new, independent monitor.
void
monitor
(
const
node_id
&
node
);
/// Adds a unidirectional `monitor` to `whom`.
/// @note Each call to `monitor` creates a new, independent monitor.
template
<
message_priority
P
=
message_priority
::
normal
,
class
Handle
=
actor
>
...
...
@@ -298,8 +302,12 @@ public:
/// Removes a monitor from `whom`.
void
demonitor
(
const
actor_addr
&
whom
);
/// Removes a monitor from `node`.
void
demonitor
(
const
node_id
&
node
);
/// Removes a monitor from `whom`.
inline
void
demonitor
(
const
actor
&
whom
)
{
template
<
class
Handle
>
void
demonitor
(
const
Handle
&
whom
)
{
demonitor
(
whom
.
address
());
}
...
...
libcaf_core/caf/scheduled_actor.hpp
View file @
e22067f8
...
...
@@ -200,6 +200,9 @@ public:
/// Function object for handling down messages.
using
down_handler
=
std
::
function
<
void
(
pointer
,
down_msg
&
)
>
;
/// Function object for handling node down messages.
using
node_down_handler
=
std
::
function
<
void
(
pointer
,
node_down_msg
&
)
>
;
/// Function object for handling exit messages.
using
exit_handler
=
std
::
function
<
void
(
pointer
,
exit_msg
&
)
>
;
...
...
@@ -241,6 +244,8 @@ public:
static
void
default_down_handler
(
pointer
ptr
,
down_msg
&
x
);
static
void
default_node_down_handler
(
pointer
ptr
,
node_down_msg
&
x
);
static
void
default_exit_handler
(
pointer
ptr
,
exit_msg
&
x
);
#ifndef CAF_NO_EXCEPTIONS
...
...
@@ -378,6 +383,22 @@ public:
set_down_handler
([
fun
](
scheduled_actor
*
,
down_msg
&
x
)
{
fun
(
x
);
});
}
/// Sets a custom handler for node down messages.
void
set_node_down_handler
(
node_down_handler
fun
)
{
if
(
fun
)
node_down_handler_
=
std
::
move
(
fun
);
else
node_down_handler_
=
default_node_down_handler
;
}
/// Sets a custom handler for down messages.
template
<
class
T
>
auto
set_node_down_handler
(
T
fun
)
->
decltype
(
fun
(
std
::
declval
<
node_down_msg
&>
()))
{
set_node_down_handler
(
[
fun
](
scheduled_actor
*
,
node_down_msg
&
x
)
{
fun
(
x
);
});
}
/// Sets a custom handler for error messages.
inline
void
set_exit_handler
(
exit_handler
fun
)
{
if
(
fun
)
...
...
@@ -884,6 +905,9 @@ protected:
/// Customization point for setting a default `down_msg` callback.
down_handler
down_handler_
;
/// Customization point for setting a default `down_msg` callback.
node_down_handler
node_down_handler_
;
/// Customization point for setting a default `exit_msg` callback.
exit_handler
exit_handler_
;
...
...
libcaf_core/caf/system_messages.hpp
View file @
e22067f8
...
...
@@ -90,6 +90,34 @@ typename Inspector::result_type inspect(Inspector& f, group_down_msg& x) {
return
f
(
meta
::
type_name
(
"group_down_msg"
),
x
.
source
);
}
/// Sent to all actors monitoring a node when CAF loses connection to it.
struct
node_down_msg
{
/// The disconnected node.
node_id
node
;
/// The cause for the disconnection. No error (a default-constructed error
/// object) indicates an ordinary shutdown.
error
reason
;
};
/// @relates node_down_msg
inline
bool
operator
==
(
const
node_down_msg
&
x
,
const
node_down_msg
&
y
)
noexcept
{
return
x
.
node
==
y
.
node
&&
x
.
reason
==
y
.
reason
;
}
/// @relates node_down_msg
inline
bool
operator
!=
(
const
node_down_msg
&
x
,
const
node_down_msg
&
y
)
noexcept
{
return
!
(
x
==
y
);
}
/// @relates node_down_msg
template
<
class
Inspector
>
typename
Inspector
::
result_type
inspect
(
Inspector
&
f
,
node_down_msg
&
x
)
{
return
f
(
meta
::
type_name
(
"node_down_msg"
),
x
.
node
,
x
.
reason
);
}
/// Signalizes a timeout event.
/// @note This message is handled implicitly by the runtime system.
struct
timeout_msg
{
...
...
libcaf_core/src/actor_system.cpp
View file @
e22067f8
...
...
@@ -211,6 +211,10 @@ const char* actor_system::module::name() const noexcept {
}
}
actor_system
::
networking_module
::~
networking_module
()
{
// nop
}
actor_system
::
actor_system
(
actor_system_config
&
cfg
)
:
profiler_
(
cfg
.
profiler
),
ids_
(
0
),
...
...
@@ -412,6 +416,24 @@ void actor_system::await_all_actors_done() const {
registry_
.
await_running_count_equal
(
0
);
}
void
actor_system
::
monitor
(
const
node_id
&
node
,
const
actor_addr
&
observer
)
{
// TODO: Currently does not work with other modules, in particular caf_net.
auto
mm
=
modules_
[
module
::
middleman
].
get
();
if
(
mm
==
nullptr
)
return
;
auto
mm_dptr
=
static_cast
<
networking_module
*>
(
mm
);
mm_dptr
->
monitor
(
node
,
observer
);
}
void
actor_system
::
demonitor
(
const
node_id
&
node
,
const
actor_addr
&
observer
)
{
// TODO: Currently does not work with other modules, in particular caf_net.
auto
mm
=
modules_
[
module
::
middleman
].
get
();
if
(
mm
==
nullptr
)
return
;
auto
mm_dptr
=
static_cast
<
networking_module
*>
(
mm
);
mm_dptr
->
demonitor
(
node
,
observer
);
}
actor_clock
&
actor_system
::
clock
()
noexcept
{
return
scheduler
().
clock
();
}
...
...
libcaf_core/src/local_actor.cpp
View file @
e22067f8
...
...
@@ -74,6 +74,10 @@ void local_actor::monitor(abstract_actor* ptr, message_priority priority) {
default_attachable
::
make_monitor
(
ptr
->
address
(),
address
(),
priority
));
}
void
local_actor
::
monitor
(
const
node_id
&
node
)
{
system
().
monitor
(
node
,
address
());
}
void
local_actor
::
demonitor
(
const
actor_addr
&
whom
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
whom
));
auto
ptr
=
actor_cast
<
strong_actor_ptr
>
(
whom
);
...
...
@@ -84,6 +88,10 @@ void local_actor::demonitor(const actor_addr& whom) {
}
}
void
local_actor
::
demonitor
(
const
node_id
&
node
)
{
system
().
demonitor
(
node
,
address
());
}
void
local_actor
::
on_exit
()
{
// nop
}
...
...
libcaf_core/src/scheduled_actor.cpp
View file @
e22067f8
...
...
@@ -86,6 +86,12 @@ void scheduled_actor::default_down_handler(scheduled_actor* ptr, down_msg& x) {
<<
", name: "
<<
ptr
->
name
()
<<
"]: "
<<
to_string
(
x
)
<<
std
::
endl
;
}
void
scheduled_actor
::
default_node_down_handler
(
scheduled_actor
*
ptr
,
node_down_msg
&
x
)
{
aout
(
ptr
)
<<
"*** unhandled node down message [id: "
<<
ptr
->
id
()
<<
", name: "
<<
ptr
->
name
()
<<
"]: "
<<
to_string
(
x
)
<<
std
::
endl
;
}
void
scheduled_actor
::
default_exit_handler
(
scheduled_actor
*
ptr
,
exit_msg
&
x
)
{
if
(
x
.
reason
)
default_error_handler
(
ptr
,
x
.
reason
);
...
...
@@ -120,6 +126,7 @@ scheduled_actor::scheduled_actor(actor_config& cfg)
default_handler_
(
print_and_drop
),
error_handler_
(
default_error_handler
),
down_handler_
(
default_down_handler
),
node_down_handler_
(
default_node_down_handler
),
exit_handler_
(
default_exit_handler
),
private_thread_
(
nullptr
)
#ifndef CAF_NO_EXCEPTIONS
...
...
@@ -609,6 +616,11 @@ scheduled_actor::categorize(mailbox_element& x) {
call_handler
(
down_handler_
,
this
,
dm
);
return
message_category
::
internal
;
}
if
(
content
.
match_elements
<
node_down_msg
>
())
{
auto
dm
=
content
.
move_if_unshared
<
node_down_msg
>
(
0
);
call_handler
(
node_down_handler_
,
this
,
dm
);
return
message_category
::
internal
;
}
if
(
content
.
match_elements
<
error
>
())
{
auto
err
=
content
.
move_if_unshared
<
error
>
(
0
);
call_handler
(
error_handler_
,
this
,
err
);
...
...
libcaf_io/CMakeLists.txt
View file @
e22067f8
...
...
@@ -60,6 +60,7 @@ set(CAF_IO_TEST_SOURCES
test/io/basp_broker.cpp
test/io/broker.cpp
test/io/http_broker.cpp
test/io/monitor.cpp
test/io/network/default_multiplexer.cpp
test/io/network/ip_endpoint.cpp
test/io/receive_buffer.cpp
...
...
libcaf_io/caf/io/basp_broker.hpp
View file @
e22067f8
...
...
@@ -104,6 +104,9 @@ public:
// -- utility functions ------------------------------------------------------
/// Sends `node_down_msg` to all registered observers.
void
emit_node_down_msg
(
const
node_id
&
node
,
const
error
&
reason
);
/// Performs bookkeeping such as managing `spawn_servers`.
void
learned_new_node
(
const
node_id
&
nid
);
...
...
@@ -147,6 +150,10 @@ public:
/// 'SpawnServ' instance on the remote side.
std
::
unordered_map
<
node_id
,
actor
>
spawn_servers
;
/// Keeps track of actors that wish to receive a `node_down_msg` if a
/// particular node fails.
std
::
unordered_map
<
node_id
,
std
::
vector
<
actor_addr
>>
node_observers
;
/// Configures whether BASP automatically open new connections to optimize
/// routing paths by forming a mesh between all nodes.
bool
automatic_connections
=
false
;
...
...
libcaf_io/caf/io/middleman.hpp
View file @
e22067f8
...
...
@@ -40,7 +40,7 @@
namespace
caf
::
io
{
/// Manages brokers and network backends.
class
CAF_IO_EXPORT
middleman
:
public
actor_system
::
module
{
class
CAF_IO_EXPORT
middleman
:
public
actor_system
::
networking_
module
{
public:
friend
class
::
caf
::
actor_system
;
...
...
@@ -195,6 +195,10 @@ public:
void
*
subtype_ptr
()
override
;
void
monitor
(
const
node_id
&
node
,
const
actor_addr
&
observer
)
override
;
void
demonitor
(
const
node_id
&
node
,
const
actor_addr
&
observer
)
override
;
/// Spawns a new functor-based broker.
template
<
spawn_options
Os
=
no_spawn_options
,
class
F
=
std
::
function
<
void
(
broker
*
)>,
class
...
Ts
>
...
...
libcaf_io/src/io/basp_broker.cpp
View file @
e22067f8
...
...
@@ -77,6 +77,13 @@ void basp_broker::on_exit() {
// that the middleman calls this in its stop() function. However,
// ultimately we should find a nonblocking solution here.
instance
.
hub
().
await_workers
();
// All nodes are offline now. We use a default-constructed error code to
// signal ordinary shutdown.
for
(
const
auto
&
[
node
,
observer_list
]
:
node_observers
)
for
(
const
auto
&
observer
:
observer_list
)
if
(
auto
hdl
=
actor_cast
<
actor
>
(
observer
))
anon_send
(
hdl
,
node_down_msg
{
node
,
error
{}});
node_observers
.
clear
();
// Release any obsolete state.
ctx
.
clear
();
// Make sure all spawn servers are down before clearing the container.
...
...
@@ -200,6 +207,47 @@ behavior basp_broker::make_behavior() {
proxy
->
id
());
flush
(
hdl
);
},
// received from the middleman whenever a node becomes observed by a local
// actor
[
=
](
monitor_atom
,
const
node_id
&
node
,
const
actor_addr
&
observer
)
{
// Sanity checks.
if
(
!
observer
||
!
node
)
return
;
// Add to the list if a list for this node already exists.
auto
i
=
node_observers
.
find
(
node
);
if
(
i
!=
node_observers
.
end
())
{
i
->
second
.
emplace_back
(
observer
);
return
;
}
// Check whether the node is still connected at the moment and send the
// observer a message immediately otherwise.
if
(
instance
.
tbl
().
lookup
(
node
)
==
none
)
{
if
(
auto
hdl
=
actor_cast
<
actor
>
(
observer
))
{
// TODO: we might want to consider keeping the exit reason for nodes,
// at least for some time. Otherwise, we'll have to send a
// generic "don't know" exit reason. Probably an improvement we
// should consider in caf_net.
anon_send
(
hdl
,
node_down_msg
{
node
,
sec
::
no_context
});
}
return
;
}
std
::
vector
<
actor_addr
>
xs
{
observer
};
node_observers
.
emplace
(
node
,
std
::
move
(
xs
));
},
// received from the middleman whenever a node becomes observed by a local
// actor
[
=
](
demonitor_atom
,
const
node_id
&
node
,
const
actor_addr
&
observer
)
{
auto
i
=
node_observers
.
find
(
node
);
if
(
i
==
node_observers
.
end
())
return
;
auto
&
observers
=
i
->
second
;
auto
j
=
std
::
find
(
observers
.
begin
(),
observers
.
end
(),
observer
);
if
(
j
!=
observers
.
end
())
{
observers
.
erase
(
j
);
if
(
observers
.
empty
())
node_observers
.
erase
(
i
);
}
},
// received from underlying broker implementation
[
=
](
const
new_connection_msg
&
msg
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
msg
.
handle
));
...
...
@@ -460,6 +508,16 @@ void basp_broker::handle_down_msg(down_msg& dm) {
monitored_actors
.
erase
(
i
);
}
void
basp_broker
::
emit_node_down_msg
(
const
node_id
&
node
,
const
error
&
reason
)
{
auto
i
=
node_observers
.
find
(
node
);
if
(
i
==
node_observers
.
end
())
return
;
for
(
const
auto
&
observer
:
i
->
second
)
if
(
auto
hdl
=
actor_cast
<
actor
>
(
observer
))
anon_send
(
hdl
,
node_down_msg
{
node
,
reason
});
node_observers
.
erase
(
i
);
}
void
basp_broker
::
learned_new_node
(
const
node_id
&
nid
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
nid
));
if
(
spawn_servers
.
count
(
nid
)
>
0
)
{
...
...
@@ -567,8 +625,10 @@ void basp_broker::connection_cleanup(connection_handle hdl, sec code) {
CAF_LOG_TRACE
(
CAF_ARG
(
hdl
)
<<
CAF_ARG
(
code
));
// Remove handle from the routing table and clean up any node-specific state
// we might still have.
if
(
auto
nid
=
instance
.
tbl
().
erase_direct
(
hdl
))
if
(
auto
nid
=
instance
.
tbl
().
erase_direct
(
hdl
))
{
emit_node_down_msg
(
nid
,
code
);
purge_state
(
nid
);
}
// Remove the context for `hdl`, making sure clients receive an error in case
// this connection was closed during handshake.
auto
i
=
ctx
.
find
(
hdl
);
...
...
libcaf_io/src/io/middleman.cpp
View file @
e22067f8
...
...
@@ -399,6 +399,16 @@ void* middleman::subtype_ptr() {
return
this
;
}
void
middleman
::
monitor
(
const
node_id
&
node
,
const
actor_addr
&
observer
)
{
auto
basp
=
named_broker
<
basp_broker
>
(
"BASP"
);
anon_send
(
basp
,
monitor_atom_v
,
node
,
observer
);
}
void
middleman
::
demonitor
(
const
node_id
&
node
,
const
actor_addr
&
observer
)
{
auto
basp
=
named_broker
<
basp_broker
>
(
"BASP"
);
anon_send
(
basp
,
demonitor_atom_v
,
node
,
observer
);
}
middleman
::~
middleman
()
{
// nop
}
...
...
libcaf_io/test/io/monitor.cpp
0 → 100644
View file @
e22067f8
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE io.monitor
#include "caf/io/middleman.hpp"
#include "caf/test/io_dsl.hpp"
using
namespace
caf
;
namespace
{
struct
fixture
:
point_to_point_fixture
<>
{
actor
observer
;
fixture
()
{
mars_id
=
mars
.
sys
.
node
();
}
// Connects earth and mars, storing the connection handles in earth_conn and
// mars_conn.
void
connect
()
{
auto
acc
=
next_accept_handle
();
std
::
tie
(
earth_conn
,
mars_conn
)
=
prepare_connection
(
earth
,
mars
,
"localhost"
,
8080
,
acc
);
CAF_CHECK_EQUAL
(
earth
.
publish
(
actor
{
earth
.
self
},
8080
),
8080
);
CAF_CHECK
(
mars
.
remote_actor
(
"localhost"
,
8080
));
}
void
disconnect
()
{
anon_send
(
earth
.
bb
,
io
::
connection_closed_msg
{
earth_conn
});
earth
.
handle_io_event
();
anon_send
(
mars
.
bb
,
io
::
connection_closed_msg
{
mars_conn
});
mars
.
handle_io_event
();
}
node_id
mars_id
;
io
::
connection_handle
earth_conn
;
io
::
connection_handle
mars_conn
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
monitor_tests
,
fixture
)
CAF_TEST
(
disconnects
cause
node_down_msg
)
{
connect
();
earth
.
self
->
monitor
(
mars_id
);
run
();
disconnect
();
expect_on
(
earth
,
(
node_down_msg
),
to
(
earth
.
self
).
with
(
node_down_msg
{
mars_id
,
error
{}}));
CAF_CHECK
(
earth
.
self
->
mailbox
().
empty
());
}
CAF_TEST
(
node_down_msg
calls
the
special
node_down_handler
)
{
connect
();
bool
node_down_handler_called
=
false
;
auto
observer
=
earth
.
sys
.
spawn
([
&
](
event_based_actor
*
self
)
->
behavior
{
self
->
monitor
(
mars_id
);
self
->
set_node_down_handler
([
&
](
node_down_msg
&
dm
)
{
CAF_CHECK_EQUAL
(
dm
.
node
,
mars_id
);
node_down_handler_called
=
true
;
});
return
[]
{};
});
run
();
disconnect
();
expect_on
(
earth
,
(
node_down_msg
),
to
(
observer
).
with
(
node_down_msg
{
mars_id
,
error
{}}));
CAF_CHECK
(
node_down_handler_called
);
}
CAF_TEST
(
calling
monitor
n
times
produces
n
node_down_msg
)
{
connect
();
for
(
int
i
=
0
;
i
<
5
;
++
i
)
earth
.
self
->
monitor
(
mars_id
);
run
();
disconnect
();
for
(
int
i
=
0
;
i
<
5
;
++
i
)
expect_on
(
earth
,
(
node_down_msg
),
to
(
earth
.
self
).
with
(
node_down_msg
{
mars_id
,
error
{}}));
CAF_CHECK_EQUAL
(
earth
.
self
->
mailbox
().
size
(),
0u
);
}
CAF_TEST
(
each
demonitor
only
cancels
one
node_down_msg
)
{
connect
();
for
(
int
i
=
0
;
i
<
5
;
++
i
)
earth
.
self
->
monitor
(
mars_id
);
run
();
earth
.
self
->
demonitor
(
mars_id
);
run
();
disconnect
();
for
(
int
i
=
0
;
i
<
4
;
++
i
)
expect_on
(
earth
,
(
node_down_msg
),
to
(
earth
.
self
).
with
(
node_down_msg
{
mars_id
,
error
{}}));
CAF_CHECK_EQUAL
(
earth
.
self
->
mailbox
().
size
(),
0u
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_test/caf/test/io_dsl.hpp
View file @
e22067f8
...
...
@@ -72,7 +72,7 @@ public:
:
mm
(
this
->
sys
.
middleman
()),
mpx
(
dynamic_cast
<
caf
::
io
::
network
::
test_multiplexer
&>
(
mm
.
backend
())),
run_all_nodes
(
std
::
move
(
fun
))
{
// nop
bb
=
mm
.
named_broker
<
caf
::
io
::
basp_broker
>
(
"BASP"
);
}
test_node_fixture
()
:
test_node_fixture
([
=
]
{
this
->
run
();
})
{
...
...
@@ -109,6 +109,9 @@ public:
/// Reference to the middleman's event multiplexer.
caf
::
io
::
network
::
test_multiplexer
&
mpx
;
/// Handle to the BASP broker.
caf
::
actor
bb
;
/// Callback for triggering all nodes when simulating a network of CAF nodes.
run_all_nodes_fun
run_all_nodes
;
...
...
@@ -284,9 +287,13 @@ public:
};
#define expect_on(where, types, fields) \
CAF_MESSAGE(#where << ": expect" << #types << "." << #fields); \
expect_clause< CAF_EXPAND(CAF_DSL_LIST types) >{where . sched} . fields
do { \
CAF_MESSAGE(#where << ": expect" << #types << "." << #fields); \
expect_clause<CAF_EXPAND(CAF_DSL_LIST types)>{where.sched}.fields; \
} while (false)
#define disallow_on(where, types, fields) \
CAF_MESSAGE(#where << ": disallow" << #types << "." << #fields); \
disallow_clause< CAF_EXPAND(CAF_DSL_LIST types) >{where . sched} . fields
do { \
CAF_MESSAGE(#where << ": disallow" << #types << "." << #fields); \
disallow_clause<CAF_EXPAND(CAF_DSL_LIST types)>{where.sched}.fields; \
} while (false)
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