Commit 60be58d1 authored by Joseph Noir's avatar Joseph Noir

Update doorman to work with scribe implementation

parent dde1273f
...@@ -24,9 +24,11 @@ ...@@ -24,9 +24,11 @@
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/net/tcp.hpp"
#include "caf/policy/scribe.hpp" #include "caf/policy/scribe.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
...@@ -44,20 +46,17 @@ public: ...@@ -44,20 +46,17 @@ public:
} }
template <class Parent> template <class Parent>
error init(Parent&) { error init(Parent& parent) {
parent.application().init(parent);
parent.mask_add(net::operation::read);
return none; return none;
} }
template <class Parent> template <class Parent>
bool handle_read_event(Parent& parent) { bool handle_read_event(Parent& parent) {
auto sck = accept(acceptor_.id, nullptr, nullptr); auto sck = net::tcp::accept(net::socket_cast<net::stream_socket>(acceptor_));
if (sck == net::invalid_socket_id) { if (!sck) {
auto err = net::last_socket_error(); CAF_LOG_ERROR("accept failed:" << sck.error());
if (err != std::errc::operation_would_block
&& err != std::errc::resource_unavailable_try_again) {
CAF_LOG_ERROR("accept failed:" << net::socket_error_as_string(err));
return false;
}
return false; return false;
} }
auto mpx = parent.multiplexer(); auto mpx = parent.multiplexer();
...@@ -65,8 +64,8 @@ public: ...@@ -65,8 +64,8 @@ public:
CAF_LOG_DEBUG("could not acquire multiplexer to create a new endpoint manager"); CAF_LOG_DEBUG("could not acquire multiplexer to create a new endpoint manager");
return false; return false;
} }
auto child = make_endpoint_manager(std::move(mpx), parent.system(), auto child = make_endpoint_manager(mpx, parent.system(),
scribe{sck}, scribe{*sck},
parent.application().make()); parent.application().make());
if (auto err = child->init()) { if (auto err = child->init()) {
return false; return false;
......
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.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/tcp.hpp"
#include "caf/policy/doorman.hpp" #include "caf/policy/doorman.hpp"
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
...@@ -64,13 +65,14 @@ public: ...@@ -64,13 +65,14 @@ public:
} }
template <class Transport> template <class Transport>
bool handle_read_event(Transport&) { void write_message(Transport& transport,
return false; std::unique_ptr<endpoint_manager::message> msg) {
transport.write_packet(msg->payload);
} }
template <class Transport> template <class Parent>
bool handle_write_event(Transport&) { void handle_data(Parent&, span<char>) {
return false; // nop
} }
template <class Transport> template <class Transport>
...@@ -95,6 +97,11 @@ public: ...@@ -95,6 +97,11 @@ public:
return dummy_application::serialize(sys, x); return dummy_application::serialize(sys, x);
} }
template <class Parent>
void init(Parent&) {
// nop
}
dummy_application make() const { dummy_application make() const {
return dummy_application{}; return dummy_application{};
} }
...@@ -104,11 +111,36 @@ public: ...@@ -104,11 +111,36 @@ public:
CAF_TEST_FIXTURE_SCOPE(doorman_tests, fixture) CAF_TEST_FIXTURE_SCOPE(doorman_tests, fixture)
CAF_TEST(doorman creation) { CAF_TEST(tcp connect) {
auto acceptor = unbox(make_accept_socket(0, nullptr, false)); auto acceptor = unbox(tcp::make_accept_socket(0, nullptr, false));
auto port = unbox(local_port(socket_cast<network_socket>(acceptor)));
CAF_MESSAGE("opened acceptor on port " << port);
auto con_fd = unbox(tcp::make_connected_socket("localhost", port));
auto acc_fd = unbox(tcp::accept(acceptor));
CAF_MESSAGE("accepted connection");
close(con_fd);
close(acc_fd);
close(acceptor);
}
CAF_TEST(doorman accept) {
auto acceptor = unbox(tcp::make_accept_socket(0, nullptr, false));
auto port = unbox(local_port(socket_cast<network_socket>(acceptor)));
CAF_MESSAGE("opened acceptor on port " << port);
auto mgr = make_endpoint_manager(mpx, sys, auto mgr = make_endpoint_manager(mpx, sys,
policy::doorman{acceptor}, policy::doorman{acceptor},
dummy_application_factory{}); dummy_application_factory{});
CAF_CHECK_EQUAL(mgr->init(), none);
handle_io_event();
auto before = mpx->num_socket_managers();
CAF_MESSAGE("connecting to doorman");
auto fd = unbox(tcp::make_connected_socket("localhost", port));
CAF_MESSAGE("waiting for connection");
while (mpx->num_socket_managers() != before + 1)
handle_io_event();
CAF_MESSAGE("connected");
close(acceptor);
close(fd);
} }
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