You need to sign in or sign up before continuing.
Commit d14838a8 authored by Dominik Charousset's avatar Dominik Charousset

Fix handling of write_packet calls

parent 290fc0c5
...@@ -71,7 +71,7 @@ public: ...@@ -71,7 +71,7 @@ public:
return err; return err;
auto hdr = to_bytes(header{message_type::handshake, auto hdr = to_bytes(header{message_type::handshake,
static_cast<uint32_t>(buf_.size()), version}); static_cast<uint32_t>(buf_.size()), version});
parent.write_packet(parent, hdr, buf_, unit); parent.write_packet(hdr, buf_);
return none; return none;
} }
...@@ -98,13 +98,14 @@ public: ...@@ -98,13 +98,14 @@ public:
header hdr{message_type::actor_message, static_cast<uint32_t>(buf_.size()), header hdr{message_type::actor_message, static_cast<uint32_t>(buf_.size()),
ptr->msg->mid.integer_value()}; ptr->msg->mid.integer_value()};
auto bytes = to_bytes(hdr); auto bytes = to_bytes(hdr);
parent.write_packet(parent, make_span(bytes), make_span(buf_), unit); parent.write_packet(make_span(bytes), make_span(buf_));
return none;
} }
template <class Parent> template <class Parent>
error handle_data(Parent& parent, byte_span bytes) { error handle_data(Parent& parent, byte_span bytes) {
auto write_packet = make_callback([&](byte_span hdr, byte_span payload) { auto write_packet = make_callback([&](byte_span hdr, byte_span payload) {
parent.write_packet(parent, hdr, payload, unit); parent.write_packet(hdr, payload);
return none; return none;
}); });
return handle(write_packet, bytes); return handle(write_packet, bytes);
...@@ -113,7 +114,7 @@ public: ...@@ -113,7 +114,7 @@ public:
template <class Parent> template <class Parent>
void resolve(Parent& parent, string_view path, actor listener) { void resolve(Parent& parent, string_view path, actor listener) {
auto write_packet = make_callback([&](byte_span hdr, byte_span payload) { auto write_packet = make_callback([&](byte_span hdr, byte_span payload) {
parent.write_packet(parent, hdr, payload, unit); parent.write_packet(hdr, payload);
return none; return none;
}); });
resolve_remote_path(write_packet, path, listener); resolve_remote_path(write_packet, path, listener);
......
...@@ -18,6 +18,10 @@ ...@@ -18,6 +18,10 @@
#pragma once #pragma once
#include "caf/abstract_actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp"
#include "caf/atom.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
namespace caf { namespace caf {
...@@ -56,6 +60,10 @@ public: ...@@ -56,6 +60,10 @@ public:
return transport_; return transport_;
} }
endpoint_manager_impl& manager() {
return *this;
}
// -- timeout management ----------------------------------------------------- // -- timeout management -----------------------------------------------------
template <class... Ts> template <class... Ts>
......
...@@ -41,6 +41,10 @@ public: ...@@ -41,6 +41,10 @@ public:
using application_type = Application; using application_type = Application;
using transport_type = stream_transport;
using worker_type = transport_worker<application_type>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket handle, application_type application) stream_transport(stream_socket handle, application_type application)
...@@ -51,7 +55,8 @@ public: ...@@ -51,7 +55,8 @@ public:
collected_(0), collected_(0),
max_(1024), max_(1024),
rd_flag_(net::receive_policy_flag::exactly), rd_flag_(net::receive_policy_flag::exactly),
written_(0) { written_(0),
manager_(nullptr) {
// nop // nop
} }
...@@ -61,18 +66,35 @@ public: ...@@ -61,18 +66,35 @@ public:
return handle_; return handle_;
} }
actor_system& system() {
return manager().system();
}
application_type& application() {
return worker_.application();
}
transport_type& transport() {
return *this;
}
endpoint_manager& manager() {
return *manager_;
}
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
if (auto err = worker_.init(parent)) manager_ = &parent;
if (auto err = worker_.init(*this))
return err; return err;
parent.mask_add(operation::read); parent.mask_add(operation::read);
return none; return none;
} }
template <class Parent> template <class Parent>
bool handle_read_event(Parent& parent) { bool handle_read_event(Parent&) {
auto buf = read_buf_.data() + collected_; auto buf = read_buf_.data() + collected_;
size_t len = read_threshold_ - collected_; size_t len = read_threshold_ - collected_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len)); CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len));
...@@ -82,8 +104,7 @@ public: ...@@ -82,8 +104,7 @@ public:
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes)); CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
collected_ += *num_bytes; collected_ += *num_bytes;
if (collected_ >= read_threshold_) { if (collected_ >= read_threshold_) {
auto decorator = make_write_packet_decorator(*this, parent); worker_.handle_data(*this, read_buf_);
worker_.handle_data(decorator, read_buf_);
prepare_next_read(); prepare_next_read();
} }
return true; return true;
...@@ -103,16 +124,15 @@ public: ...@@ -103,16 +124,15 @@ public:
// TODO: dont read all messages at once - get one by one. // TODO: dont read all messages at once - get one by one.
for (auto msg = parent.next_message(); msg != nullptr; for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) { msg = parent.next_message()) {
auto decorator = make_write_packet_decorator(*this, parent); worker_.write_message(*this, std::move(msg));
worker_.write_message(decorator, std::move(msg));
} }
// Write prepared data. // Write prepared data.
return write_some(); return write_some();
} }
template <class Parent> template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) { void resolve(Parent&, const std::string& path, actor listener) {
worker_.resolve(parent, path, listener); worker_.resolve(*this, path, listener);
} }
template <class... Ts> template <class... Ts>
...@@ -121,8 +141,8 @@ public: ...@@ -121,8 +141,8 @@ public:
} }
template <class Parent> template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) { void timeout(Parent&, atom_value value, uint64_t id) {
worker_.timeout(parent, value, id); worker_.timeout(*this, value, id);
} }
void handle_error(sec code) { void handle_error(sec code) {
...@@ -163,9 +183,8 @@ public: ...@@ -163,9 +183,8 @@ public:
prepare_next_read(); prepare_next_read();
} }
template <class Parent> void write_packet(span<const byte> header, span<const byte> payload,
void write_packet(Parent&, span<const byte> header, span<const byte> payload, typename worker_type::id_type) {
unit_t) {
write_buf_.insert(write_buf_.end(), header.begin(), header.end()); write_buf_.insert(write_buf_.end(), header.begin(), header.end());
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end()); write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
} }
...@@ -198,7 +217,7 @@ private: ...@@ -198,7 +217,7 @@ private:
return true; return true;
} }
transport_worker<application_type> worker_; worker_type worker_;
stream_socket handle_; stream_socket handle_;
std::vector<byte> read_buf_; std::vector<byte> read_buf_;
...@@ -212,6 +231,8 @@ private: ...@@ -212,6 +231,8 @@ private:
receive_policy_flag rd_flag_; receive_policy_flag rd_flag_;
size_t written_; size_t written_;
endpoint_manager* manager_;
}; };
} // namespace net } // namespace net
......
...@@ -46,16 +46,32 @@ public: ...@@ -46,16 +46,32 @@ public:
// nop // nop
} }
// -- properties -------------------------------------------------------------
application_type& application() noexcept {
return application_;
}
const application_type& application() const noexcept {
return application_;
}
const id_type& id() const noexcept {
return id_;
}
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
return application_.init(parent); auto decorator = make_write_packet_decorator(*this, parent);
return application_.init(decorator);
} }
template <class Parent> template <class Parent>
void handle_data(Parent& parent, span<const byte> data) { void handle_data(Parent& parent, span<const byte> data) {
application_.handle_data(parent, data); auto decorator = make_write_packet_decorator(*this, parent);
application_.handle_data(decorator, data);
} }
template <class Parent> template <class Parent>
...@@ -65,20 +81,16 @@ public: ...@@ -65,20 +81,16 @@ public:
application_.write_message(decorator, std::move(msg)); application_.write_message(decorator, std::move(msg));
} }
template <class Parent>
void write_packet(Parent& parent, span<const byte> header,
span<const byte> payload) {
parent.write_packet(header, payload, id_);
}
template <class Parent> template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) { void resolve(Parent& parent, const std::string& path, actor listener) {
application_.resolve(parent, path, listener); auto decorator = make_write_packet_decorator(*this, parent);
application_.resolve(decorator, path, listener);
} }
template <class Parent> template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) { void timeout(Parent& parent, atom_value value, uint64_t id) {
application_.timeout(parent, value, id); auto decorator = make_write_packet_decorator(*this, parent);
application_.timeout(decorator, value, id);
} }
void handle_error(sec error) { void handle_error(sec error) {
......
...@@ -52,12 +52,17 @@ public: ...@@ -52,12 +52,17 @@ public:
return parent_.transport(); return parent_.transport();
} }
endpoint_manager& manager() {
return parent_.manager();
}
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class... Ts> template <class... Ts>
void write_packet(span<const byte> header, span<const byte> payload, void write_packet(span<const byte> header, span<const byte> payload,
Ts&&... xs) { Ts&&... xs) {
object_.write_packet(parent_, header, payload, std::forward<Ts>(xs)...); parent_.write_packet(header, payload, std::forward<Ts>(xs)...,
object_.id());
} }
void cancel_timeout(atom_value type, uint64_t id) { void cancel_timeout(atom_value type, uint64_t id) {
......
...@@ -64,8 +64,7 @@ struct fixture : test_coordinator_fixture<>, proxy_registry::backend { ...@@ -64,8 +64,7 @@ struct fixture : test_coordinator_fixture<>, proxy_registry::backend {
input = to_buf(xs...); input = to_buf(xs...);
} }
void write_packet(fixture&, span<const byte> hdr, span<const byte> payload, void write_packet(span<const byte> hdr, span<const byte> payload) {
unit_t) {
output.insert(output.end(), hdr.begin(), hdr.end()); output.insert(output.end(), hdr.begin(), hdr.end());
output.insert(output.end(), payload.begin(), payload.end()); output.insert(output.end(), payload.begin(), payload.end());
} }
......
...@@ -84,15 +84,17 @@ public: ...@@ -84,15 +84,17 @@ public:
rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end()); rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end());
} }
template <class Manager> template <class Parent>
void resolve(Manager& manager, const std::string& path, actor listener) { void resolve(Parent& parent, const std::string& path, actor listener) {
actor_id aid = 42; actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899"; auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(42, hid)); auto nid = unbox(make_node_id(42, hid));
actor_config cfg; actor_config cfg;
endpoint_manager_ptr ptr{&parent.manager()};
auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid, auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid,
&manager.system(), &parent.system(),
cfg, &manager); cfg,
std::move(ptr));
anon_send(listener, resolve_atom::value, std::move(path), p); anon_send(listener, resolve_atom::value, std::move(path), p);
} }
......
...@@ -155,16 +155,16 @@ public: ...@@ -155,16 +155,16 @@ public:
} }
} }
template <class Manager> template <class Parent>
void resolve(Manager& manager, const std::string& path, actor listener) { void resolve(Parent& parent, const std::string& path, actor listener) {
actor_id aid = 42; actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899"; auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(aid, hid)); auto nid = unbox(make_node_id(aid, hid));
actor_config cfg; actor_config cfg;
auto p = make_actor<net::actor_proxy_impl, strong_actor_ptr>(aid, nid, auto sys = &parent.system();
&manager auto mgr = &parent.manager();
.system(), auto p = make_actor<net::actor_proxy_impl, strong_actor_ptr>(aid, nid, sys,
cfg, &manager); cfg, mgr);
anon_send(listener, resolve_atom::value, std::move(path), p); anon_send(listener, resolve_atom::value, std::move(path), p);
} }
......
...@@ -70,10 +70,10 @@ public: ...@@ -70,10 +70,10 @@ public:
return none; return none;
} }
template <class Transport> template <class Parent>
void write_message(Transport& transport, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager::message> msg) { std::unique_ptr<endpoint_manager::message> msg) {
transport.write_packet(span<byte>{}, msg->payload); parent.write_packet(span<const byte>{}, msg->payload);
} }
template <class Parent> template <class Parent>
......
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