Commit 8c04c830 authored by Joseph Noir's avatar Joseph Noir

Add test for newb tcp impl

parent 7b895416
......@@ -50,6 +50,8 @@ namespace io {
// -- atoms for the acceptor ---------------------------------------------------
using children_atom = caf::atom_constant<atom("childern")>;
using port_atom = caf::atom_constant<atom("port")>;
using quit_atom = caf::atom_constant<atom("quit")>;
// -- forward declarations -----------------------------------------------------
......@@ -436,7 +438,9 @@ struct newb_acceptor : network::newb_base {
CAF_LOG_ERROR("Creating newb with invalid socket");
}
newb_acceptor(const newb_acceptor& other) = delete;
newb_acceptor() = default;
newb_acceptor(newb_acceptor&& other) = default;
~newb_acceptor() {
// nop
......@@ -589,6 +593,12 @@ struct newb_acceptor : network::newb_base {
[=](quit_atom) {
stop();
},
[=](children_atom) {
return children_;
},
[=](port_atom) {
return network::local_port_of_fd(fd_);
},
[=](caf::exit_msg& msg) {
auto itr = std::find(std::begin(children_), std::end(children_),
msg.source);
......@@ -629,7 +639,11 @@ template <class Protocol, spawn_options Os = no_spawn_options, class Fun,
infer_handle_from_class_t<newb_acceptor<Protocol, Fun, Ts...>>
spawn_acceptor(actor_system& sys, Fun fun, policy::accept_ptr<Message> pol,
network::native_socket sockfd, Ts&&... xs) {
// TODO: check that fun accepts Message.
using first = first_argument_type<Fun>;
using message = typename std::remove_pointer<first>::type::message_type;
static_assert(std::is_same<message, Message>::value,
"Fun must accept a message type matching the protocol");
// TODO: Can we also check that the signature of Fun ends in Ts...?
using acceptor_type = newb_acceptor<Protocol, Fun, Ts...>;
auto& dm =
dynamic_cast<network::default_multiplexer&>(sys.middleman().backend());
......@@ -648,24 +662,24 @@ spawn_acceptor(actor_system& sys, Fun fun, policy::accept_ptr<Message> pol,
}
template <class Protocol, class F, class Message, class... Ts>
expected<caf::intrusive_ptr<newb_acceptor<Protocol, F, Ts...>>>
expected<infer_handle_from_class_t<newb_acceptor<Protocol, F, Ts...>>>
spawn_server(actor_system& sys, F fun, policy::accept_ptr<Message> pol,
uint16_t port, const char* addr, bool reuse, Ts&&... xs) {
uint16_t port, const char* addr, bool reuse, Ts&&... xs) {
auto esock = pol->create_socket(port, addr, reuse);
if (!esock) {
CAF_LOG_ERROR("Could not open " << CAF_ARG(port) << CAF_ARG(addr));
return sec::cannot_open_port;
}
return spawn_acceptor<Protocol, F>(sys, std::move(fun), std::move(pol), *esock,
std::forward<Ts>(xs)...);
return spawn_acceptor<Protocol>(sys, std::move(fun), std::move(pol), *esock,
std::forward<Ts>(xs)...);
}
template <class Protocol, class F, class Message>
expected<caf::intrusive_ptr<newb_acceptor<Protocol, F>>>
expected<infer_handle_from_class_t<newb_acceptor<Protocol, F>>>
spawn_server(actor_system& sys, F fun, policy::accept_ptr<Message> pol,
uint16_t port, const char* addr = nullptr, bool reuse = false) {
return spawn_server<Protocol, F>(sys, std::move(fun), std::move(pol), port,
addr, reuse);
uint16_t port) {
return spawn_server<Protocol>(sys, std::move(fun), std::move(pol), port,
nullptr, false);
}
/*
......@@ -710,4 +724,3 @@ make_server(actor_system& sys, F fun, policy::accept_ptr<Message> pol,
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_newb_tcp
#include "caf/config.hpp"
#include "caf/test/dsl.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/io/newb.hpp"
#include "caf/policy/newb_raw.hpp"
#include "caf/policy/newb_tcp.hpp"
using namespace caf;
using namespace caf::io;
using namespace caf::policy;
namespace {
using alive_atom = atom_constant<atom("alive")>;
using start_atom = atom_constant<atom("start")>;
using shutdown_atom = atom_constant<atom("shutdown")>;
constexpr auto host = "127.0.0.1";
// behavior tcp_server(newb<new_raw_msg>* self, actor responder) {
// self->send(responder, alive_atom::value);
// self->configure_read(io::receive_policy::exactly(sizeof(uint32_t)));
// return {
// [=](new_raw_msg& msg) {
// // Read message.
// uint32_t data;
// binary_deserializer bd(self->system(), msg.payload, msg.payload_len);
// bd(data);
// // Write message.
// auto whdl = self->wr_buf(nullptr);
// binary_serializer bs(&self->backend(), *whdl.buf);
// bs(data + 1);
// },
// [=](io_error_msg& msg) {
// CAF_FAIL("server got io error: " << to_string(msg.op));
// self->quit();
// self->stop();
// },
// [=](caf::exit_msg&) {
// CAF_MESSAGE("parent shut down, doing the same");
// self->stop();
// self->quit();
// self->send(responder, shutdown_atom::value);
// }
// };
// }
behavior tcp_server(newb<new_raw_msg>* self) {
self->configure_read(io::receive_policy::exactly(sizeof(uint32_t)));
return {
[=](new_raw_msg& msg) {
// Read message.
uint32_t data;
binary_deserializer bd(self->system(), msg.payload, msg.payload_len);
bd(data);
CAF_MESSAGE("server got message from client: " << data);
// Write message.
auto whdl = self->wr_buf(nullptr);
binary_serializer bs(&self->backend(), *whdl.buf);
bs(data + 1);
},
[=](io_error_msg& msg) {
//CAF_FAIL("server got io error: " << to_string(msg.op));
CAF_MESSAGE("server: connection lost");
self->quit();
self->stop();
},
[=](caf::exit_msg&) {
CAF_MESSAGE("parent shut down, doing the same");
self->stop();
self->quit();
}
};
}
behavior tcp_client(newb<new_raw_msg>* self, uint32_t value) {
self->configure_read(io::receive_policy::exactly(sizeof(uint32_t)));
auto whdl = self->wr_buf(nullptr);
binary_serializer bs(&self->backend(), *whdl.buf);
bs(value);
return {
[=](new_raw_msg& msg) {
CAF_MESSAGE("client received answer from server");
uint32_t response;
binary_deserializer bd(self->system(), msg.payload, msg.payload_len);
bd(response);
CAF_CHECK_EQUAL(response, value + 1);
self->stop();
self->quit();
},
[=](io_error_msg& msg) {
// CAF_FAIL("client got io error: " << to_string(msg.op));
// self->stop();
// self->quit();
// self->send(responder, quit_atom::value);
CAF_MESSAGE("client: connection lost");
self->stop();
self->quit();
}
};
}
} // namespace anonymous
CAF_TEST(newb tcp communication) {
actor_system_config config;
config.load<io::middleman>();
actor_system system{config};
{
scoped_actor self{system};
// Create acceptor.
accept_ptr<policy::new_raw_msg> pol{new accept_tcp<policy::new_raw_msg>};
auto eserver = io::spawn_server<tcp_protocol<raw>>(system, tcp_server,
std::move(pol), 0,
nullptr, true);
if (!eserver)
CAF_FAIL("failed to start server: " << system.render(eserver.error()));
uint16_t port;
self->send(*eserver, port_atom::value);
self->receive(
[&](uint16_t published_on) {
port = published_on;
CAF_MESSAGE("server listening on port " << port);
}
);
transport_ptr transport{new tcp_transport};
auto eclient = io::spawn_client<tcp_protocol<raw>>(system, tcp_client,
std::move(transport),
host, port, 23);
if (!eclient)
CAF_FAIL("failed to start client: " << system.render(eclient.error()));
}
system.await_all_actors_done();
}
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