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
fbdb3351
Commit
fbdb3351
authored
Nov 08, 2019
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Remove localhost connections from I/O unit tests
parent
c07f3664
Changes
8
Show whitespace changes
Inline
Side-by-side
Showing
8 changed files
with
261 additions
and
599 deletions
+261
-599
libcaf_io/test/io/basp/message_queue.cpp
libcaf_io/test/io/basp/message_queue.cpp
+1
-1
libcaf_io/test/io/broker.cpp
libcaf_io/test/io/broker.cpp
+76
-88
libcaf_io/test/io/network/default_multiplexer.cpp
libcaf_io/test/io/network/default_multiplexer.cpp
+0
-23
libcaf_io/test/io/remote_actor.cpp
libcaf_io/test/io/remote_actor.cpp
+70
-112
libcaf_io/test/io/remote_spawn.cpp
libcaf_io/test/io/remote_spawn.cpp
+45
-45
libcaf_io/test/io/triggering.cpp
libcaf_io/test/io/triggering.cpp
+0
-252
libcaf_io/test/io/unpublish.cpp
libcaf_io/test/io/unpublish.cpp
+46
-69
libcaf_test/caf/test/io_dsl.hpp
libcaf_test/caf/test/io_dsl.hpp
+23
-9
No files found.
libcaf_io/test/io/message_queue.cpp
→
libcaf_io/test/io/
basp/
message_queue.cpp
View file @
fbdb3351
...
...
@@ -16,7 +16,7 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE io
_basp_
message_queue
#define CAF_SUITE io
.basp.
message_queue
#include "caf/io/basp/message_queue.hpp"
...
...
libcaf_io/test/io/broker.cpp
View file @
fbdb3351
...
...
@@ -20,7 +20,7 @@
#include "caf/io/broker.hpp"
#include "caf/test/dsl.hpp"
#include "caf/test/
io_
dsl.hpp"
#include <iostream>
#include <memory>
...
...
@@ -28,7 +28,6 @@
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using
namespace
std
;
using
namespace
caf
;
using
namespace
caf
::
io
;
...
...
@@ -39,40 +38,41 @@ 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"
)
>
;
behavior
ping
(
event_based_actor
*
self
,
size_t
num_pings
)
{
CAF_MESSAGE
(
"num_pings: "
<<
num_pings
);
auto
count
=
std
::
make_shared
<
size_t
>
(
0
);
struct
suite_state
{
int
pings
=
0
;
int
pongs
=
0
;
suite_state
()
=
default
;
};
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`"
);
self
->
send
(
pong
,
ping_atom
::
value
,
1
);
self
->
become
([
=
](
pong_atom
,
int
value
)
->
std
::
tuple
<
atom_value
,
int
>
{
if
(
++*
count
>=
num_pings
)
{
CAF_MESSAGE
(
"received "
<<
num_pings
<<
" pings, call self->quit"
);
++
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"
);
}
return
std
::
make_tuple
(
ping_atom
::
value
,
value
+
1
);
});
},
};
}
behavior
pong
(
event_based_actor
*
self
)
{
CAF_MESSAGE
(
"pong actor started"
);
self
->
set_down_handler
([
=
](
down_msg
&
dm
)
{
CAF_MESSAGE
(
"received down_msg{"
<<
to_string
(
dm
.
reason
)
<<
"}"
);
self
->
quit
(
dm
.
reason
);
});
behavior
pong
(
event_based_actor
*
self
,
suite_state_ptr
ssp
)
{
return
{
[
=
](
ping_atom
,
int
value
)
->
std
::
tuple
<
atom_value
,
int
>
{
CAF_MESSAGE
(
"received `ping_atom`"
);
self
->
monitor
(
self
->
current_sender
());
// set next behavior
self
->
become
([](
ping_atom
,
int
val
)
{
return
std
::
make_tuple
(
pong_atom
::
value
,
val
);
});
// reply to 'ping'
return
std
::
make_tuple
(
pong_atom
::
value
,
value
);
[
=
](
ping_atom
)
->
atom_value
{
CAF_MESSAGE
(
"pong: received ping"
);
if
(
++
ssp
->
pongs
==
10
)
{
self
->
quit
();
CAF_MESSAGE
(
"pong is done"
);
}
return
pong_atom
::
value
;
},
};
}
...
...
@@ -81,24 +81,21 @@ behavior peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
CAF_MESSAGE
(
"peer_fun called"
);
CAF_REQUIRE
(
self
->
subtype
()
==
resumable
::
io_actor
);
CAF_CHECK
(
self
!=
nullptr
);
//
self
->
monitor
(
buddy
);
// assume exactly one connection
self
->
set_down_handler
([
self
](
down_msg
&
dm
)
{
// Stop if buddy is done.
self
->
quit
(
std
::
move
(
dm
.
reason
));
});
// Assume exactly one connection.
CAF_REQUIRE
(
self
->
connections
().
size
()
==
1
);
self
->
configure_read
(
hdl
,
receive_policy
::
exactly
(
sizeof
(
atom_value
)
+
sizeof
(
int
)));
auto
write
=
[
=
](
atom_value
type
,
int
value
)
{
self
->
configure_read
(
hdl
,
receive_policy
::
exactly
(
sizeof
(
atom_value
)));
auto
write
=
[
=
](
atom_value
type
)
{
auto
&
buf
=
self
->
wr_buf
(
hdl
);
auto
first
=
reinterpret_cast
<
char
*>
(
&
type
);
buf
.
insert
(
buf
.
end
(),
first
,
first
+
sizeof
(
atom_value
));
first
=
reinterpret_cast
<
char
*>
(
&
value
);
buf
.
insert
(
buf
.
end
(),
first
,
first
+
sizeof
(
int
));
self
->
flush
(
hdl
);
};
self
->
set_down_handler
([
=
](
down_msg
&
dm
)
{
CAF_MESSAGE
(
"received: "
<<
to_string
(
dm
));
if
(
dm
.
source
==
buddy
)
self
->
quit
(
dm
.
reason
);
});
return
{
[
=
](
const
connection_closed_msg
&
)
{
CAF_MESSAGE
(
"received connection_closed_msg"
);
...
...
@@ -107,19 +104,11 @@ 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
;
int
value
;
memcpy
(
&
type
,
msg
.
buf
.
data
(),
sizeof
(
atom_value
));
memcpy
(
&
value
,
msg
.
buf
.
data
()
+
sizeof
(
atom_value
),
sizeof
(
int
));
self
->
send
(
buddy
,
type
,
value
);
},
[
=
](
ping_atom
,
int
value
)
{
CAF_MESSAGE
(
"received: ping "
<<
value
);
write
(
ping_atom
::
value
,
value
);
},
[
=
](
pong_atom
,
int
value
)
{
CAF_MESSAGE
(
"received: pong "
<<
value
);
write
(
pong_atom
::
value
,
value
);
self
->
send
(
buddy
,
type
);
},
[
=
](
ping_atom
)
{
write
(
ping_atom
::
value
);
},
[
=
](
pong_atom
)
{
write
(
pong_atom
::
value
);
},
};
}
...
...
@@ -132,7 +121,7 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) {
self
->
quit
();
},
[
=
](
publish_atom
)
->
expected
<
uint16_t
>
{
auto
res
=
self
->
add_tcp_doorman
(
0
,
"127.0.0.1"
);
auto
res
=
self
->
add_tcp_doorman
(
8080
);
if
(
!
res
)
return
std
::
move
(
res
.
error
());
return
res
->
second
;
...
...
@@ -140,49 +129,48 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) {
};
}
void
run_client
(
int
argc
,
char
**
argv
,
uint16_t
port
)
{
actor_system_config
cfg
;
cfg
.
load
<
io
::
middleman
>
();
if
(
auto
err
=
cfg
.
parse
(
argc
,
argv
))
CAF_FAIL
(
"failed to parse config: "
<<
to_string
(
err
));
actor_system
system
{
cfg
};
auto
p
=
system
.
spawn
(
ping
,
size_t
{
10
});
CAF_MESSAGE
(
"spawn_client..."
);
auto
cl
=
unbox
(
system
.
middleman
().
spawn_client
(
peer_fun
,
"127.0.0.1"
,
port
,
p
));
CAF_MESSAGE
(
"spawn_client finished"
);
anon_send
(
p
,
kickoff_atom
::
value
,
cl
);
CAF_MESSAGE
(
"`kickoff_atom` has been send"
);
}
using
int_peer
=
connection_handler
::
extend
<
replies_to
<
int
>::
with
<
int
>>
;
void
run_server
(
int
argc
,
char
**
argv
)
{
actor_system_config
cfg
;
cfg
.
load
<
io
::
middleman
>
();
if
(
auto
err
=
cfg
.
parse
(
argc
,
argv
))
CAF_FAIL
(
"failed to parse config: "
<<
to_string
(
err
));
actor_system
system
{
cfg
};
scoped_actor
self
{
system
};
CAF_MESSAGE
(
"spawn peer acceptor"
);
auto
serv
=
system
.
middleman
().
spawn_broker
(
peer_acceptor_fun
,
system
.
spawn
(
pong
));
std
::
thread
child
;
self
->
request
(
serv
,
infinite
,
publish_atom
::
value
)
.
receive
(
[
&
](
uint16_t
port
)
{
CAF_MESSAGE
(
"server is running on port "
<<
port
);
child
=
std
::
thread
([
=
]
{
run_client
(
argc
,
argv
,
port
);
});
int_peer
::
behavior_type
int_peer_fun
(
int_peer
::
broker_pointer
)
{
return
{
[
=
](
const
connection_closed_msg
&
)
{
CAF_FAIL
(
"received connection_closed_msg"
);
},
[
&
](
const
error
&
err
)
{
CAF_ERROR
(
"Error: "
<<
self
->
system
().
render
(
err
));
});
self
->
await_all_other_actors_done
();
child
.
join
();
[
=
](
const
new_data_msg
&
)
{
CAF_FAIL
(
"received new_data_msg"
);
},
[
=
](
int
value
)
{
CAF_MESSAGE
(
"received: "
<<
value
);
return
value
;
},
};
}
}
// namespace
CAF_TEST
(
test_broker
)
{
auto
argc
=
test
::
engine
::
argc
();
auto
argv
=
test
::
engine
::
argv
();
run_server
(
argc
,
argv
);
CAF_TEST_FIXTURE_SCOPE
(
broker_tests
,
point_to_point_fixture
<>
)
CAF_TEST
(
test
broker
to
broker
communication
)
{
prepare_connection
(
mars
,
earth
,
"mars"
,
8080
);
CAF_MESSAGE
(
"spawn peer acceptor on mars"
);
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
);
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
);
run
();
CAF_CHECK_EQUAL
(
ssp
->
pings
,
10
);
CAF_CHECK_EQUAL
(
ssp
->
pongs
,
10
);
}
CAF_TEST
(
test
whether
we
can
spawn
typed
broker
)
{
auto
peer
=
mars
.
mm
.
spawn_broker
(
int_peer_fun
);
mars
.
self
->
send
(
peer
,
42
);
run
();
expect_on
(
mars
,
(
int
),
from
(
peer
).
to
(
mars
.
self
).
with
(
42
));
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_io/test/io/network/default_multiplexer.cpp
View file @
fbdb3351
...
...
@@ -79,27 +79,4 @@ CAF_TEST(doorman io_failure) {
CAF_CHECK_EQUAL
(
server
.
mpx
.
num_socket_handlers
(),
1u
);
}
CAF_TEST
(
scribe
io_failure
)
{
CAF_MESSAGE
(
"add doorman to server"
);
CAF_CHECK_EQUAL
(
server
.
mpx
.
num_socket_handlers
(),
1u
);
auto
doorman
=
unbox
(
server
.
mpx
.
new_tcp_doorman
(
0
,
nullptr
,
false
));
doorman
->
add_to_loop
();
server
.
mpx
.
handle_internal_events
();
CAF_CHECK_EQUAL
(
server
.
mpx
.
num_socket_handlers
(),
2u
);
CAF_MESSAGE
(
"connect to server (add scribe to client)"
);
auto
scribe
=
unbox
(
client
.
mpx
.
new_tcp_scribe
(
"localhost"
,
doorman
->
port
()));
CAF_CHECK_EQUAL
(
client
.
mpx
.
num_socket_handlers
(),
1u
);
scribe
->
add_to_loop
();
client
.
mpx
.
handle_internal_events
();
CAF_CHECK_EQUAL
(
client
.
mpx
.
num_socket_handlers
(),
2u
);
CAF_MESSAGE
(
"trigger I/O failure in scribe"
);
scribe
->
io_failure
(
&
client
.
mpx
,
io
::
network
::
operation
::
propagate_error
);
client
.
mpx
.
handle_internal_events
();
CAF_CHECK_EQUAL
(
client
.
mpx
.
num_socket_handlers
(),
1u
);
CAF_MESSAGE
(
"trigger I/O failure in doorman"
);
doorman
->
io_failure
(
&
server
.
mpx
,
io
::
network
::
operation
::
propagate_error
);
server
.
mpx
.
handle_internal_events
();
CAF_CHECK_EQUAL
(
server
.
mpx
.
num_socket_handlers
(),
1u
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_io/test/io/remote_actor.cpp
View file @
fbdb3351
...
...
@@ -16,10 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#
include "caf/config.hpp"
#
define CAF_SUITE io.remote_actor
#define CAF_SUITE io_dynamic_remote_actor
#include "caf/test/dsl.hpp"
#include "caf/test/io_dsl.hpp"
#include <algorithm>
#include <sstream>
...
...
@@ -33,89 +32,55 @@ using namespace caf;
namespace
{
constexpr
char
local_host
[]
=
"127.0.0.1"
;
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"
)
>
;
class
config
:
public
actor_system_config
{
public:
config
()
{
load
<
io
::
middleman
>
();
add_message_type
<
std
::
vector
<
int
>>
(
"std::vector<int>"
);
if
(
auto
err
=
parse
(
test
::
engine
::
argc
(),
test
::
engine
::
argv
()))
CAF_FAIL
(
"failed to parse config: "
<<
to_string
(
err
));
}
struct
suite_state
{
int
pings
=
0
;
int
pongs
=
0
;
error
linking_result
;
suite_state
()
=
default
;
};
struct
fixture
{
// State for the server.
config
server_side_config
;
actor_system
server_side
;
io
::
middleman
&
server_side_mm
;
// State for the client.
config
client_side_config
;
actor_system
client_side
;
io
::
middleman
&
client_side_mm
;
fixture
()
:
server_side
(
server_side_config
),
server_side_mm
(
server_side
.
middleman
()),
client_side
(
client_side_config
),
client_side_mm
(
client_side
.
middleman
())
{
// nop
}
};
using
suite_state_ptr
=
std
::
shared_ptr
<
suite_state
>
;
behavior
make_pong_behavior
(
)
{
behavior
ping
(
event_based_actor
*
self
,
suite_state_ptr
ssp
)
{
return
{
[](
int
val
)
->
int
{
++
val
;
CAF_MESSAGE
(
"pong with "
<<
val
);
return
val
;
},
};
}
behavior
make_ping_behavior
(
event_based_actor
*
self
,
const
actor
&
pong
)
{
CAF_MESSAGE
(
"ping with "
<<
0
);
self
->
send
(
pong
,
0
);
return
{[
=
](
int
val
)
->
int
{
if
(
val
==
3
)
{
CAF_MESSAGE
(
"ping with exit"
);
self
->
send_exit
(
self
->
current_sender
(),
exit_reason
::
user_shutdown
);
CAF_MESSAGE
(
"ping quits"
);
[
=
](
kickoff_atom
,
const
actor
&
pong
)
{
CAF_MESSAGE
(
"received `kickoff_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"
);
}
CAF_MESSAGE
(
"ping with "
<<
val
);
return
val
;
}};
}
behavior
make_sort_behavior
()
{
return
{
[](
std
::
vector
<
int
>&
vec
)
->
std
::
vector
<
int
>
{
CAF_MESSAGE
(
"sorter received: "
<<
deep_to_string
(
vec
));
std
::
sort
(
vec
.
begin
(),
vec
.
end
());
CAF_MESSAGE
(
"sorter sent: "
<<
deep_to_string
(
vec
));
return
std
::
move
(
vec
);
});
},
};
}
behavior
make_sort_requester_behavior
(
event_based_actor
*
self
,
const
actor
&
sorter
)
{
self
->
send
(
sorter
,
std
::
vector
<
int
>
{
5
,
4
,
3
,
2
,
1
});
behavior
pong
(
event_based_actor
*
self
,
suite_state_ptr
ssp
)
{
return
{
[
=
](
const
std
::
vector
<
int
>&
vec
)
{
CAF_MESSAGE
(
"sort requester received: "
<<
deep_to_string
(
vec
));
std
::
vector
<
int
>
expected_vec
{
1
,
2
,
3
,
4
,
5
};
CAF_CHECK_EQUAL
(
vec
,
expected_vec
);
self
->
send_exit
(
sorter
,
exit_reason
::
user_shutdown
);
[
=
](
ping_atom
)
->
atom_value
{
CAF_MESSAGE
(
"pong: received ping"
);
if
(
++
ssp
->
pongs
==
10
)
{
self
->
quit
();
CAF_MESSAGE
(
"pong is done"
);
}
return
pong_atom
::
value
;
},
};
}
behavior
fragile_mirror
(
event_based_actor
*
self
)
{
using
fragile_mirror_actor
=
typed_actor
<
replies_to
<
int
>::
with
<
int
>>
;
fragile_mirror_actor
::
behavior_type
fragile_mirror
(
fragile_mirror_actor
::
pointer
self
)
{
return
{
[
=
](
int
i
)
{
self
->
quit
(
exit_reason
::
user_shutdown
);
...
...
@@ -124,67 +89,60 @@ behavior fragile_mirror(event_based_actor* self) {
};
}
behavior
linking_actor
(
event_based_actor
*
self
,
const
actor
&
buddy
)
{
behavior
linking_actor
(
event_based_actor
*
self
,
const
fragile_mirror_actor
&
buddy
,
suite_state_ptr
ssp
)
{
CAF_MESSAGE
(
"link to mirror and send dummy message"
);
self
->
link_to
(
buddy
);
self
->
send
(
buddy
,
42
);
self
->
link_to
(
buddy
);
self
->
set_exit_handler
([
=
](
exit_msg
&
msg
)
{
// Record exit reason for checking it later.
ssp
->
linking_result
=
msg
.
reason
;
self
->
quit
(
std
::
move
(
msg
.
reason
));
});
return
{
[](
int
i
)
{
CAF_CHECK_EQUAL
(
i
,
42
);
},
};
}
struct
fixture
:
point_to_point_fixture
<>
{
fixture
()
{
prepare_connection
(
mars
,
earth
,
"mars"
,
8080
);
ssp
=
std
::
make_shared
<
suite_state
>
();
}
suite_state_ptr
ssp
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
dynamic_remote_actor_tests
,
fixture
)
CAF_TEST
(
identity_semantics
)
{
// server side
auto
server
=
server_side
.
spawn
(
make_pong_behavior
);
auto
port1
=
unbox
(
server_side_mm
.
publish
(
server
,
0
,
local_host
));
auto
port2
=
unbox
(
server_side_mm
.
publish
(
server
,
0
,
local_host
));
CAF_REQUIRE_NOT_EQUAL
(
port1
,
port2
);
auto
same_server
=
unbox
(
server_side_mm
.
remote_actor
(
local_host
,
port2
));
auto
server
=
mars
.
sys
.
spawn
(
pong
,
ssp
);
auto
port
=
mars
.
publish
(
server
,
8080
);
CAF_CHECK_EQUAL
(
port
,
8080u
);
auto
same_server
=
earth
.
remote_actor
(
"mars"
,
8080
);
CAF_REQUIRE_EQUAL
(
same_server
,
server
);
CAF_CHECK_EQUAL
(
same_server
->
node
(),
server_side
.
node
());
auto
server1
=
unbox
(
client_side_mm
.
remote_actor
(
local_host
,
port1
));
auto
server2
=
unbox
(
client_side_mm
.
remote_actor
(
local_host
,
port2
));
CAF_CHECK_EQUAL
(
server1
,
client_side_mm
.
remote_actor
(
local_host
,
port1
));
CAF_CHECK_EQUAL
(
server2
,
client_side_mm
.
remote_actor
(
local_host
,
port2
));
anon_send_exit
(
server
,
exit_reason
::
user_shutdown
);
}
CAF_TEST
(
ping_pong
)
{
// server side
auto
port
=
unbox
(
server_side_mm
.
publish
(
server_side
.
spawn
(
make_pong_behavior
),
0
,
local_host
));
// client side
auto
pong
=
unbox
(
client_side_mm
.
remote_actor
(
local_host
,
port
));
client_side
.
spawn
(
make_ping_behavior
,
pong
);
}
CAF_TEST
(
custom_message_type
)
{
// server side
auto
port
=
unbox
(
server_side_mm
.
publish
(
server_side
.
spawn
(
make_sort_behavior
),
0
,
local_host
));
// client side
auto
sorter
=
unbox
(
client_side_mm
.
remote_actor
(
local_host
,
port
));
client_side
.
spawn
(
make_sort_requester_behavior
,
sorter
);
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
);
run
();
CAF_CHECK_EQUAL
(
ssp
->
pings
,
10
);
CAF_CHECK_EQUAL
(
ssp
->
pongs
,
10
);
}
CAF_TEST
(
remote_link
)
{
// server side
auto
port
=
unbox
(
server_side_mm
.
publish
(
server_side
.
spawn
(
fragile_mirror
),
0
,
local_host
));
// client side
auto
mirror
=
unbox
(
client_side_mm
.
remote_actor
(
local_host
,
port
));
auto
linker
=
client_side
.
spawn
(
linking_actor
,
mirror
);
scoped_actor
self
{
client_side
};
self
->
wait_for
(
linker
);
CAF_MESSAGE
(
"linker exited"
);
self
->
wait_for
(
mirror
);
CAF_MESSAGE
(
"mirror exited"
);
auto
port
=
mars
.
publish
(
mars
.
sys
.
spawn
(
fragile_mirror
),
8080
);
CAF_CHECK_EQUAL
(
port
,
8080u
);
auto
mirror
=
earth
.
remote_actor
<
fragile_mirror_actor
>
(
"mars"
,
8080
);
earth
.
sys
.
spawn
(
linking_actor
,
mirror
,
ssp
);
run
();
CAF_CHECK_EQUAL
(
ssp
->
linking_result
,
exit_reason
::
user_shutdown
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_io/test/io/remote_spawn.cpp
View file @
fbdb3351
...
...
@@ -16,10 +16,11 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE io.remote_spawn
#include "caf/config.hpp"
#define CAF_SUITE io_remote_spawn
#include "caf/test/dsl.hpp"
#include "caf/test/io_dsl.hpp"
#include <cstring>
#include <functional>
...
...
@@ -72,8 +73,7 @@ calculator::behavior_type typed_calculator_fun() {
}
struct
config
:
actor_system_config
{
config
(
int
argc
,
char
**
argv
)
{
parse
(
argc
,
argv
);
config
()
{
load
<
io
::
middleman
>
();
add_actor_type
<
calculator_class
>
(
"calculator-class"
);
add_actor_type
(
"calculator"
,
calculator_fun
);
...
...
@@ -81,48 +81,48 @@ struct config : actor_system_config {
}
};
void
run_client
(
int
argc
,
char
**
argv
,
uint16_t
port
)
{
config
cfg
{
argc
,
argv
};
actor_system
sys
{
cfg
};
scoped_actor
self
{
sys
};
auto
&
mm
=
sys
.
middleman
();
auto
nid
=
mm
.
connect
(
"localhost"
,
port
);
CAF_REQUIRE
(
nid
);
CAF_REQUIRE_NOT_EQUAL
(
sys
.
node
(),
*
nid
);
auto
calc
=
mm
.
remote_spawn
<
calculator
>
(
*
nid
,
"calculator"
,
make_message
());
CAF_REQUIRE
(
!
calc
);
CAF_REQUIRE_EQUAL
(
calc
.
error
().
category
(),
atom
(
"system"
));
CAF_REQUIRE_EQUAL
(
static_cast
<
sec
>
(
calc
.
error
().
code
()),
sec
::
unexpected_actor_messaging_interface
);
calc
=
mm
.
remote_spawn
<
calculator
>
(
*
nid
,
"typed_calculator"
,
make_message
());
CAF_REQUIRE
(
calc
);
auto
f1
=
make_function_view
(
*
calc
);
CAF_REQUIRE_EQUAL
(
f1
(
add_atom
::
value
,
10
,
20
),
30
);
CAF_REQUIRE_EQUAL
(
f1
(
sub_atom
::
value
,
10
,
20
),
-
10
);
f1
.
reset
();
anon_send_exit
(
*
calc
,
exit_reason
::
kill
);
auto
dyn_calc
=
unbox
(
mm
.
remote_spawn
<
actor
>
(
*
nid
,
"calculator-class"
,
make_message
()));
CAF_REQUIRE
(
dyn_calc
);
self
->
request
(
dyn_calc
,
infinite
,
add_atom
::
value
,
10
,
20
)
.
receive
([](
int
result
)
{
CAF_CHECK_EQUAL
(
result
,
30
);
},
[
&
](
const
error
&
err
)
{
CAF_FAIL
(
"error: "
<<
sys
.
render
(
err
));
});
anon_send_exit
(
dyn_calc
,
exit_reason
::
kill
);
mm
.
close
(
port
);
}
void
run_server
(
int
argc
,
char
**
argv
)
{
config
cfg
{
argc
,
argv
};
actor_system
system
{
cfg
};
auto
port
=
unbox
(
system
.
middleman
().
open
(
0
));
std
::
thread
child
{[
=
]
{
run_client
(
argc
,
argv
,
port
);
}};
child
.
join
();
}
struct
fixture
:
point_to_point_fixture
<
test_coordinator_fixture
<
config
>>
{
fixture
()
{
prepare_connection
(
mars
,
earth
,
"mars"
,
8080
);
// ssp = std::make_shared<suite_state>();
}
};
}
// namespace
CAF_TEST
(
remote_spawn
)
{
auto
argc
=
test
::
engine
::
argc
();
auto
argv
=
test
::
engine
::
argv
();
run_server
(
argc
,
argv
);
CAF_TEST_FIXTURE_SCOPE
(
dynamic_remote_actor_tests
,
fixture
)
CAF_TEST
(
nodes
can
spawn
actors
remotely
)
{
loop_after_next_enqueue
(
mars
);
CAF_CHECK_EQUAL
(
unbox
(
mars
.
mm
.
open
(
8080
)),
8080
);
loop_after_next_enqueue
(
earth
);
auto
nid
=
unbox
(
earth
.
mm
.
connect
(
"mars"
,
8080
));
CAF_REQUIRE_EQUAL
(
nid
,
mars
.
sys
.
node
());
CAF_MESSAGE
(
"remote_spawn perform type checks on the handle"
);
loop_after_next_enqueue
(
earth
);
auto
calc
=
earth
.
mm
.
remote_spawn
<
calculator
>
(
nid
,
"calculator"
,
make_message
());
CAF_REQUIRE_EQUAL
(
calc
,
sec
::
unexpected_actor_messaging_interface
);
loop_after_next_enqueue
(
earth
);
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
);
run
();
expect_on
(
earth
,
(
int
),
from
(
*
calc
).
to
(
earth
.
self
).
with
(
30
));
earth
.
self
->
send
(
*
calc
,
sub_atom
::
value
,
10
,
20
);
run
();
expect_on
(
earth
,
(
int
),
from
(
*
calc
).
to
(
earth
.
self
).
with
(
-
10
));
anon_send_exit
(
*
calc
,
exit_reason
::
user_shutdown
);
CAF_MESSAGE
(
"remote_spawn works with class-based actors as well"
);
loop_after_next_enqueue
(
earth
);
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
);
run
();
expect_on
(
earth
,
(
int
),
from
(
*
dyn_calc
).
to
(
earth
.
self
).
with
(
30
));
anon_send_exit
(
*
dyn_calc
,
exit_reason
::
user_shutdown
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_io/test/io/triggering.cpp
deleted
100644 → 0
View file @
c07f3664
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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.triggering
#include "caf/io/all.hpp"
#include "caf/test/unit_test.hpp"
#include <iostream>
#include <memory>
#include "caf/all.hpp"
using
namespace
std
;
using
namespace
caf
;
using
namespace
caf
::
io
;
namespace
{
// -- client implementation, used for both test servers ------------------------
behavior
client
(
broker
*
self
,
connection_handle
hdl
)
{
std
::
vector
<
char
>
buf
;
buf
.
resize
(
200
);
std
::
iota
(
buf
.
begin
(),
buf
.
end
(),
0
);
self
->
write
(
hdl
,
buf
.
size
(),
buf
.
data
());
CAF_REQUIRE_EQUAL
(
self
->
wr_buf
(
hdl
).
size
(),
200u
);
self
->
configure_read
(
hdl
,
receive_policy
::
at_least
(
1
));
self
->
flush
(
hdl
);
return
{
[
=
](
const
new_data_msg
&
)
{
CAF_FAIL
(
"server unexpectedly sent data"
);
},
[
=
](
const
connection_closed_msg
&
)
{
self
->
quit
();
},
};
}
// -- first test server --------------------------------------------------------
struct
server1_state
{
size_t
received
=
0
;
connection_handle
peer
=
invalid_connection_handle
;
};
// consumes 5 more tokens, then waits for passivated message to shutdown
behavior
server1_stage4
(
stateful_actor
<
server1_state
,
broker
>*
self
)
{
CAF_MESSAGE
(
"enter server stage 4"
);
self
->
trigger
(
self
->
state
.
peer
,
5
);
return
{
[
=
](
const
new_data_msg
&
dm
)
{
CAF_REQUIRE_EQUAL
(
dm
.
buf
.
size
(),
10u
);
self
->
state
.
received
+=
1
;
},
[
=
](
const
connection_passivated_msg
&
cp
)
{
CAF_REQUIRE_EQUAL
(
cp
.
handle
,
self
->
state
.
peer
);
CAF_REQUIRE_EQUAL
(
self
->
state
.
received
,
15u
);
CAF_REQUIRE_NOT_EQUAL
(
self
->
state
.
peer
,
invalid_connection_handle
);
// delay new tokens to force MM to remove this broker from its loop
CAF_MESSAGE
(
"server is done"
);
self
->
quit
();
},
};
}
// consumes 5 more tokens, then waits for passivated message to send itself
// a message that generates 5 more (force MM to actually remove this broker
// from its event loop and then re-adding it)
behavior
server1_stage3
(
stateful_actor
<
server1_state
,
broker
>*
self
)
{
CAF_MESSAGE
(
"enter server stage 3"
);
self
->
trigger
(
self
->
state
.
peer
,
5
);
return
{
[
=
](
const
new_data_msg
&
dm
)
{
CAF_REQUIRE_EQUAL
(
dm
.
buf
.
size
(),
10u
);
self
->
state
.
received
+=
1
;
},
[
=
](
const
connection_passivated_msg
&
cp
)
{
CAF_REQUIRE_EQUAL
(
cp
.
handle
,
self
->
state
.
peer
);
CAF_REQUIRE_EQUAL
(
self
->
state
.
received
,
10u
);
CAF_REQUIRE_NOT_EQUAL
(
self
->
state
.
peer
,
invalid_connection_handle
);
// delay new tokens to force MM to remove this broker from its loop
self
->
send
(
self
,
ok_atom
::
value
);
},
[
=
](
ok_atom
)
{
self
->
become
(
server1_stage4
(
self
));
},
};
}
// consumes 5 tokens, then waits for passivated message and generates 5 more
behavior
server1_stage2
(
stateful_actor
<
server1_state
,
broker
>*
self
)
{
CAF_MESSAGE
(
"enter server stage 2"
);
self
->
trigger
(
self
->
state
.
peer
,
5
);
return
{
[
=
](
const
new_data_msg
&
dm
)
{
CAF_REQUIRE_EQUAL
(
dm
.
buf
.
size
(),
10u
);
self
->
state
.
received
+=
1
;
},
[
=
](
const
connection_passivated_msg
&
cp
)
{
CAF_REQUIRE_EQUAL
(
cp
.
handle
,
self
->
state
.
peer
);
CAF_REQUIRE_EQUAL
(
self
->
state
.
received
,
5u
);
CAF_REQUIRE_NOT_EQUAL
(
self
->
state
.
peer
,
invalid_connection_handle
);
self
->
become
(
server1_stage3
(
self
));
},
};
}
// waits for the connection to the client
behavior
server1
(
stateful_actor
<
server1_state
,
broker
>*
self
)
{
return
{
[
=
](
const
new_connection_msg
&
nc
)
{
CAF_REQUIRE_EQUAL
(
self
->
state
.
peer
,
invalid_connection_handle
);
self
->
state
.
peer
=
nc
.
handle
;
self
->
configure_read
(
nc
.
handle
,
receive_policy
::
exactly
(
10
));
self
->
become
(
server1_stage2
(
self
));
},
};
}
// -- second test server -------------------------------------------------------
struct
server2_state
{
size_t
accepted
=
0
;
};
// consumes 5 more tokens, then waits for passivated message to shutdown
behavior
server2_stage4
(
stateful_actor
<
server2_state
,
broker
>*
self
)
{
CAF_MESSAGE
(
"enter server stage 4"
);
return
{
[
=
](
const
new_connection_msg
&
)
{
self
->
state
.
accepted
+=
1
;
},
[
=
](
const
acceptor_passivated_msg
&
)
{
CAF_REQUIRE_EQUAL
(
self
->
state
.
accepted
,
16u
);
CAF_MESSAGE
(
"server is done"
);
self
->
quit
();
},
};
}
// consumes 5 more tokens, then waits for passivated message to send itself
// a message that generates 5 more (force MM to actually remove this broker
// from its event loop and then re-adding it)
behavior
server2_stage3
(
stateful_actor
<
server2_state
,
broker
>*
self
)
{
CAF_MESSAGE
(
"enter server stage 3"
);
return
{
[
=
](
const
new_connection_msg
&
)
{
self
->
state
.
accepted
+=
1
;
},
[
=
](
const
acceptor_passivated_msg
&
cp
)
{
CAF_REQUIRE_EQUAL
(
self
->
state
.
accepted
,
11u
);
// delay new tokens to force MM to remove this broker from its loop
self
->
send
(
self
,
ok_atom
::
value
,
cp
.
handle
);
},
[
=
](
ok_atom
,
accept_handle
hdl
)
{
self
->
trigger
(
hdl
,
5
);
self
->
become
(
server2_stage4
(
self
));
},
};
}
// consumes 5 tokens, then waits for passivated message and generates 5 more
behavior
server2_stage2
(
stateful_actor
<
server2_state
,
broker
>*
self
)
{
CAF_MESSAGE
(
"enter server stage 2"
);
CAF_REQUIRE_EQUAL
(
self
->
state
.
accepted
,
1u
);
return
{
[
=
](
const
new_connection_msg
&
)
{
self
->
state
.
accepted
+=
1
;
},
[
=
](
const
acceptor_passivated_msg
&
cp
)
{
CAF_REQUIRE_EQUAL
(
self
->
state
.
accepted
,
6u
);
self
->
trigger
(
cp
.
handle
,
5
);
self
->
become
(
server2_stage3
(
self
));
},
};
}
// waits for the connection to the client
behavior
server2
(
stateful_actor
<
server2_state
,
broker
>*
self
)
{
return
{
[
=
](
const
new_connection_msg
&
nc
)
{
self
->
state
.
accepted
+=
1
;
self
->
trigger
(
nc
.
source
,
5
);
self
->
become
(
server2_stage2
(
self
));
},
};
}
// -- config and fixture -------------------------------------------------------
struct
config
:
actor_system_config
{
config
()
{
auto
argc
=
test
::
engine
::
argc
();
auto
argv
=
test
::
engine
::
argv
();
load
<
io
::
middleman
>
();
parse
(
argc
,
argv
);
}
};
struct
fixture
{
config
client_cfg
;
actor_system
client_system
;
config
server_cfg
;
actor_system
server_system
;
fixture
()
:
client_system
(
client_cfg
),
server_system
(
server_cfg
)
{
// nop
}
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
trigger_tests
,
fixture
)
CAF_TEST
(
trigger_connection
)
{
CAF_MESSAGE
(
"spawn server"
);
uint16_t
port
=
0
;
auto
serv
=
server_system
.
middleman
().
spawn_server
(
server1
,
port
);
CAF_REQUIRE
(
serv
);
CAF_REQUIRE_NOT_EQUAL
(
port
,
0
);
CAF_MESSAGE
(
"server spawned at port "
<<
port
);
std
::
thread
child
{[
&
]
{
auto
cl
=
client_system
.
middleman
().
spawn_client
(
client
,
"localhost"
,
port
);
CAF_REQUIRE
(
cl
);
}};
child
.
join
();
}
CAF_TEST
(
trigger_acceptor
)
{
CAF_MESSAGE
(
"spawn server"
);
uint16_t
port
=
0
;
auto
serv
=
server_system
.
middleman
().
spawn_server
(
server2
,
port
);
CAF_REQUIRE
(
serv
);
CAF_REQUIRE_NOT_EQUAL
(
port
,
0
);
CAF_MESSAGE
(
"server spawned at port "
<<
port
);
std
::
thread
child
{[
&
]
{
// 16 clients will succeed to connect
for
(
int
i
=
0
;
i
<
16
;
++
i
)
{
auto
cl
=
client_system
.
middleman
().
spawn_client
(
client
,
"localhost"
,
port
);
CAF_REQUIRE
(
cl
);
}
}};
child
.
join
();
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_io/test/io/unpublish.cpp
View file @
fbdb3351
...
...
@@ -16,12 +16,14 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE io.unpublish
#include "caf/config.hpp"
#define CAF_SUITE io_unpublish
#include "caf/test/dsl.hpp"
#include "caf/test/io_dsl.hpp"
#include <atomic>
#include <memory>
#include <new>
#include <thread>
...
...
@@ -32,16 +34,22 @@ using namespace caf;
namespace
{
std
::
atomic
<
long
>
s_dtor_called
;
struct
suite_state
{
long
dtors_called
=
0
;
suite_state
()
=
default
;
};
using
suite_state_ptr
=
std
::
shared_ptr
<
suite_state
>
;
class
dummy
:
public
event_based_actor
{
public:
dummy
(
actor_config
&
cfg
)
:
event_based_actor
(
cfg
)
{
dummy
(
actor_config
&
cfg
,
suite_state_ptr
ssp
)
:
event_based_actor
(
cfg
),
ssp_
(
std
::
move
(
ssp
))
{
// nop
}
~
dummy
()
override
{
++
s_dtor_called
;
ssp_
->
dtors_called
+=
1
;
}
behavior
make_behavior
()
override
{
...
...
@@ -49,85 +57,54 @@ public:
// nop
}};
}
};
struct
config
:
actor_system_config
{
config
()
{
load
<
io
::
middleman
>
();
if
(
auto
err
=
parse
(
test
::
engine
::
argc
(),
test
::
engine
::
argv
()))
CAF_FAIL
(
"failed to parse config: "
<<
to_string
(
err
));
}
private:
suite_state_ptr
ssp_
;
};
struct
fixture
{
struct
fixture
:
point_to_point_fixture
<>
{
fixture
()
{
new
(
&
system
)
actor_system
(
cfg
);
testee
=
system
.
spawn
<
dummy
>
();
prepare_connection
(
mars
,
earth
,
"mars"
,
8080
);
ssp
=
std
::
make_shared
<
suite_state
>
();
}
~
fixture
()
{
anon_send_exit
(
testee
,
exit_reason
::
user_shutdown
);
destroy
(
testee
);
system
.
~
actor_system
();
CAF_CHECK_EQUAL
(
s_dtor_called
.
load
(),
2
);
run
();
CAF_CHECK_EQUAL
(
ssp
->
dtors_called
,
2
);
}
actor
remote_actor
(
const
char
*
hostname
,
uint16_t
port
,
bool
expect_fail
=
false
)
{
actor
result
;
scoped_actor
self
{
system
,
true
};
self
->
request
(
system
.
middleman
().
actor_handle
(),
infinite
,
connect_atom
::
value
,
hostname
,
port
)
.
receive
(
[
&
](
node_id
&
,
strong_actor_ptr
&
res
,
std
::
set
<
std
::
string
>&
xs
)
{
CAF_REQUIRE
(
xs
.
empty
());
if
(
res
)
result
=
actor_cast
<
actor
>
(
std
::
move
(
res
));
},
[
&
](
error
&
)
{
// nop
});
if
(
expect_fail
)
CAF_REQUIRE
(
!
result
);
else
CAF_REQUIRE
(
result
);
return
result
;
}
config
cfg
;
union
{
actor_system
system
;
};
// manually control ctor/dtor
actor
testee
;
suite_state_ptr
ssp
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
unpublish_tests
,
fixture
)
CAF_TEST
(
unpublishing
)
{
auto
port
=
unbox
(
system
.
middleman
().
publish
(
testee
,
0
));
CAF_REQUIRE
(
port
!=
0
);
CAF_MESSAGE
(
"published actor on port "
<<
port
);
CAF_MESSAGE
(
"test invalid unpublish"
);
auto
testee2
=
system
.
spawn
<
dummy
>
();
system
.
middleman
().
unpublish
(
testee2
,
port
);
auto
x0
=
remote_actor
(
"127.0.0.1"
,
port
);
CAF_CHECK_NOT_EQUAL
(
x0
,
testee2
);
CAF_CHECK_EQUAL
(
x0
,
testee
);
anon_send_exit
(
testee2
,
exit_reason
::
kill
);
CAF_MESSAGE
(
"unpublish testee"
);
system
.
middleman
().
unpublish
(
testee
,
port
);
CAF_MESSAGE
(
"check whether testee is still available via cache"
);
auto
x1
=
remote_actor
(
"127.0.0.1"
,
port
);
CAF_CHECK_EQUAL
(
x1
,
testee
);
CAF_MESSAGE
(
"fake death of testee and check if testee becomes unavailable"
);
anon_send
(
actor_cast
<
actor
>
(
system
.
middleman
().
actor_handle
()),
down_msg
{
testee
.
address
(),
exit_reason
::
normal
});
// must fail now
auto
x2
=
remote_actor
(
"127.0.0.1"
,
port
,
true
);
CAF_CHECK
(
!
x2
);
CAF_TEST
(
actors
can
become
unpublished
)
{
auto
testee
=
mars
.
sys
.
spawn
<
dummy
>
(
ssp
);
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
// The MM holds a reference to this actor publishing it, so we need to kill
// it manually.
anon_send_exit
(
testee
,
exit_reason
::
user_shutdown
);
});
loop_after_next_enqueue
(
mars
);
auto
port
=
unbox
(
mars
.
mm
.
publish
(
testee
,
8080
));
CAF_REQUIRE_EQUAL
(
port
,
8080
);
CAF_MESSAGE
(
"the middleman ignores invalid unpublish() calls"
);
auto
testee2
=
mars
.
sys
.
spawn
<
dummy
>
(
ssp
);
loop_after_next_enqueue
(
mars
);
auto
res
=
mars
.
mm
.
unpublish
(
testee2
,
8080
);
CAF_CHECK
(
!
res
&&
res
.
error
()
==
sec
::
no_actor_published_at_port
);
anon_send_exit
(
testee2
,
exit_reason
::
user_shutdown
);
CAF_MESSAGE
(
"after unpublishing an actor, remotes can no longer connect"
);
loop_after_next_enqueue
(
mars
);
CAF_CHECK
(
mars
.
mm
.
unpublish
(
testee
,
8080
));
// TODO: ideally, we'd check that remote actors in fact can no longer connect.
// However, the test multiplexer does not support "closing" connections
// and the remote_actor blocks forever.
// run();
// loop_after_next_enqueue(earth);
// CAF_CHECK(!earth.mm.remote_actor("mars", 8080));
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_test/caf/test/io_dsl.hpp
View file @
fbdb3351
...
...
@@ -25,6 +25,15 @@
#include "caf/test/dsl.hpp"
class
test_node_fixture_config
:
public
caf
::
actor_system_config
{
public:
test_node_fixture_config
()
{
load
<
caf
::
io
::
middleman
>
();
}
};
using
io_base_fixture
=
test_coordinator_fixture
<
test_node_fixture_config
>
;
/// Ensures that `test_node_fixture` can override `run_exhaustively` even if
/// the base fixture does not declare these member functions virtual.
template
<
class
BaseFixture
>
...
...
@@ -46,10 +55,9 @@ public:
};
/// A fixture containing all required state to simulate a single CAF node.
template
<
class
BaseFixture
=
test_coordinator_fixture
<
caf
::
actor_system_config
>
>
template
<
class
BaseFixture
=
io_base_fixture
>
class
test_node_fixture
:
public
BaseFixture
,
test_node_fixture_base
<
BaseFixture
>
{
public
test_node_fixture_base
<
BaseFixture
>
{
public:
// -- member types -----------------------------------------------------------
...
...
@@ -224,8 +232,7 @@ private:
/// A simple fixture that includes two nodes (`earth` and `mars`) that can
/// connect to each other.
template
<
class
BaseFixture
=
test_coordinator_fixture
<
caf
::
actor_system_config
>
>
template
<
class
BaseFixture
=
io_base_fixture
>
class
point_to_point_fixture
:
public
test_network_fixture_base
<
test_node_fixture
<
BaseFixture
>>
{
public:
...
...
@@ -243,12 +250,19 @@ public:
// Run initialization code.
this
->
exec_all
();
}
~
point_to_point_fixture
()
{
run
();
}
void
run
()
{
this
->
exec_all
();
}
};
/// A simple fixture that includes three nodes (`earth`, `mars`, and `jupiter`)
/// that can connect to each other.
template
<
class
BaseFixture
=
test_coordinator_fixture
<
caf
::
actor_system_config
>
>
template
<
class
BaseFixture
=
io_base_fixture
>
class
belt_fixture
:
public
test_network_fixture_base
<
test_node_fixture
<
BaseFixture
>>
{
public:
...
...
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