Commit dff48ceb authored by Jakob Otto's avatar Jakob Otto

Add packet_writer_interface

parent 01c487f1
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/byte.hpp"
#include "caf/net/fwd.hpp"
#include "caf/span.hpp"
namespace caf {
namespace net {
class packet_writer {
public:
using buffer_type = std::vector<byte>;
// virtual ~packet_writer() = default;
virtual buffer_type next_header_buffer() = 0;
virtual buffer_type next_buffer() = 0;
// TODO: some documentation
// First buffer is header.
// Following buffers are one or more payload buffers for this packet.
/// @warning takes ownership of `buffers`.
template <class... Ts>
void write_packet(Ts... buffers) {
buffer_type* bufs[] = {&buffers...};
write_impl(make_span(bufs, sizeof...(Ts)));
}
protected:
virtual void write_impl(span<buffer_type*> buffers) = 0;
};
} // namespace net
} // namespace caf
...@@ -18,6 +18,8 @@ ...@@ -18,6 +18,8 @@
#pragma once #pragma once
#include "caf/net/packet_writer.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
...@@ -27,7 +29,7 @@ namespace net { ...@@ -27,7 +29,7 @@ namespace net {
/// Implements the interface for transport and application policies and /// Implements the interface for transport and application policies and
/// dispatches member functions either to `decorator` or `parent`. /// dispatches member functions either to `decorator` or `parent`.
template <class Object, class Parent> template <class Object, class Parent>
class write_packet_decorator { class packet_writer_impl final : public packet_writer {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -37,7 +39,7 @@ public: ...@@ -37,7 +39,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
write_packet_decorator(Object& object, Parent& parent) packet_writer_impl(Object& object, Parent& parent)
: object_(object), parent_(parent) { : object_(object), parent_(parent) {
// nop // nop
} }
...@@ -56,6 +58,14 @@ public: ...@@ -56,6 +58,14 @@ public:
return parent_.manager(); return parent_.manager();
} }
buffer_type next_header_buffer() override {
return transport().next_header_buffer();
}
buffer_type next_buffer() override {
return transport().next_buffer();
}
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class... Ts> template <class... Ts>
...@@ -74,14 +84,20 @@ public: ...@@ -74,14 +84,20 @@ public:
return parent_.set_timeout(tout, type, std::forward<Ts>(xs)...); return parent_.set_timeout(tout, type, std::forward<Ts>(xs)...);
} }
protected:
// TODO: this should replace the current `write_packet()`
void write_impl(span<buffer_type*>) override {
// parent_.write_packet(buffers);
}
private: private:
Object& object_; Object& object_;
Parent& parent_; Parent& parent_;
}; };
template <class Object, class Parent> template <class Object, class Parent>
write_packet_decorator<Object, Parent> packet_writer_impl<Object, Parent> make_packet_writer_impl(Object& object,
make_write_packet_decorator(Object& object, Parent& parent) { Parent& parent) {
return {object, parent}; return {object, parent};
} }
......
...@@ -45,6 +45,8 @@ public: ...@@ -45,6 +45,8 @@ public:
using worker_type = transport_worker<application_type>; using worker_type = transport_worker<application_type>;
using buffer_type = std::vector<byte>;
// -- 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)
...@@ -206,6 +208,26 @@ public: ...@@ -206,6 +208,26 @@ public:
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end()); write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
} }
// -- buffer recycling -------------------------------------------------------
buffer_type next_header_buffer() {
return next_buffer_impl(free_header_bufs_);
}
buffer_type next_buffer() {
return next_buffer_impl(free_bufs_);
}
buffer_type next_buffer_impl(std::deque<buffer_type>& container) {
if (container.empty()) {
return {};
} else {
auto buf = std::move(container.front());
container.pop_front();
return buf;
}
}
private: private:
// -- private member functions ----------------------------------------------- // -- private member functions -----------------------------------------------
...@@ -239,8 +261,11 @@ private: ...@@ -239,8 +261,11 @@ private:
worker_type worker_; worker_type worker_;
stream_socket handle_; stream_socket handle_;
std::vector<byte> read_buf_; std::deque<buffer_type> free_header_bufs_;
std::vector<byte> write_buf_; std::deque<buffer_type> free_bufs_;
buffer_type read_buf_;
buffer_type write_buf_;
// TODO implement retries using this member! // TODO implement retries using this member!
// size_t max_consecutive_reads_; // size_t max_consecutive_reads_;
......
...@@ -22,7 +22,7 @@ ...@@ -22,7 +22,7 @@
#include "caf/ip_endpoint.hpp" #include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/write_packet_decorator.hpp" #include "caf/net/packet_writer_impl.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/unit.hpp" #include "caf/unit.hpp"
...@@ -64,46 +64,46 @@ public: ...@@ -64,46 +64,46 @@ public:
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
auto decorator = make_write_packet_decorator(*this, parent); auto writer = make_packet_writer_impl(*this, parent);
return application_.init(decorator); return application_.init(writer);
} }
template <class Parent> template <class Parent>
error handle_data(Parent& parent, span<const byte> data) { error handle_data(Parent& parent, span<const byte> data) {
auto decorator = make_write_packet_decorator(*this, parent); auto writer = make_packet_writer_impl(*this, parent);
return application_.handle_data(decorator, data); return application_.handle_data(writer, data);
} }
template <class Parent> template <class Parent>
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) { std::unique_ptr<endpoint_manager_queue::message> msg) {
auto decorator = make_write_packet_decorator(*this, parent); auto writer = make_packet_writer_impl(*this, parent);
application_.write_message(decorator, std::move(msg)); application_.write_message(writer, std::move(msg));
} }
template <class Parent> template <class Parent>
void resolve(Parent& parent, string_view path, const actor& listener) { void resolve(Parent& parent, string_view path, const actor& listener) {
auto decorator = make_write_packet_decorator(*this, parent); auto writer = make_packet_writer_impl(*this, parent);
application_.resolve(decorator, path, listener); application_.resolve(writer, path, listener);
} }
template <class Parent> template <class Parent>
void new_proxy(Parent& parent, const node_id&, actor_id id) { void new_proxy(Parent& parent, const node_id&, actor_id id) {
auto decorator = make_write_packet_decorator(*this, parent); auto writer = make_packet_writer_impl(*this, parent);
application_.new_proxy(decorator, id); application_.new_proxy(writer, id);
} }
template <class Parent> template <class Parent>
void local_actor_down(Parent& parent, const node_id&, actor_id id, void local_actor_down(Parent& parent, const node_id&, actor_id id,
error reason) { error reason) {
auto decorator = make_write_packet_decorator(*this, parent); auto writer = make_packet_writer_impl(*this, parent);
application_.local_actor_down(decorator, id, std::move(reason)); application_.local_actor_down(writer, id, std::move(reason));
} }
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) {
auto decorator = make_write_packet_decorator(*this, parent); auto writer = make_packet_writer_impl(*this, parent);
application_.timeout(decorator, value, id); application_.timeout(writer, value, id);
} }
void handle_error(sec error) { void handle_error(sec error) {
......
...@@ -24,8 +24,8 @@ ...@@ -24,8 +24,8 @@
#include "caf/ip_endpoint.hpp" #include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_impl.hpp"
#include "caf/net/transport_worker.hpp" #include "caf/net/transport_worker.hpp"
#include "caf/net/write_packet_decorator.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/unit.hpp" #include "caf/unit.hpp"
......
...@@ -130,6 +130,18 @@ public: ...@@ -130,6 +130,18 @@ public:
res_->ep = ep; res_->ep = ep;
} }
transport_type& transport() {
return *this;
}
std::vector<byte> next_buffer() {
return {};
}
std::vector<byte> next_header_buffer() {
return {};
}
private: private:
std::shared_ptr<transport_result> res_; std::shared_ptr<transport_result> res_;
}; };
......
...@@ -130,6 +130,18 @@ struct dummy_transport { ...@@ -130,6 +130,18 @@ struct dummy_transport {
buf_->insert(buf_->end(), payload.begin(), payload.end()); buf_->insert(buf_->end(), payload.begin(), payload.end());
} }
transport_type& transport() {
return *this;
}
std::vector<byte> next_buffer() {
return {};
}
std::vector<byte> next_header_buffer() {
return {};
}
private: private:
std::shared_ptr<std::vector<byte>> buf_; std::shared_ptr<std::vector<byte>> buf_;
}; };
......
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