Unverified Commit 89dddbcf authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #35

Streamline stream_transport test
parents 493ec8b2 75277e98
...@@ -27,21 +27,48 @@ namespace net { ...@@ -27,21 +27,48 @@ namespace net {
template <class Socket> template <class Socket>
class socket_guard { class socket_guard {
public: public:
explicit socket_guard(Socket sock) : sock_(sock) { socket_guard() noexcept : sock_(invalid_socket_id) {
// nop // nop
} }
explicit socket_guard(Socket sock) noexcept : sock_(sock) {
// nop
}
socket_guard(socket_guard&& other) noexcept : sock_(other.release()) {
// nop
}
socket_guard(const socket_guard&) = delete;
socket_guard& operator=(socket_guard&& other) noexcept {
reset(other.release());
return *this;
}
socket_guard& operator=(const socket_guard&) = delete;
~socket_guard() { ~socket_guard() {
if (sock_.id != invalid_socket_id) if (sock_.id != invalid_socket_id)
close(sock_); close(sock_);
} }
Socket release() { void reset(Socket x) noexcept {
if (sock_.id != invalid_socket_id)
close(sock_);
sock_ = x;
}
Socket release() noexcept {
auto sock = sock_; auto sock = sock_;
sock_.id = invalid_socket_id; sock_.id = invalid_socket_id;
return sock; return sock;
} }
Socket socket() const noexcept {
return sock_;
}
private: private:
Socket sock_; Socket sock_;
}; };
......
...@@ -32,7 +32,7 @@ namespace { ...@@ -32,7 +32,7 @@ namespace {
constexpr socket_id dummy_id = 13; constexpr socket_id dummy_id = 13;
struct dummy_socket { struct dummy_socket {
dummy_socket(socket_id& id, bool& closed) : id(id), closed(closed) { dummy_socket(socket_id id, bool& closed) : id(id), closed(closed) {
// nop // nop
} }
...@@ -44,20 +44,20 @@ struct dummy_socket { ...@@ -44,20 +44,20 @@ struct dummy_socket {
return *this; return *this;
} }
socket_id& id; socket_id id;
bool& closed; bool& closed;
}; };
void close(dummy_socket x) { void close(dummy_socket x) {
x.id = invalid_socket_id;
x.closed = true; x.closed = true;
} }
struct fixture { struct fixture {
fixture() : id{dummy_id}, closed{false}, sock{id, closed} { fixture() : closed{false}, sock{dummy_id, closed} {
// nop // nop
} }
socket_id id;
bool closed; bool closed;
dummy_socket sock; dummy_socket sock;
}; };
...@@ -69,17 +69,29 @@ CAF_TEST_FIXTURE_SCOPE(socket_guard_tests, fixture) ...@@ -69,17 +69,29 @@ CAF_TEST_FIXTURE_SCOPE(socket_guard_tests, fixture)
CAF_TEST(cleanup) { CAF_TEST(cleanup) {
{ {
auto guard = make_socket_guard(sock); auto guard = make_socket_guard(sock);
CAF_CHECK_EQUAL(sock.id, dummy_id); CAF_CHECK_EQUAL(guard.socket().id, dummy_id);
} }
CAF_CHECK(sock.closed); CAF_CHECK(sock.closed);
} }
CAF_TEST(reset) {
{
auto guard = make_socket_guard(sock);
CAF_CHECK_EQUAL(guard.socket().id, dummy_id);
guard.release();
CAF_CHECK_EQUAL(guard.socket().id, invalid_socket_id);
guard.reset(sock);
CAF_CHECK_EQUAL(guard.socket().id, dummy_id);
}
CAF_CHECK_EQUAL(sock.closed, true);
}
CAF_TEST(release) { CAF_TEST(release) {
{ {
auto guard = make_socket_guard(sock); auto guard = make_socket_guard(sock);
CAF_CHECK_EQUAL(sock.id, dummy_id); CAF_CHECK_EQUAL(guard.socket().id, dummy_id);
guard.release(); guard.release();
CAF_CHECK_EQUAL(sock.id, invalid_socket_id); CAF_CHECK_EQUAL(guard.socket().id, invalid_socket_id);
} }
CAF_CHECK_EQUAL(sock.closed, false); CAF_CHECK_EQUAL(sock.closed, false);
} }
......
...@@ -29,8 +29,10 @@ ...@@ -29,8 +29,10 @@
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp" #include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_impl.hpp"
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/serializer_impl.hpp" #include "caf/serializer_impl.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
...@@ -39,15 +41,24 @@ using namespace caf; ...@@ -39,15 +41,24 @@ using namespace caf;
using namespace caf::net; using namespace caf::net;
namespace { namespace {
constexpr string_view hello_manager = "hello manager!"; constexpr string_view hello_manager = "hello manager!";
struct fixture : test_coordinator_fixture<>, host_fixture { struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() { using buffer_type = std::vector<byte>;
using buffer_ptr = std::shared_ptr<buffer_type>;
fixture() : recv_buf(1024), shared_buf{std::make_shared<buffer_type>()} {
mpx = std::make_shared<multiplexer>(); mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init()) if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err)); CAF_FAIL("mpx->init failed: " << sys.render(err));
mpx->set_thread_id(); mpx->set_thread_id();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto sockets = unbox(make_stream_socket_pair());
send_socket_guard.reset(sockets.first);
recv_socket_guard.reset(sockets.second);
if (auto err = nonblocking(recv_socket_guard.socket(), true))
CAF_FAIL("nonblocking returned an error: " << err);
} }
bool handle_io_event() override { bool handle_io_event() override {
...@@ -55,11 +66,19 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -55,11 +66,19 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
} }
multiplexer_ptr mpx; multiplexer_ptr mpx;
buffer_type recv_buf;
socket_guard<stream_socket> send_socket_guard;
socket_guard<stream_socket> recv_socket_guard;
buffer_ptr shared_buf;
}; };
class dummy_application { class dummy_application {
using buffer_type = std::vector<byte>;
using buffer_ptr = std::shared_ptr<buffer_type>;
public: public:
dummy_application(std::shared_ptr<std::vector<byte>> rec_buf) dummy_application(buffer_ptr rec_buf)
: rec_buf_(std::move(rec_buf)){ : rec_buf_(std::move(rec_buf)){
// nop // nop
}; };
...@@ -118,17 +137,17 @@ public: ...@@ -118,17 +137,17 @@ public:
// nop // nop
} }
static expected<std::vector<byte>> serialize(actor_system& sys, static expected<buffer_type> serialize(actor_system& sys,
const type_erased_tuple& x) { const type_erased_tuple& x) {
std::vector<byte> result; buffer_type result;
serializer_impl<std::vector<byte>> sink{sys, result}; serializer_impl<buffer_type> sink{sys, result};
if (auto err = message::save(sink, x)) if (auto err = message::save(sink, x))
return err; return err;
return result; return result;
} }
private: private:
std::shared_ptr<std::vector<byte>> rec_buf_; buffer_ptr rec_buf_;
}; };
} // namespace } // namespace
...@@ -137,39 +156,32 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture) ...@@ -137,39 +156,32 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) { CAF_TEST(receive) {
using transport_type = stream_transport<dummy_application>; using transport_type = stream_transport<dummy_application>;
std::vector<byte> read_buf(1024); auto mgr = make_endpoint_manager(mpx, sys,
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u); transport_type{recv_socket_guard.release(),
auto buf = std::make_shared<std::vector<byte>>(); dummy_application{
auto sockets = unbox(make_stream_socket_pair()); shared_buf}});
if (auto err = nonblocking(sockets.second, true))
CAF_FAIL("nonblocking() returned an error: " << err);
CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)),
sec::unavailable_or_would_block);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
transport_type transport{sockets.first, dummy_application{buf}};
transport.configure_read(net::receive_policy::exactly(hello_manager.size()));
auto mgr = make_endpoint_manager(mpx, sys, transport);
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
auto mgr_impl = mgr.downcast<endpoint_manager_impl<transport_type>>();
CAF_CHECK(mgr_impl != nullptr);
auto& transport = mgr_impl->transport();
transport.configure_read(receive_policy::exactly(hello_manager.size()));
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_CHECK_EQUAL(write(sockets.second, as_bytes(make_span(hello_manager))), CAF_CHECK_EQUAL(write(send_socket_guard.socket(),
as_bytes(make_span(hello_manager))),
hello_manager.size()); hello_manager.size());
CAF_MESSAGE("wrote " << hello_manager.size() << " bytes."); CAF_MESSAGE("wrote " << hello_manager.size() << " bytes.");
run(); run();
CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(buf->data()), CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(shared_buf->data()),
buf->size()), shared_buf->size()),
hello_manager); hello_manager);
} }
CAF_TEST(resolve and proxy communication) { CAF_TEST(resolve and proxy communication) {
using transport_type = stream_transport<dummy_application>; using transport_type = stream_transport<dummy_application>;
std::vector<byte> read_buf(1024);
auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys, auto mgr = make_endpoint_manager(mpx, sys,
transport_type{sockets.first, transport_type{send_socket_guard.release(),
dummy_application{buf}}); dummy_application{
shared_buf}});
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
run(); run();
mgr->resolve(unbox(make_uri("test:/id/42")), self); mgr->resolve(unbox(make_uri("test:/id/42")), self);
...@@ -182,13 +194,13 @@ CAF_TEST(resolve and proxy communication) { ...@@ -182,13 +194,13 @@ CAF_TEST(resolve and proxy communication) {
after(std::chrono::seconds(0)) >> after(std::chrono::seconds(0)) >>
[&] { CAF_FAIL("manager did not respond with a proxy."); }); [&] { CAF_FAIL("manager did not respond with a proxy."); });
run(); run();
auto read_res = read(sockets.second, make_span(read_buf)); auto read_res = read(recv_socket_guard.socket(), make_span(recv_buf));
if (!holds_alternative<size_t>(read_res)) if (!holds_alternative<size_t>(read_res))
CAF_FAIL("read() returned an error: " << sys.render(get<sec>(read_res))); CAF_FAIL("read() returned an error: " << sys.render(get<sec>(read_res)));
read_buf.resize(get<size_t>(read_res)); recv_buf.resize(get<size_t>(read_res));
CAF_MESSAGE("receive buffer contains " << read_buf.size() << " bytes"); CAF_MESSAGE("receive buffer contains " << recv_buf.size() << " bytes");
message msg; message msg;
binary_deserializer source{sys, read_buf}; binary_deserializer source{sys, recv_buf};
CAF_CHECK_EQUAL(source(msg), none); CAF_CHECK_EQUAL(source(msg), none);
if (msg.match_elements<std::string>()) if (msg.match_elements<std::string>())
CAF_CHECK_EQUAL(msg.get_as<std::string>(0), "hello proxy!"); CAF_CHECK_EQUAL(msg.get_as<std::string>(0), "hello proxy!");
......
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