Unverified Commit d6e0c75f authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #46

Add Doorman
parents 442e8336 e6f83819
...@@ -36,10 +36,10 @@ set(LIBCAF_NET_SRCS ...@@ -36,10 +36,10 @@ set(LIBCAF_NET_SRCS
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/stream_socket.cpp
src/tcp_accept_socket.cpp src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp src/tcp_stream_socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/udp_datagram_socket.cpp src/udp_datagram_socket.cpp
src/defaults.cpp src/defaults.cpp
src/message_queue.cpp src/message_queue.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <vector>
#include "caf/logger.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/send.hpp"
namespace caf::net {
/// A doorman accepts TCP connections and creates stream_transports to handle
/// them.
template <class Factory>
class doorman {
public:
// -- member types -----------------------------------------------------------
using factory_type = Factory;
using application_type = typename Factory::application_type;
// -- constructors, destructors, and assignment operators --------------------
explicit doorman(net::tcp_accept_socket acceptor, factory_type factory)
: acceptor_(acceptor), factory_(std::move(factory)) {
// nop
}
// -- properties -------------------------------------------------------------
net::tcp_accept_socket handle() {
return acceptor_;
}
// -- member functions -------------------------------------------------------
template <class Parent>
error init(Parent& parent) {
// TODO: is initializing application factory nessecary?
if (auto err = factory_.init(parent))
return err;
return none;
}
template <class Parent>
bool handle_read_event(Parent& parent) {
auto x = net::accept(acceptor_);
if (!x) {
CAF_LOG_ERROR("accept failed:" << parent.system().render(x.error()));
return false;
}
auto mpx = parent.multiplexer();
if (!mpx) {
CAF_LOG_DEBUG("unable to get multiplexer from parent");
return false;
}
auto child = make_endpoint_manager(mpx, parent.system(),
stream_transport<
application_type>{*x,
factory_.make()});
if (auto err = child->init())
return false;
return true;
}
template <class Parent>
bool handle_write_event(Parent&) {
CAF_LOG_ERROR("doorman received write event");
return false;
}
template <class Parent>
void resolve(Parent&, const uri& locator, const actor& listener) {
CAF_LOG_ERROR("doorman called to resolve" << CAF_ARG(locator));
anon_send(listener, resolve_atom::value, "doormen cannot resolve paths");
}
void new_proxy(endpoint_manager&, const node_id& peer, actor_id id) {
CAF_LOG_ERROR("doorman received new_proxy" << CAF_ARG(peer) << CAF_ARG(id));
CAF_IGNORE_UNUSED(peer);
CAF_IGNORE_UNUSED(id);
}
void local_actor_down(endpoint_manager&, const node_id& peer, actor_id id,
error reason) {
CAF_LOG_ERROR("doorman received local_actor_down"
<< CAF_ARG(peer) << CAF_ARG(id) << CAF_ARG(reason));
CAF_IGNORE_UNUSED(peer);
CAF_IGNORE_UNUSED(id);
CAF_IGNORE_UNUSED(reason);
}
template <class Parent>
void timeout(Parent&, atom_value x, uint64_t id) {
CAF_LOG_ERROR("doorman received timeout" << CAF_ARG(x) << CAF_ARG(id));
CAF_IGNORE_UNUSED(x);
CAF_IGNORE_UNUSED(id);
}
void handle_error(sec err) {
CAF_LOG_ERROR("doorman encounterd error: " << err);
}
private:
net::tcp_accept_socket acceptor_;
factory_type factory_;
};
} // namespace caf::net
...@@ -36,7 +36,7 @@ public: ...@@ -36,7 +36,7 @@ public:
/// @pre `handle != invalid_socket` /// @pre `handle != invalid_socket`
socket_manager(socket handle, const multiplexer_ptr& parent); socket_manager(socket handle, const multiplexer_ptr& parent);
virtual ~socket_manager(); ~socket_manager() override;
socket_manager(const socket_manager&) = delete; socket_manager(const socket_manager&) = delete;
...@@ -49,6 +49,11 @@ public: ...@@ -49,6 +49,11 @@ public:
return handle_; return handle_;
} }
/// Returns a pointer to the multiplexer running this `socket_manager`.
multiplexer_ptr multiplexer() const {
return parent_.lock();
}
/// Returns registered operations (read, write, or both). /// Returns registered operations (read, write, or both).
operation mask() const noexcept { operation mask() const noexcept {
return mask_; return mask_;
......
...@@ -26,7 +26,6 @@ ...@@ -26,7 +26,6 @@
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/io/network/protocol.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
......
...@@ -24,6 +24,9 @@ ...@@ -24,6 +24,9 @@
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
#ifdef CAF_POSIX #ifdef CAF_POSIX
......
...@@ -118,7 +118,7 @@ make_tcp_accept_socket(const uri::authority_type& node, bool reuse_addr) { ...@@ -118,7 +118,7 @@ make_tcp_accept_socket(const uri::authority_type& node, bool reuse_addr) {
if (auto ip = get_if<ip_address>(&node.host)) if (auto ip = get_if<ip_address>(&node.host))
return make_tcp_accept_socket(ip_endpoint{*ip, node.port}, reuse_addr); return make_tcp_accept_socket(ip_endpoint{*ip, node.port}, reuse_addr);
auto host = get<std::string>(node.host); auto host = get<std::string>(node.host);
auto addrs = ip::resolve(host); auto addrs = ip::local_addresses(host);
if (addrs.empty()) if (addrs.empty())
return make_error(sec::cannot_open_port, "no local interface available", return make_error(sec::cannot_open_port, "no local interface available",
to_string(node)); to_string(node));
......
...@@ -83,7 +83,7 @@ make_connected_tcp_stream_socket(const uri::authority_type& node) { ...@@ -83,7 +83,7 @@ make_connected_tcp_stream_socket(const uri::authority_type& node) {
return make_error(sec::cannot_connect_to_node, "port is zero"); return make_error(sec::cannot_connect_to_node, "port is zero");
std::vector<ip_address> addrs; std::vector<ip_address> addrs;
if (auto str = get_if<std::string>(&node.host)) if (auto str = get_if<std::string>(&node.host))
addrs = ip::resolve(std::move(*str)); addrs = ip::local_addresses(*str);
else if (auto addr = get_if<ip_address>(&node.host)) else if (auto addr = get_if<ip_address>(&node.host))
addrs.push_back(*addr); addrs.push_back(*addr);
if (addrs.empty()) if (addrs.empty())
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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 net.tcp_accept_socket
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/uri.hpp"
#include "caf/test/dsl.hpp"
#include "caf/net/test/host_fixture.hpp"
using namespace caf;
using namespace caf::net;
using namespace std::literals::string_literals;
namespace {
struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() {
auth.port = 0;
auth.host = "0.0.0.0"s;
}
uri::authority_type auth;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(doorman_tests, fixture)
CAF_TEST(tcp connect) {
auto acceptor = unbox(make_tcp_accept_socket(auth, false));
auto port = unbox(local_port(socket_cast<network_socket>(acceptor)));
auto acceptor_guard = make_socket_guard(acceptor);
CAF_MESSAGE("opened acceptor on port " << port);
uri::authority_type dst;
dst.port = port;
dst.host = "localhost"s;
auto conn = make_socket_guard(unbox(make_connected_tcp_stream_socket(dst)));
auto accepted = make_socket_guard(unbox(accept(acceptor)));
CAF_MESSAGE("accepted connection");
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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 net.doorman
#include "caf/net/doorman.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/uri.hpp"
#include "caf/test/dsl.hpp"
#include "caf/net/test/host_fixture.hpp"
using namespace caf;
using namespace caf::net;
using namespace std::literals::string_literals;
namespace {
struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
mpx->set_thread_id();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auth.port = 0;
auth.host = "0.0.0.0"s;
}
bool handle_io_event() override {
return mpx->poll_once(false);
}
multiplexer_ptr mpx;
uri::authority_type auth;
};
class dummy_application {
public:
static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<byte> result;
binary_serializer sink{sys, result};
if (auto err = message::save(sink, x))
return err.value();
return result;
}
template <class Parent>
error init(Parent&) {
return none;
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
parent.write_packet(msg->payload);
}
template <class Parent>
error handle_data(Parent&, span<const byte>) {
return none;
}
template <class Parent>
void resolve(Parent&, string_view path, const actor& listener) {
anon_send(listener, resolve_atom::value,
"the resolved path is still "
+ std::string(path.begin(), path.end()));
}
template <class Parent>
void timeout(Parent&, atom_value, uint64_t) {
// nop
}
template <class Parent>
void new_proxy(Parent&, actor_id) {
// nop
}
template <class Parent>
void local_actor_down(Parent&, actor_id, error) {
// nop
}
void handle_error(sec) {
// nop
}
};
class dummy_application_factory {
public:
using application_type = dummy_application;
static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x) {
return dummy_application::serialize(sys, x);
}
template <class Parent>
error init(Parent&) {
return none;
}
application_type make() const {
return dummy_application{};
}
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(doorman_tests, fixture)
CAF_TEST(doorman accept) {
auto acceptor = unbox(make_tcp_accept_socket(auth, false));
auto port = unbox(local_port(socket_cast<network_socket>(acceptor)));
auto acceptor_guard = make_socket_guard(acceptor);
CAF_MESSAGE("opened acceptor on port " << port);
auto mgr = make_endpoint_manager(
mpx, sys,
doorman<dummy_application_factory>{acceptor_guard.release(),
dummy_application_factory{}});
CAF_CHECK_EQUAL(mgr->init(), none);
auto before = mpx->num_socket_managers();
CAF_CHECK_EQUAL(before, 2u);
uri::authority_type dst;
dst.port = port;
dst.host = "localhost"s;
CAF_MESSAGE("connecting to doorman on: " << dst);
auto conn = make_socket_guard(unbox(make_connected_tcp_stream_socket(dst)));
CAF_MESSAGE("waiting for connection");
while (mpx->num_socket_managers() != before + 1)
run();
CAF_MESSAGE("connected");
}
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