Unverified Commit 4d4c1fdb authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #965

Remove localhost connections from I/O unit tests
parents c07f3664 d5842e84
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
* http://www.boost.org/LICENSE_1_0.txt. * * 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" #include "caf/io/basp/message_queue.hpp"
......
...@@ -20,7 +20,7 @@ ...@@ -20,7 +20,7 @@
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/test/dsl.hpp" #include "caf/test/io_dsl.hpp"
#include <iostream> #include <iostream>
#include <memory> #include <memory>
...@@ -28,7 +28,6 @@ ...@@ -28,7 +28,6 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/io/all.hpp" #include "caf/io/all.hpp"
using namespace std;
using namespace caf; using namespace caf;
using namespace caf::io; using namespace caf::io;
...@@ -39,40 +38,41 @@ using pong_atom = caf::atom_constant<caf::atom("pong")>; ...@@ -39,40 +38,41 @@ using pong_atom = caf::atom_constant<caf::atom("pong")>;
using publish_atom = caf::atom_constant<caf::atom("publish")>; using publish_atom = caf::atom_constant<caf::atom("publish")>;
using kickoff_atom = caf::atom_constant<caf::atom("kickoff")>; using kickoff_atom = caf::atom_constant<caf::atom("kickoff")>;
behavior ping(event_based_actor* self, size_t num_pings) { struct suite_state {
CAF_MESSAGE("num_pings: " << num_pings); int pings = 0;
auto count = std::make_shared<size_t>(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 { return {
[=](kickoff_atom, const actor& pong) { [=](kickoff_atom, const actor& pong) {
CAF_MESSAGE("received `kickoff_atom`"); CAF_MESSAGE("received `kickoff_atom`");
self->send(pong, ping_atom::value, 1); ++ssp->pings;
self->become([=](pong_atom, int value) -> std::tuple<atom_value, int> { self->send(pong, ping_atom::value);
if (++*count >= num_pings) { self->become([=](pong_atom) {
CAF_MESSAGE("received " << num_pings << " pings, call self->quit"); CAF_MESSAGE("ping: received pong");
self->send(pong, ping_atom::value);
if (++ssp->pings == 10) {
self->quit(); self->quit();
CAF_MESSAGE("ping is done");
} }
return std::make_tuple(ping_atom::value, value + 1);
}); });
}, },
}; };
} }
behavior pong(event_based_actor* self) { behavior pong(event_based_actor* self, suite_state_ptr ssp) {
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);
});
return { return {
[=](ping_atom, int value) -> std::tuple<atom_value, int> { [=](ping_atom) -> atom_value {
CAF_MESSAGE("received `ping_atom`"); CAF_MESSAGE("pong: received ping");
self->monitor(self->current_sender()); if (++ssp->pongs == 10) {
// set next behavior self->quit();
self->become([](ping_atom, int val) { CAF_MESSAGE("pong is done");
return std::make_tuple(pong_atom::value, val); }
}); return pong_atom::value;
// reply to 'ping'
return std::make_tuple(pong_atom::value, value);
}, },
}; };
} }
...@@ -82,23 +82,19 @@ behavior peer_fun(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -82,23 +82,19 @@ behavior peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
CAF_REQUIRE(self->subtype() == resumable::io_actor); CAF_REQUIRE(self->subtype() == resumable::io_actor);
CAF_CHECK(self != nullptr); CAF_CHECK(self != nullptr);
self->monitor(buddy); 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); CAF_REQUIRE(self->connections().size() == 1);
self->configure_read(hdl, receive_policy::exactly(sizeof(atom_value) self->configure_read(hdl, receive_policy::exactly(sizeof(atom_value)));
+ sizeof(int))); auto write = [=](atom_value type) {
auto write = [=](atom_value type, int value) {
auto& buf = self->wr_buf(hdl); auto& buf = self->wr_buf(hdl);
auto first = reinterpret_cast<char*>(&type); auto first = reinterpret_cast<char*>(&type);
buf.insert(buf.end(), first, first + sizeof(atom_value)); 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->flush(hdl);
}; };
self->set_down_handler([=](down_msg& dm) {
CAF_MESSAGE("received: " << to_string(dm));
if (dm.source == buddy)
self->quit(dm.reason);
});
return { return {
[=](const connection_closed_msg&) { [=](const connection_closed_msg&) {
CAF_MESSAGE("received connection_closed_msg"); CAF_MESSAGE("received connection_closed_msg");
...@@ -107,19 +103,11 @@ behavior peer_fun(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -107,19 +103,11 @@ behavior peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
[=](const new_data_msg& msg) { [=](const new_data_msg& msg) {
CAF_MESSAGE("received new_data_msg"); CAF_MESSAGE("received new_data_msg");
atom_value type; atom_value type;
int value;
memcpy(&type, msg.buf.data(), sizeof(atom_value)); memcpy(&type, msg.buf.data(), sizeof(atom_value));
memcpy(&value, msg.buf.data() + sizeof(atom_value), sizeof(int)); self->send(buddy, type);
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);
}, },
[=](ping_atom) { write(ping_atom::value); },
[=](pong_atom) { write(pong_atom::value); },
}; };
} }
...@@ -132,7 +120,7 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) { ...@@ -132,7 +120,7 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) {
self->quit(); self->quit();
}, },
[=](publish_atom) -> expected<uint16_t> { [=](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) if (!res)
return std::move(res.error()); return std::move(res.error());
return res->second; return res->second;
...@@ -140,49 +128,48 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) { ...@@ -140,49 +128,48 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) {
}; };
} }
void run_client(int argc, char** argv, uint16_t port) { using int_peer = connection_handler::extend<replies_to<int>::with<int>>;
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");
}
void run_server(int argc, char** argv) { int_peer::behavior_type int_peer_fun(int_peer::broker_pointer) {
actor_system_config cfg; return {
cfg.load<io::middleman>(); [=](const connection_closed_msg&) {
if (auto err = cfg.parse(argc, argv)) CAF_FAIL("received connection_closed_msg");
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); });
}, },
[&](const error& err) { [=](const new_data_msg&) { CAF_FAIL("received new_data_msg"); },
CAF_ERROR("Error: " << self->system().render(err)); [=](int value) {
}); CAF_MESSAGE("received: " << value);
self->await_all_other_actors_done(); return value;
child.join(); },
};
} }
} // namespace } // namespace
CAF_TEST(test_broker) { CAF_TEST_FIXTURE_SCOPE(broker_tests, point_to_point_fixture<>)
auto argc = test::engine::argc();
auto argv = test::engine::argv(); CAF_TEST(test broker to broker communication) {
run_server(argc, argv); 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()
...@@ -79,27 +79,4 @@ CAF_TEST(doorman io_failure) { ...@@ -79,27 +79,4 @@ CAF_TEST(doorman io_failure) {
CAF_CHECK_EQUAL(server.mpx.num_socket_handlers(), 1u); 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() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -16,10 +16,9 @@ ...@@ -16,10 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. * * 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/io_dsl.hpp"
#include "caf/test/dsl.hpp"
#include <algorithm> #include <algorithm>
#include <sstream> #include <sstream>
...@@ -33,89 +32,55 @@ using namespace caf; ...@@ -33,89 +32,55 @@ using namespace caf;
namespace { 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 { struct suite_state {
public: int pings = 0;
config() { int pongs = 0;
load<io::middleman>(); error linking_result;
add_message_type<std::vector<int>>("std::vector<int>"); suite_state() = default;
if (auto err = parse(test::engine::argc(), test::engine::argv()))
CAF_FAIL("failed to parse config: " << to_string(err));
}
}; };
struct fixture { using suite_state_ptr = std::shared_ptr<suite_state>;
// 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
}
};
behavior make_pong_behavior() { behavior ping(event_based_actor* self, suite_state_ptr ssp) {
return { return {
[](int val) -> int { [=](kickoff_atom, const actor& pong) {
++val; CAF_MESSAGE("received `kickoff_atom`");
CAF_MESSAGE("pong with " << val); ++ssp->pings;
return val; 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) {
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");
self->quit(); 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, behavior pong(event_based_actor* self, suite_state_ptr ssp) {
const actor& sorter) {
self->send(sorter, std::vector<int>{5, 4, 3, 2, 1});
return { return {
[=](const std::vector<int>& vec) { [=](ping_atom) -> atom_value {
CAF_MESSAGE("sort requester received: " << deep_to_string(vec)); CAF_MESSAGE("pong: received ping");
std::vector<int> expected_vec{1, 2, 3, 4, 5}; if (++ssp->pongs == 10) {
CAF_CHECK_EQUAL(vec, expected_vec);
self->send_exit(sorter, exit_reason::user_shutdown);
self->quit(); 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 { return {
[=](int i) { [=](int i) {
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
...@@ -124,67 +89,60 @@ behavior fragile_mirror(event_based_actor* self) { ...@@ -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"); CAF_MESSAGE("link to mirror and send dummy message");
self->link_to(buddy);
self->send(buddy, 42); 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 { return {
[](int i) { CAF_CHECK_EQUAL(i, 42); }, [](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 } // namespace
CAF_TEST_FIXTURE_SCOPE(dynamic_remote_actor_tests, fixture) CAF_TEST_FIXTURE_SCOPE(dynamic_remote_actor_tests, fixture)
CAF_TEST(identity_semantics) { CAF_TEST(identity_semantics) {
// server side auto server = mars.sys.spawn(pong, ssp);
auto server = server_side.spawn(make_pong_behavior); auto port = mars.publish(server, 8080);
auto port1 = unbox(server_side_mm.publish(server, 0, local_host)); CAF_CHECK_EQUAL(port, 8080u);
auto port2 = unbox(server_side_mm.publish(server, 0, local_host)); auto same_server = earth.remote_actor("mars", 8080);
CAF_REQUIRE_NOT_EQUAL(port1, port2);
auto same_server = unbox(server_side_mm.remote_actor(local_host, port2));
CAF_REQUIRE_EQUAL(same_server, server); 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); anon_send_exit(server, exit_reason::user_shutdown);
} }
CAF_TEST(ping_pong) { CAF_TEST(ping_pong) {
// server side auto port = mars.publish(mars.sys.spawn(pong, ssp), 8080);
auto port = unbox( CAF_CHECK_EQUAL(port, 8080u);
server_side_mm.publish(server_side.spawn(make_pong_behavior), 0, auto remote_pong = earth.remote_actor("mars", 8080);
local_host)); anon_send(earth.sys.spawn(ping, ssp), kickoff_atom::value, remote_pong);
// client side run();
auto pong = unbox(client_side_mm.remote_actor(local_host, port)); CAF_CHECK_EQUAL(ssp->pings, 10);
client_side.spawn(make_ping_behavior, pong); CAF_CHECK_EQUAL(ssp->pongs, 10);
}
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);
} }
CAF_TEST(remote_link) { CAF_TEST(remote_link) {
// server side auto port = mars.publish(mars.sys.spawn(fragile_mirror), 8080);
auto port = unbox( CAF_CHECK_EQUAL(port, 8080u);
server_side_mm.publish(server_side.spawn(fragile_mirror), 0, local_host)); auto mirror = earth.remote_actor<fragile_mirror_actor>("mars", 8080);
// client side earth.sys.spawn(linking_actor, mirror, ssp);
auto mirror = unbox(client_side_mm.remote_actor(local_host, port)); run();
auto linker = client_side.spawn(linking_actor, mirror); CAF_CHECK_EQUAL(ssp->linking_result, exit_reason::user_shutdown);
scoped_actor self{client_side};
self->wait_for(linker);
CAF_MESSAGE("linker exited");
self->wait_for(mirror);
CAF_MESSAGE("mirror exited");
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -16,10 +16,11 @@ ...@@ -16,10 +16,11 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#define CAF_SUITE io.remote_spawn
#include "caf/config.hpp" #include "caf/config.hpp"
#define CAF_SUITE io_remote_spawn #include "caf/test/io_dsl.hpp"
#include "caf/test/dsl.hpp"
#include <cstring> #include <cstring>
#include <functional> #include <functional>
...@@ -72,8 +73,7 @@ calculator::behavior_type typed_calculator_fun() { ...@@ -72,8 +73,7 @@ calculator::behavior_type typed_calculator_fun() {
} }
struct config : actor_system_config { struct config : actor_system_config {
config(int argc, char** argv) { config() {
parse(argc, argv);
load<io::middleman>(); load<io::middleman>();
add_actor_type<calculator_class>("calculator-class"); add_actor_type<calculator_class>("calculator-class");
add_actor_type("calculator", calculator_fun); add_actor_type("calculator", calculator_fun);
...@@ -81,48 +81,47 @@ struct config : actor_system_config { ...@@ -81,48 +81,47 @@ struct config : actor_system_config {
} }
}; };
void run_client(int argc, char** argv, uint16_t port) { struct fixture : point_to_point_fixture<test_coordinator_fixture<config>> {
config cfg{argc, argv}; fixture() {
actor_system sys{cfg}; prepare_connection(mars, earth, "mars", 8080);
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();
}
} // namespace } // namespace
CAF_TEST(remote_spawn) { CAF_TEST_FIXTURE_SCOPE(dynamic_remote_actor_tests, fixture)
auto argc = test::engine::argc();
auto argv = test::engine::argv(); CAF_TEST(nodes can spawn actors remotely) {
run_server(argc, argv); 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()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE io_typed_broker
#include "caf/test/dsl.hpp"
#include <iostream>
#include <memory>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/string_algorithms.hpp"
using namespace std;
using namespace caf;
using namespace caf::io;
namespace {
using publish_atom = atom_constant<atom("publish")>;
using ping_atom = caf::atom_constant<atom("ping")>;
using pong_atom = caf::atom_constant<atom("pong")>;
using kickoff_atom = caf::atom_constant<atom("kickoff")>;
using peer = connection_handler::extend<reacts_to<ping_atom, int>,
reacts_to<pong_atom, int>>;
using acceptor = accept_handler::extend<
replies_to<publish_atom>::with<uint16_t>>;
using ping_actor = typed_actor<
replies_to<pong_atom, int>::with<ping_atom, int>>;
using pong_actor = typed_actor<
replies_to<ping_atom, int>::with<pong_atom, int>>;
behavior ping(event_based_actor* self, size_t num_pings) {
CAF_MESSAGE("num_pings: " << num_pings);
auto count = std::make_shared<size_t>(0);
return {
[=](kickoff_atom, const peer& pong) {
CAF_MESSAGE("received `kickoff_atom`");
self->send(pong, ping_atom::value, 1);
self->become([=](pong_atom, int value) -> std::tuple<ping_atom, int> {
if (++*count >= num_pings) {
CAF_MESSAGE("received " << num_pings << " pings, call self->quit");
self->quit();
}
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: " << to_string(dm.reason));
self->quit(dm.reason);
});
return {
[=](ping_atom, int value) -> std::tuple<atom_value, int> {
CAF_MESSAGE("received: 'ping', " << value);
self->monitor(self->current_sender());
// set next behavior
self->become([](ping_atom, int val) {
// CAF_MESSAGE("received: 'ping', " << val);
return std::make_tuple(pong_atom::value, val);
});
// reply to 'ping'
return std::make_tuple(pong_atom::value, value);
},
};
}
peer::behavior_type peer_fun(peer::broker_pointer self, connection_handle hdl,
const actor& buddy) {
CAF_MESSAGE("peer_fun called");
self->monitor(buddy);
// assume exactly one connection
CAF_REQUIRE_EQUAL(self->connections().size(), 1u);
self->configure_read(hdl, receive_policy::exactly(sizeof(atom_value)
+ sizeof(int)));
auto write = [=](atom_value x, int y) {
auto& buf = self->wr_buf(hdl);
binary_serializer sink{self->system(), buf};
auto e = sink(x, y);
CAF_REQUIRE(!e);
self->flush(hdl);
};
self->set_down_handler([=](down_msg& dm) {
CAF_MESSAGE("received down_msg");
if (dm.source == buddy)
self->quit(std::move(dm.reason));
});
return {
[=](const connection_closed_msg&) {
CAF_MESSAGE("received connection_closed_msg");
self->quit();
},
[=](const new_data_msg& msg) {
CAF_MESSAGE("received new_data_msg");
auto x = static_cast<atom_value>(0);
auto y = 0;
binary_deserializer source{self->system(), msg.buf};
auto e = source(x, y);
CAF_REQUIRE(!e);
if (x == pong_atom::value)
self->send(actor_cast<ping_actor>(buddy), pong_atom::value, y);
else
self->send(actor_cast<pong_actor>(buddy), ping_atom::value, y);
},
[=](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);
},
};
}
acceptor::behavior_type acceptor_fun(acceptor::broker_pointer self,
const actor& buddy) {
CAF_MESSAGE("peer_acceptor_fun");
return {
[=](const new_connection_msg& msg) {
CAF_MESSAGE("received `new_connection_msg`");
self->fork(peer_fun, msg.handle, buddy);
self->quit();
},
[](const acceptor_closed_msg&) {
// nop
},
[=](publish_atom) -> expected<uint16_t> {
auto dm = self->add_tcp_doorman(0, "127.0.0.1");
if (dm)
return get<1>(*dm);
return std::move(dm.error());
},
};
}
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_typed...");
auto cl = unbox(
system.middleman().spawn_client(peer_fun, "localhost", port, p));
CAF_MESSAGE("spawn_client_typed finished");
anon_send(p, kickoff_atom::value, cl);
CAF_MESSAGE("`kickoff_atom` has been send");
}
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};
auto serv = system.middleman().spawn_broker(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); });
},
[&](error& err) { CAF_FAIL("error: " << system.render(err)); });
self->await_all_other_actors_done();
CAF_MESSAGE("wait for client system");
child.join();
}
} // namespace
CAF_TEST(test_typed_broker) {
run_server(test::engine::argc(), test::engine::argv());
}
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE io_typed_remote_actor
#include "caf/test/dsl.hpp"
#include <cstring>
#include <functional>
#include <iostream>
#include <sstream>
#include <string>
#include <thread>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using namespace std;
using namespace caf;
struct ping {
int32_t value;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, ping& x) {
return f(meta::type_name("ping"), x.value);
}
bool operator==(const ping& lhs, const ping& rhs) {
return lhs.value == rhs.value;
}
struct pong {
int32_t value;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, pong& x) {
return f(meta::type_name("pong"), x.value);
}
bool operator==(const pong& lhs, const pong& rhs) {
return lhs.value == rhs.value;
}
using server_type = typed_actor<replies_to<ping>::with<pong>>;
using client_type = typed_actor<>;
server_type::behavior_type server() {
return {[](const ping& p) -> pong {
CAF_CHECK_EQUAL(p.value, 42);
return pong{p.value};
}};
}
void run_client(int argc, char** argv, uint16_t port) {
actor_system_config cfg;
cfg.load<io::middleman>()
.add_message_type<ping>("ping")
.add_message_type<pong>("pong");
if (auto err = cfg.parse(argc, argv))
CAF_FAIL("failed to parse config: " << to_string(err));
actor_system sys{cfg};
// check whether invalid_argument is thrown
// when trying to connect to get an untyped
// handle to the server
auto res = sys.middleman().remote_actor("127.0.0.1", port);
CAF_REQUIRE(!res);
CAF_MESSAGE(sys.render(res.error()));
CAF_MESSAGE("connect to typed_remote_actor");
auto serv = unbox(
sys.middleman().remote_actor<server_type>("127.0.0.1", port));
auto f = make_function_view(serv);
CAF_CHECK_EQUAL(f(ping{42}), pong{42});
anon_send_exit(serv, exit_reason::user_shutdown);
}
void run_server(int argc, char** argv) {
actor_system_config cfg;
cfg.load<io::middleman>()
.add_message_type<ping>("ping")
.add_message_type<pong>("pong")
.parse(argc, argv);
actor_system sys{cfg};
auto port = unbox(sys.middleman().publish(sys.spawn(server), 0, "127.0.0.1"));
CAF_REQUIRE(port != 0);
CAF_MESSAGE("running on port " << port << ", start client");
std::thread child{[=] { run_client(argc, argv, port); }};
child.join();
}
CAF_TEST(test_typed_remote_actor) {
auto argc = test::engine::argc();
auto argv = test::engine::argv();
run_server(argc, argv);
}
...@@ -16,12 +16,14 @@ ...@@ -16,12 +16,14 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#define CAF_SUITE io.unpublish
#include "caf/config.hpp" #include "caf/config.hpp"
#define CAF_SUITE io_unpublish #include "caf/test/io_dsl.hpp"
#include "caf/test/dsl.hpp"
#include <atomic> #include <atomic>
#include <memory>
#include <new> #include <new>
#include <thread> #include <thread>
...@@ -32,16 +34,22 @@ using namespace caf; ...@@ -32,16 +34,22 @@ using namespace caf;
namespace { 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 { class dummy : public event_based_actor {
public: 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 // nop
} }
~dummy() override { ~dummy() override {
++s_dtor_called; ssp_->dtors_called += 1;
} }
behavior make_behavior() override { behavior make_behavior() override {
...@@ -49,85 +57,54 @@ public: ...@@ -49,85 +57,54 @@ public:
// nop // nop
}}; }};
} }
};
struct config : actor_system_config { private:
config() { suite_state_ptr ssp_;
load<io::middleman>();
if (auto err = parse(test::engine::argc(), test::engine::argv()))
CAF_FAIL("failed to parse config: " << to_string(err));
}
}; };
struct fixture { struct fixture : point_to_point_fixture<> {
fixture() { fixture() {
new (&system) actor_system(cfg); prepare_connection(mars, earth, "mars", 8080);
testee = system.spawn<dummy>(); ssp = std::make_shared<suite_state>();
} }
~fixture() { ~fixture() {
anon_send_exit(testee, exit_reason::user_shutdown); run();
destroy(testee); CAF_CHECK_EQUAL(ssp->dtors_called, 2);
system.~actor_system();
CAF_CHECK_EQUAL(s_dtor_called.load(), 2);
} }
actor remote_actor(const char* hostname, uint16_t port, suite_state_ptr ssp;
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;
}; };
} // namespace } // namespace
CAF_TEST_FIXTURE_SCOPE(unpublish_tests, fixture) CAF_TEST_FIXTURE_SCOPE(unpublish_tests, fixture)
CAF_TEST(unpublishing) { CAF_TEST(actors can become unpublished) {
auto port = unbox(system.middleman().publish(testee, 0)); auto testee = mars.sys.spawn<dummy>(ssp);
CAF_REQUIRE(port != 0); auto guard = detail::make_scope_guard([&] {
CAF_MESSAGE("published actor on port " << port); // The MM holds a reference to this actor publishing it, so we need to kill
CAF_MESSAGE("test invalid unpublish"); // it manually.
auto testee2 = system.spawn<dummy>(); anon_send_exit(testee, exit_reason::user_shutdown);
system.middleman().unpublish(testee2, port); });
auto x0 = remote_actor("127.0.0.1", port); loop_after_next_enqueue(mars);
CAF_CHECK_NOT_EQUAL(x0, testee2); auto port = unbox(mars.mm.publish(testee, 8080));
CAF_CHECK_EQUAL(x0, testee); CAF_REQUIRE_EQUAL(port, 8080);
anon_send_exit(testee2, exit_reason::kill); CAF_MESSAGE("the middleman ignores invalid unpublish() calls");
CAF_MESSAGE("unpublish testee"); auto testee2 = mars.sys.spawn<dummy>(ssp);
system.middleman().unpublish(testee, port); loop_after_next_enqueue(mars);
CAF_MESSAGE("check whether testee is still available via cache"); auto res = mars.mm.unpublish(testee2, 8080);
auto x1 = remote_actor("127.0.0.1", port); CAF_CHECK(!res && res.error() == sec::no_actor_published_at_port);
CAF_CHECK_EQUAL(x1, testee); anon_send_exit(testee2, exit_reason::user_shutdown);
CAF_MESSAGE("fake death of testee and check if testee becomes unavailable"); CAF_MESSAGE("after unpublishing an actor, remotes can no longer connect");
anon_send(actor_cast<actor>(system.middleman().actor_handle()), loop_after_next_enqueue(mars);
down_msg{testee.address(), exit_reason::normal}); CAF_CHECK(mars.mm.unpublish(testee, 8080));
// must fail now // TODO: ideally, we'd check that remote actors in fact can no longer connect.
auto x2 = remote_actor("127.0.0.1", port, true); // However, the test multiplexer does not support "closing" connections
CAF_CHECK(!x2); // 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() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -25,6 +25,15 @@ ...@@ -25,6 +25,15 @@
#include "caf/test/dsl.hpp" #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 /// Ensures that `test_node_fixture` can override `run_exhaustively` even if
/// the base fixture does not declare these member functions virtual. /// the base fixture does not declare these member functions virtual.
template <class BaseFixture> template <class BaseFixture>
...@@ -46,10 +55,9 @@ public: ...@@ -46,10 +55,9 @@ public:
}; };
/// A fixture containing all required state to simulate a single CAF node. /// A fixture containing all required state to simulate a single CAF node.
template <class BaseFixture = template <class BaseFixture = io_base_fixture>
test_coordinator_fixture<caf::actor_system_config>>
class test_node_fixture : public BaseFixture, class test_node_fixture : public BaseFixture,
test_node_fixture_base<BaseFixture> { public test_node_fixture_base<BaseFixture> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -224,8 +232,7 @@ private: ...@@ -224,8 +232,7 @@ private:
/// A simple fixture that includes two nodes (`earth` and `mars`) that can /// A simple fixture that includes two nodes (`earth` and `mars`) that can
/// connect to each other. /// connect to each other.
template <class BaseFixture = template <class BaseFixture = io_base_fixture>
test_coordinator_fixture<caf::actor_system_config>>
class point_to_point_fixture class point_to_point_fixture
: public test_network_fixture_base<test_node_fixture<BaseFixture>> { : public test_network_fixture_base<test_node_fixture<BaseFixture>> {
public: public:
...@@ -243,12 +250,19 @@ public: ...@@ -243,12 +250,19 @@ public:
// Run initialization code. // Run initialization code.
this->exec_all(); this->exec_all();
} }
~point_to_point_fixture() {
run();
}
void run() {
this->exec_all();
}
}; };
/// A simple fixture that includes three nodes (`earth`, `mars`, and `jupiter`) /// A simple fixture that includes three nodes (`earth`, `mars`, and `jupiter`)
/// that can connect to each other. /// that can connect to each other.
template <class BaseFixture = template <class BaseFixture = io_base_fixture>
test_coordinator_fixture<caf::actor_system_config>>
class belt_fixture class belt_fixture
: public test_network_fixture_base<test_node_fixture<BaseFixture>> { : public test_network_fixture_base<test_node_fixture<BaseFixture>> {
public: public:
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment