Commit 794358ff authored by Jakob Otto's avatar Jakob Otto

Refactor tcp backend test

parent 0affeae9
...@@ -16,7 +16,7 @@ caf_incubator_add_enum_consistency_check("caf/net/operation.hpp" ...@@ -16,7 +16,7 @@ caf_incubator_add_enum_consistency_check("caf/net/operation.hpp"
# -- utility function for setting default properties --------------------------- # -- utility function for setting default properties ---------------------------
function(caf_net_set_default_properties) function(caf_net_set_default_properties)
foreach(target ${ARGN}) foreach (target ${ARGN})
caf_incubator_set_default_properties(${target}) caf_incubator_set_default_properties(${target})
# Make sure we find our headers plus the the generated export header. # Make sure we find our headers plus the the generated export header.
target_include_directories(${target} PRIVATE target_include_directories(${target} PRIVATE
...@@ -25,48 +25,48 @@ function(caf_net_set_default_properties) ...@@ -25,48 +25,48 @@ function(caf_net_set_default_properties)
target_compile_definitions(${target} PRIVATE libcaf_net_EXPORTS) target_compile_definitions(${target} PRIVATE libcaf_net_EXPORTS)
# Pull in public dependencies. # Pull in public dependencies.
target_link_libraries(${target} PUBLIC CAF::core) target_link_libraries(${target} PUBLIC CAF::core)
if(MSVC) if (MSVC)
target_link_libraries(${target} PUBLIC ws2_32 iphlpapi) target_link_libraries(${target} PUBLIC ws2_32 iphlpapi)
endif() endif ()
endforeach() endforeach ()
endfunction() endfunction()
# -- add library targets ------------------------------------------------------- # -- add library targets -------------------------------------------------------
add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS} add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/actor_proxy_impl.cpp src/actor_proxy_impl.cpp
src/application.cpp src/basp/application.cpp
src/basp/connection_state_strings.cpp src/basp/connection_state_strings.cpp
src/basp/ec_strings.cpp src/basp/ec_strings.cpp
src/basp/message_type_strings.cpp src/basp/message_type_strings.cpp
src/basp/operation_strings.cpp src/basp/operation_strings.cpp
src/convert_ip_endpoint.cpp src/convert_ip_endpoint.cpp
src/datagram_socket.cpp src/datagram_socket.cpp
src/defaults.cpp src/defaults.cpp
src/defaults.cpp src/defaults.cpp
src/endpoint_manager.cpp src/endpoint_manager.cpp
src/header.cpp src/header.cpp
src/host.cpp src/host.cpp
src/ip.cpp src/ip.cpp
src/message_queue.cpp src/message_queue.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/net/backend/tcp.cpp src/net/backend/tcp.cpp
src/net/backend/test.cpp src/net/backend/test.cpp
src/net/endpoint_manager_queue.cpp src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp src/net/middleman.cpp
src/net/middleman_backend.cpp src/net/middleman_backend.cpp
src/net/packet_writer.cpp src/net/packet_writer.cpp
src/network_socket.cpp src/network_socket.cpp
src/pipe_socket.cpp src/pipe_socket.cpp
src/pollset_updater.cpp src/pollset_updater.cpp
src/socket.cpp src/socket.cpp
src/socket_manager.cpp src/socket_manager.cpp
src/stream_socket.cpp src/stream_socket.cpp
src/tcp_accept_socket.cpp src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp src/tcp_stream_socket.cpp
src/udp_datagram_socket.cpp src/udp_datagram_socket.cpp
src/worker.cpp src/worker.cpp
) )
add_library(libcaf_net "${PROJECT_SOURCE_DIR}/cmake/dummy.cpp" add_library(libcaf_net "${PROJECT_SOURCE_DIR}/cmake/dummy.cpp"
$<TARGET_OBJECTS:libcaf_net_obj>) $<TARGET_OBJECTS:libcaf_net_obj>)
...@@ -80,7 +80,7 @@ set_property(TARGET libcaf_net_obj PROPERTY POSITION_INDEPENDENT_CODE ON) ...@@ -80,7 +80,7 @@ set_property(TARGET libcaf_net_obj PROPERTY POSITION_INDEPENDENT_CODE ON)
caf_net_set_default_properties(libcaf_net_obj libcaf_net) caf_net_set_default_properties(libcaf_net_obj libcaf_net)
target_include_directories(libcaf_net INTERFACE target_include_directories(libcaf_net INTERFACE
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>) $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>)
add_library(CAF::net ALIAS libcaf_net) add_library(CAF::net ALIAS libcaf_net)
...@@ -108,9 +108,9 @@ install(DIRECTORY "${CMAKE_CURRENT_SOURCE_DIR}/caf" ...@@ -108,9 +108,9 @@ install(DIRECTORY "${CMAKE_CURRENT_SOURCE_DIR}/caf"
# -- build unit tests ---------------------------------------------------------- # -- build unit tests ----------------------------------------------------------
if(NOT CAF_INC_ENABLE_TESTING) if (NOT CAF_INC_ENABLE_TESTING)
return() return()
endif() endif ()
add_executable(caf-net-test add_executable(caf-net-test
test/net-test.cpp test/net-test.cpp
...@@ -123,30 +123,30 @@ target_include_directories(caf-net-test PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/tes ...@@ -123,30 +123,30 @@ target_include_directories(caf-net-test PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/tes
target_link_libraries(caf-net-test PRIVATE CAF::test) target_link_libraries(caf-net-test PRIVATE CAF::test)
caf_incubator_add_test_suites(caf-net-test caf_incubator_add_test_suites(caf-net-test
net.basp.message_queue net.basp.message_queue
net.basp.ping_pong net.basp.ping_pong
net.basp.worker net.basp.worker
accept_socket accept_socket
pipe_socket pipe_socket
application application
socket socket
convert_ip_endpoint convert_ip_endpoint
socket_guard socket_guard
datagram_socket datagram_socket
stream_application stream_application
datagram_transport datagram_transport
stream_socket stream_socket
doorman doorman
stream_transport stream_transport
endpoint_manager endpoint_manager
string_application string_application
header header
tcp_sockets tcp_sockets
ip ip
transport_worker transport_worker
multiplexer multiplexer
transport_worker_dispatcher transport_worker_dispatcher
udp_datagram_socket udp_datagram_socket
network_socket network_socket
net.backend.tcp net.backend.tcp
) )
...@@ -53,7 +53,7 @@ public: ...@@ -53,7 +53,7 @@ public:
endpoint_manager_ptr peer(const node_id& id) override; endpoint_manager_ptr peer(const node_id& id) override;
void publish(actor handle, const uri& locator) override; void publish(actor handle, string_view path) override;
void resolve(const uri& locator, const actor& listener) override; void resolve(const uri& locator, const actor& listener) override;
......
...@@ -53,7 +53,7 @@ public: ...@@ -53,7 +53,7 @@ public:
endpoint_manager_ptr connect(const uri& locator) override; endpoint_manager_ptr connect(const uri& locator) override;
void publish(actor handle, const uri& locator) override; void publish(actor handle, string_view path) override;
void resolve(const uri& locator, const actor& listener) override; void resolve(const uri& locator, const actor& listener) override;
......
...@@ -81,14 +81,17 @@ public: ...@@ -81,14 +81,17 @@ public:
// Publishes an actor. // Publishes an actor.
template <class Handle = actor> template <class Handle = actor>
error publish(Handle whom, const uri& locator) { error publish(Handle whom, const uri& locator) {
auto be = backend(locator.scheme()); if (auto be = backend(locator.scheme()))
be->publish(whom, locator); be->publish(whom, locator.path());
else
return sec::runtime_error;
return none;
} }
/// Resolves a path to a remote actor. /// Resolves a path to a remote actor.
void resolve(const uri& locator, const actor& listener); void resolve(const uri& locator, const actor& listener);
template <class Handle> template <class Handle = actor>
expected<Handle> remote_actor(const uri& locator) { expected<Handle> remote_actor(const uri& locator) {
// TODO: Use function view? // TODO: Use function view?
scoped_actor self{sys_}; scoped_actor self{sys_};
...@@ -99,7 +102,14 @@ public: ...@@ -99,7 +102,14 @@ public:
[&actor_ptr](strong_actor_ptr& ptr, const std::set<std::string>&) { [&actor_ptr](strong_actor_ptr& ptr, const std::set<std::string>&) {
actor_ptr = ptr; actor_ptr = ptr;
}, },
[&err](const error& e) { err = e; }); [&err](const error& e) {
err = e;
} /*,
after(std::chrono::seconds(5)) >>
[&err] {
err = make_error(sec::runtime_error,
"manager did not respond with a proxy.");
}*/);
if (err) if (err)
return err; return err;
auto res = actor_cast<Handle>(actor_ptr); auto res = actor_cast<Handle>(actor_ptr);
......
...@@ -50,7 +50,7 @@ public: ...@@ -50,7 +50,7 @@ public:
virtual endpoint_manager_ptr connect(const uri& locator) = 0; virtual endpoint_manager_ptr connect(const uri& locator) = 0;
/// Publishes an actor. /// Publishes an actor.
virtual void publish(actor handle, const uri& locator) = 0; virtual void publish(actor handle, string_view path) = 0;
/// Resolves a path to a remote actor. /// Resolves a path to a remote actor.
virtual void resolve(const uri& locator, const actor& listener) = 0; virtual void resolve(const uri& locator, const actor& listener) = 0;
......
...@@ -29,7 +29,6 @@ ...@@ -29,7 +29,6 @@
#include "caf/detail/network_order.hpp" #include "caf/detail/network_order.hpp"
#include "caf/detail/parse.hpp" #include "caf/detail/parse.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/basp/constants.hpp" #include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
......
...@@ -101,8 +101,7 @@ endpoint_manager_ptr tcp::peer(const node_id& id) { ...@@ -101,8 +101,7 @@ endpoint_manager_ptr tcp::peer(const node_id& id) {
return get_peer(id); return get_peer(id);
} }
void tcp::publish(actor handle, const uri& locator) { void tcp::publish(actor handle, string_view path) {
auto path = locator.path();
mm_.system().registry().put(std::string(path.begin(), path.end()), handle); mm_.system().registry().put(std::string(path.begin(), path.end()), handle);
} }
......
...@@ -58,7 +58,7 @@ endpoint_manager_ptr test::connect(const uri&) { ...@@ -58,7 +58,7 @@ endpoint_manager_ptr test::connect(const uri&) {
return nullptr; return nullptr;
} }
void test::publish(actor, const uri&) { void test::publish(actor, string_view) {
// nop // nop
} }
......
...@@ -24,38 +24,130 @@ ...@@ -24,38 +24,130 @@
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include <string> #include <string>
#include <thread>
#include "caf/actor_system_config.hpp"
#include "caf/ip_endpoint.hpp" #include "caf/ip_endpoint.hpp"
#include "caf/net/middleman.hpp" #include "caf/net/middleman.hpp"
#include "caf/net/socket_guard.hpp" #include "caf/net/socket_guard.hpp"
#include "caf/net/tcp_accept_socket.hpp" #include "caf/net/tcp_accept_socket.hpp"
#include "caf/uri.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
using namespace std::literals::string_literals;
namespace { namespace {
behavior dummy_actor(event_based_actor*) {
return {
// nop
};
}
struct earth_node {
uri operator()() {
return unbox(make_uri("tcp://earth"));
}
};
struct mars_node {
uri operator()() {
return unbox(make_uri("tcp://mars"));
}
};
template <class Node>
struct config : actor_system_config { struct config : actor_system_config {
config() { config() {
put(content, "middleman.this-node", unbox(make_uri("tcp://earth"))); Node this_node;
put(content, "middleman.this-node", this_node());
load<middleman, backend::tcp>(); load<middleman, backend::tcp>();
} }
}; };
struct fixture : test_coordinator_fixture<config>, host_fixture { class planet_driver {
using byte_buffer_ptr = std::shared_ptr<byte_buffer>; public:
virtual ~planet_driver() {
// nop
}
virtual bool consume_message() = 0;
fixture() : mm{sys.network_manager()}, mpx{mm.mpx()} { virtual bool handle_io_event() = 0;
virtual bool trigger_timeout() = 0;
};
template <class Node>
class planet : public test_coordinator_fixture<config<Node>> {
public:
planet(planet_driver& driver)
: mm(this->sys.network_manager()), mpx(mm.mpx()), driver_(driver) {
mpx->set_thread_id(); mpx->set_thread_id();
handle_io_events(); }
node_id id() const {
return this->sys.node();
}
bool consume_message() override {
return driver_.consume_message();
} }
bool handle_io_event() override { bool handle_io_event() override {
return mpx->poll_once(false); return driver_.handle_io_event();
}
bool trigger_timeout() override {
return driver_.trigger_timeout();
}
actor resolve(string_view locator) {
auto hdl = actor_cast<actor>(this->self);
this->sys.network_manager().resolve(unbox(make_uri(locator)), hdl);
this->run();
actor result;
this->self->receive(
[&](strong_actor_ptr& ptr, const std::set<std::string>&) {
CAF_MESSAGE("resolved " << locator);
result = actor_cast<actor>(std::move(ptr));
});
return result;
} }
net::middleman& mm; net::middleman& mm;
const multiplexer_ptr& mpx; multiplexer_ptr mpx;
private:
planet_driver& driver_;
};
struct fixture : host_fixture, planet_driver {
fixture() : earth(*this), mars(*this) {
run();
CAF_REQUIRE_EQUAL(earth.mpx->num_socket_managers(), 2);
CAF_REQUIRE_EQUAL(mars.mpx->num_socket_managers(), 2);
}
bool consume_message() override {
return earth.sched.try_run_once() || mars.sched.try_run_once();
}
bool handle_io_event() override {
return earth.mpx->poll_once(false) || mars.mpx->poll_once(false);
}
bool trigger_timeout() override {
return earth.sched.trigger_timeout() || mars.sched.trigger_timeout();
}
void run() {
earth.run();
}
planet<earth_node> earth;
planet<mars_node> mars;
}; };
} // namespace } // namespace
...@@ -63,45 +155,72 @@ struct fixture : test_coordinator_fixture<config>, host_fixture { ...@@ -63,45 +155,72 @@ struct fixture : test_coordinator_fixture<config>, host_fixture {
CAF_TEST_FIXTURE_SCOPE(tcp_backend_tests, fixture) CAF_TEST_FIXTURE_SCOPE(tcp_backend_tests, fixture)
CAF_TEST(doorman accept) { CAF_TEST(doorman accept) {
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2); auto backend = earth.mm.backend("tcp");
auto backend = mm.backend("tcp");
CAF_CHECK(backend); CAF_CHECK(backend);
auto port = backend->port(); uri::authority_type auth;
CAF_MESSAGE("trying to connect to system with " << CAF_ARG(port)); auth.host = "localhost"s;
ip_endpoint ep; auth.port = backend->port();
auto ep_str = std::string("[::1]:") + std::to_string(port); CAF_MESSAGE("trying to connect to earth at " << CAF_ARG(auth));
if (auto err = detail::parse(ep_str, ep)) auto sock = make_connected_tcp_stream_socket(auth);
CAF_FAIL("could not parse " << CAF_ARG(ep_str) << " " << CAF_ARG(err));
auto sock = make_connected_tcp_stream_socket(ep);
if (!sock)
CAF_FAIL("could not connect");
auto guard = make_socket_guard(*sock);
handle_io_event(); handle_io_event();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3); CAF_CHECK(sock);
auto guard = make_socket_guard(*sock);
CAF_REQUIRE_EQUAL(earth.mpx->num_socket_managers(), 3);
} }
CAF_TEST(connect) { CAF_TEST(connect) {
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2); uri::authority_type auth;
ip_endpoint ep; auth.host = "0.0.0.0"s;
if (auto err = detail::parse("[::]:0", ep)) auth.port = 0;
CAF_FAIL("could not parse endpoint" << err); auto acceptor = unbox(make_tcp_accept_socket(auth, false));
auto acceptor = make_tcp_accept_socket(ep); auto acc_guard = make_socket_guard(acceptor);
if (!acceptor) auto port = unbox(local_port(acc_guard.socket()));
CAF_FAIL("could not create acceptor" << CAF_ARG2("err", acceptor.error())); auto uri_str = std::string("tcp://localhost:") + std::to_string(port);
auto acc_guard = make_socket_guard(*acceptor);
auto port = local_port(acc_guard.socket());
if (!port)
CAF_FAIL("could not read port" << CAF_ARG2("err", port.error()));
auto uri_str = std::string("tcp://localhost:") + std::to_string(*port);
CAF_MESSAGE("connecting to " << CAF_ARG(uri_str)); CAF_MESSAGE("connecting to " << CAF_ARG(uri_str));
auto ret = mm.connect(*make_uri(uri_str)); unbox(earth.mm.connect(*make_uri(uri_str)));
if (!ret) CAF_CHECK(accept(acc_guard.socket()));
CAF_FAIL("could not connect to " << uri_str);
auto sock = accept(acc_guard.socket());
if (!sock)
CAF_FAIL("accepting failed");
handle_io_event(); handle_io_event();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3); CAF_REQUIRE_EQUAL(earth.mpx->num_socket_managers(), 3);
} }
/*
CAF_TEST(publish and resolve) {
auto dummy = sys.spawn(dummy_actor);
auto port = unbox(mm.port("tcp"));
auto locator = unbox(make_uri("tcp://localhost:"s + std::to_string(port)));
mm.publish(dummy, locator);
auto ret = sys.registry().get(
std::string(locator.path().begin(), locator.path().end()));
if (!ret)
CAF_FAIL("could not retrieve published actor");
config<mars_node> mars_cfg;
actor_system mars_sys{mars_cfg};
auto& mars_mm = mars_sys.network_manager();
auto mars_mpx = mars_mm.mpx();
mars_mpx->set_thread_id();
mars_mpx->poll_once(false);
CAF_REQUIRE(mars_mpx->num_socket_managers(), 2);
CAF_MESSAGE("connecting to " << CAF_ARG(locator));
unbox(mars_mm.connect(locator));
run();
mars_mpx->poll_once(false);
auto locator
= unbox(make_uri("tcp://localhost:"s + std::to_string(port) + "/dummy"));
CAF_MESSAGE("resolving actor " << CAF_ARG(actor_locator));
mars_mm.resolve(locator, self);
mars_mpx->poll_once(false);
run();
mars_mpx->poll_once(false);
CAF_MESSAGE("receiving actor");
self->receive(
[](strong_actor_ptr&, const std::set<std::string>&) {
},
[](const error& err) {
CAF_FAIL("error resolving actor " << CAF_ARG(err));
},
after(std::chrono::seconds(5)) >>
[] { CAF_FAIL("manager did not respond with a proxy."); });
}
*/
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
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