Commit 958471aa authored by Joseph Noir's avatar Joseph Noir

In transition from sink/source to endpoint

parent d1324ce7
...@@ -112,29 +112,22 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -112,29 +112,22 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// stores meta information for dgram endpoints // stores meta information for dgram endpoints
struct endpoint_context { struct endpoint_context {
// stores information about remote datagram enpoints
struct remote_endpoint {
// local sink
datagram_sink_handle sink;
// ordering information
uint32_t sequence_number_send = 0;
uint32_t sequence_number_receive = 0;
// TODO: local buffer for ordering
uint16_t remote_port;
};
// our current processed BSAP header // our current processed BSAP header
basp::header hdr; basp::header hdr;
// local source // local source
datagram_source_handle source; endpoint_handle hdl;
// network-agnositc node identifier // network-agnositc node identifier
node_id id; node_id id;
// ports
uint16_t remote_port;
uint16_t local_port;
// pending operations to be performed after handshake completed // pending operations to be performed after handshake completed
optional<response_promise> callback; optional<response_promise> callback;
// TODO: reliability things // TODO: ordering and reliability things
}; };
void set_context(connection_handle hdl); void set_context(connection_handle hdl);
void set_context(datagram_source_handle hdl); void set_context(endpoint_handle hdl);
// pointer to ourselves // pointer to ourselves
broker* self; broker* self;
...@@ -144,10 +137,11 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -144,10 +137,11 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// keeps context information for all open connections // keeps context information for all open connections
std::unordered_map<connection_handle, connection_context> tcp_ctx; std::unordered_map<connection_handle, connection_context> tcp_ctx;
std::unordered_map<datagram_sink_handle, endpoint_context> udp_ctx; std::unordered_map<endpoint_handle, endpoint_context> udp_ctx;
// points to the current context for callbacks such as `make_proxy` // points to the current context for callbacks such as `make_proxy`
connection_context* this_context = nullptr; connection_context* this_context = nullptr;
endpoint_context* udp_context = nullptr;
// stores handles to spawn servers for other nodes; these servers // stores handles to spawn servers for other nodes; these servers
// are spawned whenever the broker learns a new node ID and try to // are spawned whenever the broker learns a new node ID and try to
......
...@@ -465,6 +465,21 @@ void basp_broker_state::set_context(connection_handle hdl) { ...@@ -465,6 +465,21 @@ void basp_broker_state::set_context(connection_handle hdl) {
this_context = &i->second; this_context = &i->second;
} }
void basp_broker_state::set_context(endpoint_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
auto i = udp_ctx.find(hdl);
if (i == udp_ctx.end()) {
CAF_LOG_INFO("create new BASP context:" << CAF_ARG(hdl));
i = udp_ctx.emplace(hdl,
endpoint_context{
basp::header{basp::message_type::client_handshake, 0,
0, 0, none, none,
invalid_actor_id, invalid_actor_id},
hdl, none, 0, 0, none}).first;
}
udp_context = &i->second;
}
/****************************************************************************** /******************************************************************************
* basp_broker * * basp_broker *
******************************************************************************/ ******************************************************************************/
...@@ -525,6 +540,9 @@ behavior basp_broker::make_behavior() { ...@@ -525,6 +540,9 @@ behavior basp_broker::make_behavior() {
[=](datagram_msg& msg) { [=](datagram_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
CAF_LOG_DEBUG("Received new_datagram_msg: " << CAF_ARG(msg)); CAF_LOG_DEBUG("Received new_datagram_msg: " << CAF_ARG(msg));
state.set_context(msg.handle);
// auto& ctx = *state.udp_context;
//auto next = state.instance.handle(context(), msg, ctx.hdr, ...);
// TODO: implement this // TODO: implement this
// look for existing context or create a new one // look for existing context or create a new one
// handle messge // handle messge
......
...@@ -165,7 +165,8 @@ public: ...@@ -165,7 +165,8 @@ public:
} else if (std::distance(u.scheme().first, u.scheme().second) >= 3 && } else if (std::distance(u.scheme().first, u.scheme().second) >= 3 &&
equal(std::begin(udp), std::end(udp), u.scheme().first)) { equal(std::begin(udp), std::end(udp), u.scheme().first)) {
// connect to endpoint and initiate handhsake etc. (UDP) // connect to endpoint and initiate handhsake etc. (UDP)
auto y = system().middleman().backend().new_datagram_sink(host, port); auto y
= system().middleman().backend().new_remote_endpoint(host, port);
if (!y) { if (!y) {
rp.deliver(std::move(y.error())); rp.deliver(std::move(y.error()));
return {}; return {};
...@@ -256,11 +257,11 @@ private: ...@@ -256,11 +257,11 @@ private:
equal(u.scheme().first, u.scheme().second, std::begin(udp))) { equal(u.scheme().first, u.scheme().second, std::begin(udp))) {
// UDP // UDP
auto res auto res
= system().middleman().backend().new_datagram_source(u.port_as_int(), = system().middleman().backend().new_local_endpoint(u.port_as_int(),
in, reuse_addr); in, reuse_addr);
if (!res) if (!res)
return std::move(res.error()); return std::move(res.error());
datagram_source_handle hdl = res->first; endpoint_handle hdl = res->first;
actual_port = res->second; actual_port = res->second;
anon_send(broker_, publish_atom::value, hdl, actual_port, anon_send(broker_, publish_atom::value, hdl, actual_port,
std::move(whom), std::move(sigs)); std::move(whom), std::move(sigs));
......
...@@ -73,33 +73,33 @@ behavior make_pong_behavior() { ...@@ -73,33 +73,33 @@ behavior make_pong_behavior() {
CAF_TEST_FIXTURE_SCOPE(datagrams, fixture) CAF_TEST_FIXTURE_SCOPE(datagrams, fixture)
CAF_TEST(test_datagram_sinks) { CAF_TEST(test_remote_endpoint) {
auto& mp = client_side_mm.backend(); auto& mp = client_side_mm.backend();
auto hdl = client_side_mm.named_broker<basp_broker>(atom("BASP")); auto hdl = client_side_mm.named_broker<basp_broker>(atom("BASP"));
auto basp = static_cast<basp_broker*>(actor_cast<abstract_actor*>(hdl)); auto basp = static_cast<basp_broker*>(actor_cast<abstract_actor*>(hdl));
CAF_MESSAGE("Calling new_datagram_sink"); CAF_MESSAGE("Calling new_remote_endpoint");
auto res1 = mp.new_datagram_sink(host, port + 0); auto res1 = mp.new_remote_endpoint(host, port + 0);
CAF_REQUIRE(res1); CAF_REQUIRE(res1);
CAF_MESSAGE("Calling assign_datagram_sink"); CAF_MESSAGE("Calling assign_endpoint (on \"remote\" endpoint).");
auto res2 = mp.assign_datagram_sink(basp, *res1); auto res2 = mp.assign_endpoint(basp, *res1);
CAF_REQUIRE(res2); CAF_REQUIRE(res2);
CAF_MESSAGE("Calling add_datagram_sink"); CAF_MESSAGE("Calling add_remote_endpoint.");
auto res3 = mp.add_datagram_sink(basp, host, port + 1); auto res3 = mp.add_remote_endpoint(basp, host, port + 1);
CAF_REQUIRE(res3); CAF_REQUIRE(res3);
} }
CAF_TEST(test_datagram_sources) { CAF_TEST(test_local_endpoint) {
auto& mp = client_side_mm.backend(); auto& mp = client_side_mm.backend();
auto hdl = client_side_mm.named_broker<basp_broker>(atom("BASP")); auto hdl = client_side_mm.named_broker<basp_broker>(atom("BASP"));
auto basp = static_cast<basp_broker*>(actor_cast<abstract_actor*>(hdl)); auto basp = static_cast<basp_broker*>(actor_cast<abstract_actor*>(hdl));
CAF_MESSAGE("Calling new_datagram_source"); CAF_MESSAGE("Calling new_local_endpoint.");
auto res1 = mp.new_datagram_source(port + 0); auto res1 = mp.new_local_endpoint(port + 0);
CAF_REQUIRE(res1); CAF_REQUIRE(res1);
CAF_MESSAGE("Calling assign_datagram_source"); CAF_MESSAGE("Calling assign_endpoint (on \"local\" endpoint).");
auto res2 = mp.assign_datagram_source(basp, res1->first); auto res2 = mp.assign_endpoint(basp, res1->first);
CAF_REQUIRE(res2); CAF_REQUIRE(res2);
CAF_MESSAGE("Calling add_datagram_source"); CAF_MESSAGE("Calling add_local_endpoint.");
auto res3 = mp.add_datagram_source(basp, port + 1, nullptr); auto res3 = mp.add_local_endpoint(basp, port + 1, nullptr);
CAF_REQUIRE(res3); CAF_REQUIRE(res3);
} }
......
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