Commit 2614f57e authored by Joseph Noir's avatar Joseph Noir

Add raw TCP example

parent 8532a64a
...@@ -51,8 +51,9 @@ add(remoting group_chat) ...@@ -51,8 +51,9 @@ add(remoting group_chat)
add(remoting group_server) add(remoting group_server)
add(remoting remote_spawn) add(remoting remote_spawn)
add(remoting distributed_calculator) add(remoting distributed_calculator)
add(remoting udp_newb) add(remoting basp_udp_newb)
add(remoting tcp_newb) add(remoting basp_tcp_newb)
add(remoting raw_tcp_newb)
# basic I/O with brokers # basic I/O with brokers
add(broker simple_broker) add(broker simple_broker)
......
#include "caf/io/network/newb.hpp" #include "caf/io/network/newb.hpp"
#include "caf/config.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
...@@ -13,7 +12,6 @@ using namespace caf; ...@@ -13,7 +12,6 @@ using namespace caf;
using io::network::native_socket; using io::network::native_socket;
using io::network::invalid_native_socket; using io::network::invalid_native_socket;
using io::network::default_multiplexer; using io::network::default_multiplexer;
using io::network::last_socket_error_as_string;
namespace { namespace {
...@@ -120,35 +118,6 @@ struct basp { ...@@ -120,35 +118,6 @@ struct basp {
} }
}; };
template <class T>
struct tcp_protocol
: public io::network::protocol_policy<typename T::message_type> {
T impl;
tcp_protocol(io::network::newb<typename T::message_type>* parent)
: impl(parent) {
// nop
}
error read(char* bytes, size_t count) override {
return impl.read(bytes, count);
}
error timeout(atom_value atm, uint32_t id) override {
return impl.timeout(atm, id);
}
void write_header(io::network::byte_buffer& buf,
io::network::header_writer* hw) override {
impl.write_header(buf, hw);
}
void prepare_for_sending(io::network::byte_buffer& buf, size_t hstart,
size_t offset, size_t plen) override {
impl.prepare_for_sending(buf, hstart, offset, plen);
}
};
struct basp_newb : public io::network::newb<new_basp_message> { struct basp_newb : public io::network::newb<new_basp_message> {
basp_newb(caf::actor_config& cfg, default_multiplexer& dm, basp_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd) native_socket sockfd)
...@@ -217,20 +186,12 @@ struct tcp_acceptor ...@@ -217,20 +186,12 @@ struct tcp_acceptor
tcp_acceptor(default_multiplexer& dm, native_socket sockfd) tcp_acceptor(default_multiplexer& dm, native_socket sockfd)
: super(dm, sockfd) { : super(dm, sockfd) {
CAF_LOG_TRACE("");
std::cerr << "constructing newb acceptor" << std::endl;
// nop // nop
} }
~tcp_acceptor() {
CAF_LOG_TRACE("");
std::cerr << "terminating newb acceptor" << std::endl;
}
expected<actor> create_newb(native_socket sockfd, expected<actor> create_newb(native_socket sockfd,
io::network::transport_policy_ptr pol) override { io::network::transport_policy_ptr pol) override {
CAF_LOG_TRACE(CAF_ARG(sockfd)); CAF_LOG_TRACE(CAF_ARG(sockfd));
std::cerr << "acceptor creating new newb" << std::endl;
auto n = io::network::make_newb<basp_newb>(this->backend().system(), sockfd); auto n = io::network::make_newb<basp_newb>(this->backend().system(), sockfd);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n); auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
if (ptr == nullptr) if (ptr == nullptr)
...@@ -255,11 +216,12 @@ struct tcp_test_broker_state { ...@@ -255,11 +216,12 @@ struct tcp_test_broker_state {
}; };
void caf_main(actor_system& sys, const actor_system_config&) { void caf_main(actor_system& sys, const actor_system_config&) {
using acceptor_t = tcp_acceptor<tcp_protocol<basp>>;
using caf::io::network::make_server_newb; using caf::io::network::make_server_newb;
using caf::io::network::make_client_newb; using caf::io::network::make_client_newb;
using caf::policy::accept_tcp; using caf::policy::accept_tcp;
using caf::policy::tcp_transport; using caf::policy::tcp_transport;
using caf::policy::tcp_protocol;
using acceptor_t = tcp_acceptor<tcp_protocol<basp>>;
const char* host = "localhost"; const char* host = "localhost";
const uint16_t port = 12345; const uint16_t port = 12345;
scoped_actor self{sys}; scoped_actor self{sys};
......
#include "caf/io/network/newb.hpp"
#include "caf/logger.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp"
#include "caf/policy/newb_tcp.hpp"
using namespace caf;
using io::network::native_socket;
using io::network::invalid_native_socket;
using io::network::default_multiplexer;
namespace {
using ordering_atom = atom_constant<atom("ordering")>;
using send_atom = atom_constant<atom("send")>;
using quit_atom = atom_constant<atom("quit")>;
using responder_atom = atom_constant<atom("responder")>;
struct new_data {
char* payload;
size_t payload_len;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, new_data& data) {
return fun(meta::type_name("new_tcp_data"), data.payload_len);
}
struct raw_tcp {
using message_type = new_data;
using result_type = optional<message_type>;
io::network::newb<message_type>* parent;
message_type msg;
raw_tcp(io::network::newb<message_type>* parent) : parent(parent) {
// nop
}
error read(char* bytes, size_t count) {
msg.payload = bytes;
msg.payload_len = count;
parent->handle(msg);
return none;
}
error timeout(atom_value, uint32_t) {
return none;
}
size_t write_header(io::network::byte_buffer&,
io::network::header_writer*) {
return 0;
}
void prepare_for_sending(io::network::byte_buffer&, size_t, size_t, size_t) {
// nop
}
};
struct raw_newb : public io::network::newb<new_data> {
using message_type = new_data;
raw_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd)
: newb<message_type>(cfg, dm, sockfd) {
// nop
CAF_LOG_TRACE("");
std::cerr << "constructing newb" << std::endl;
}
~raw_newb() {
std::cerr << "terminating newb" << std::endl;
CAF_LOG_TRACE("");
}
void handle(message_type& msg) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
std::string res;
binary_deserializer bd(&backend(), msg.payload, msg.payload_len);
bd(res);
send(responder, res);
}
behavior make_behavior() override {
set_default_handler(print_and_drop);
return {
// Must be implemented at the moment, will be cought by the broker in a
// later implementation.
[=](atom_value atm, uint32_t id) {
protocol->timeout(atm, id);
},
[=](send_atom, std::string payload) {
auto whdl = wr_buf(nullptr);
CAF_ASSERT(whdl.buf != nullptr);
CAF_ASSERT(whdl.protocol != nullptr);
binary_serializer bs(&backend(), *whdl.buf);
auto from = whdl.buf->size();
whdl.buf->resize(1000);
std::fill(whdl.buf->begin() + from, whdl.buf->end(), 0);
bs(payload);
},
[=](responder_atom, actor r) {
aout(this) << "got responder assigned" << std::endl;
responder = r;
send(r, this);
},
[=](quit_atom) {
aout(this) << "got quit message" << std::endl;
// Remove from multiplexer loop.
stop();
// Quit actor.
quit();
}
};
}
actor responder;
};
template <class ProtocolPolicy>
struct tcp_acceptor
: public io::network::newb_acceptor<typename ProtocolPolicy::message_type> {
using super = io::network::newb_acceptor<typename ProtocolPolicy::message_type>;
tcp_acceptor(default_multiplexer& dm, native_socket sockfd)
: super(dm, sockfd) {
// nop
}
expected<actor> create_newb(native_socket sockfd,
io::network::transport_policy_ptr pol) override {
CAF_LOG_TRACE(CAF_ARG(sockfd));
auto n = io::network::make_newb<raw_newb>(this->backend().system(), sockfd);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
if (ptr == nullptr)
return sec::runtime_error;
auto& ref = dynamic_cast<raw_newb&>(*ptr);
// TODO: Transport has to be assigned before protocol ... which sucks.
// (basp protocol calls configure read which accesses the transport.)
ref.transport = std::move(pol);
ref.protocol.reset(new ProtocolPolicy(&ref));
ref.responder = responder;
ref.configure_read(io::receive_policy::exactly(1000));
// TODO: Just a workaround.
anon_send(responder, n);
return n;
}
actor responder;
};
void caf_main(actor_system& sys, const actor_system_config&) {
using caf::io::network::make_server_newb;
using caf::io::network::make_client_newb;
using caf::policy::accept_tcp;
using caf::policy::tcp_transport;
using caf::policy::tcp_protocol;
using acceptor_t = tcp_acceptor<tcp_protocol<raw_tcp>>;
const char* host = "localhost";
const uint16_t port = 12345;
scoped_actor self{sys};
auto running = [=](event_based_actor* self, std::string name,
actor, actor b) -> behavior {
return {
[=](std::string str) {
aout(self) << "[" << name << "] received '" << str << "'" << std::endl;
},
[=](send_atom, std::string str) {
aout(self) << "[" << name << "] sending '" << str << "'" << std::endl;
self->send(b, send_atom::value, self->id(), actor_id{}, str);
},
};
};
auto init = [=](event_based_actor* self, std::string name,
actor m) -> behavior {
self->set_default_handler(skip);
return {
[=](actor b) {
aout(self) << "[" << name << "] got broker, let's do this" << std::endl;
self->become(running(self, name, m, b));
self->set_default_handler(print_and_drop);
}
};
};
auto server_helper = sys.spawn(init, "s", self);
auto client_helper = sys.spawn(init, "c", self);
aout(self) << "creating new server" << std::endl;
auto server_ptr = make_server_newb<acceptor_t, accept_tcp>(sys, port, nullptr,
true);
server_ptr->responder = server_helper;
aout(self) << "creating new client" << std::endl;
auto client = make_client_newb<raw_newb, tcp_transport,
tcp_protocol<raw_tcp>>(sys, host, port);
self->send(client, responder_atom::value, client_helper);
self->send(client_helper, send_atom::value, "hallo");
self->send(server_helper, send_atom::value, "hallo");
self->receive(
[&](quit_atom) {
aout(self) << "check" << std::endl;
}
);
/*
main_actor->receive(
[](quit_atom) {
CAF_LOG_DEBUG("check");
}
);
CAF_LOG_DEBUG("shutting everything down");
newb_acceptor_ptr->stop();
anon_send(newb_actor, quit_atom::value);
anon_send(helper_actor, quit_atom::value);
anon_send(test_broker, quit_atom::value);
sys.await_all_actors_done();
CAF_LOG_DEBUG("done");
*/
}
} // namespace anonymous
CAF_MAIN(io::middleman);
...@@ -70,5 +70,34 @@ struct accept_tcp : public io::network::accept_policy { ...@@ -70,5 +70,34 @@ struct accept_tcp : public io::network::accept_policy {
void init(io::network::newb_base& n) override; void init(io::network::newb_base& n) override;
}; };
template <class T>
struct tcp_protocol
: public io::network::protocol_policy<typename T::message_type> {
T impl;
tcp_protocol(io::network::newb<typename T::message_type>* parent)
: impl(parent) {
// nop
}
error read(char* bytes, size_t count) override {
return impl.read(bytes, count);
}
error timeout(atom_value atm, uint32_t id) override {
return impl.timeout(atm, id);
}
void write_header(io::network::byte_buffer& buf,
io::network::header_writer* hw) override {
impl.write_header(buf, hw);
}
void prepare_for_sending(io::network::byte_buffer& buf, size_t hstart,
size_t offset, size_t plen) override {
impl.prepare_for_sending(buf, hstart, offset, plen);
}
};
} // namespace policy } // namespace policy
} // namespace caf } // namespace caf
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