Commit bd29f8b3 authored by Jakob Otto's avatar Jakob Otto

Integrate review Feedback

parent 421aa67f
......@@ -50,10 +50,10 @@ public:
using worker_type = transport_worker<application_type>;
using id_type = unit_t;
using super = stream_transport_base<application_type>;
using id_type = typename super::id_type;
using buffer_type = typename super::buffer_type;
using write_queue_type = std::deque<std::pair<bool, buffer_type>>;
......@@ -187,12 +187,10 @@ private:
auto write_ret = write(this->handle(), make_span(data, len));
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
if (*num_bytes + written_ >= buf.size()) {
written_ += *num_bytes;
if (written_ >= buf.size()) {
recycle();
written_ = 0;
} else {
written_ += *num_bytes;
return false;
}
} else {
auto err = get<sec>(write_ret);
......@@ -213,6 +211,8 @@ private:
size_t collected_;
size_t max_;
receive_policy_flag rd_flag_;
// TODO implement retries using this member!
// size_t max_consecutive_reads_;
};
} // namespace caf::net
......@@ -27,6 +27,12 @@
namespace caf::net {
/// Implements base class for transports.
/// @tparam Transport The actual type of the implementin transport.
/// @tparam NextLayer The Following Layer. Either `transport_worker` or
/// `transport_worker_dispatcher`
/// @tparam Handle The type of the used socket_handle
/// @tparam Application The type of the application used in this stack.
/// @tparam IdType The type of the Id used by this transport.
template <class Transport, class NextLayer, class Handle, class Application,
class IdType>
class transport_base {
......@@ -50,37 +56,45 @@ public:
// -- constructors, destructors, and assignment operators --------------------
transport_base(handle_type handle, application_type application)
: next_layer_(std::move(application)),
handle_(handle),
// max_consecutive_reads_(0),
manager_(nullptr) {
: next_layer_(std::move(application)), handle_(handle), manager_(nullptr) {
// nop
}
// -- properties -------------------------------------------------------------
/// @return the `socket_handle` of this transport.
handle_type handle() const noexcept {
return handle_;
}
/// @return reference to the `actor_system` of this transport.
/// @pre `init` must be called before calling this getter.
actor_system& system() {
return manager().system();
}
/// @return reference to the `application` of this transport.
application_type& application() {
return next_layer_.application();
}
/// @return reference to `this` transport.
transport_type& transport() {
return *reinterpret_cast<transport_type*>(this);
}
/// @return reference to the `endpoint_manager` of this transport.
/// @pre `init` must be called before calling this getter.
endpoint_manager& manager() {
CAF_ASSERT(manager_);
return *manager_;
}
// -- transport member functions ---------------------------------------------
/// Initializes this transport.
/// @param parent the endpoint manager that manages this transport.
/// @return `error` on failure, none on success.
virtual error init(endpoint_manager& parent) {
CAF_LOG_TRACE("");
manager_ = &parent;
......@@ -96,6 +110,10 @@ public:
return none;
}
/// Resolves a remote actor using 'locator' and sends the resolved actor to
/// listener on success - an error otherwise.
/// @param locator The `uri` of the remote actor.
/// @param listener actor_handle which the resolved actor should be sent to.
auto resolve(endpoint_manager&, const uri& locator, const actor& listener) {
CAF_LOG_TRACE(CAF_ARG(locator) << CAF_ARG(listener));
auto f = detail::make_overload(
......@@ -109,45 +127,73 @@ public:
f(next_layer_);
}
///
void new_proxy(endpoint_manager&, const node_id& peer, actor_id id) {
next_layer_.new_proxy(*this, peer, id);
}
/// Notifies the remote endpoint that the local actor is down.
/// @param peer The `node_id` of the remote endpoint.
/// @param id The `actor_id` of the remote actor.
/// @param reason The reason why the local actor has shut_down.
void local_actor_down(endpoint_manager&, const node_id& peer, actor_id id,
error reason) {
next_layer_.local_actor_down(*this, peer, id, std::move(reason));
}
/// timeout callback for this transport. Will be called after a timeout is
/// triggered.
/// @param value The `atom_value` of the timeout.
/// @param id The timeout id of this timeout.
void timeout(endpoint_manager&, atom_value value, uint64_t id) {
next_layer_.timeout(*this, value, id);
}
void set_timeout(uint64_t timeout_id, id_type id) {
next_layer_.set_timeout(timeout_id, id);
/// Callback for setting a timeout. Will be called after setting a timeout to
/// get the timeout id for local use.
/// @param timeout_id The timeout_id of the previously set timeout.
/// @param ts Any further information that was passed when setting the
/// timeout.
template <class... Ts>
void set_timeout(uint64_t timeout_id, Ts&&... ts) {
next_layer_.set_timeout(timeout_id, std::forward<Ts>(ts)...);
}
/// Callback for when an error occurs.
/// @param code The error code to handle.
void handle_error(sec code) {
next_layer_.handle_error(code);
}
// -- (pure) virtual functions -----------------------------------------------
/// Configures the next read_event.
virtual void configure_read(receive_policy::config){
// nop
};
/// Callback when a read_event occurs.
virtual bool handle_read_event(endpoint_manager&) = 0;
/// Callback when a write_event occurs.
virtual bool handle_write_event(endpoint_manager& parent) = 0;
/// Moves a packet that is scattered across multiple buffers into the
/// write_queue of this transport.
/// @param id the endpoint id
/// @param buffers the packet scattered across multiple buffers.
virtual void write_packet(id_type id, span<buffer_type*> buffers) = 0;
// -- buffer management ------------------------------------------------------
/// Returns the next cached header_buffer or creates a new one if no buffers
/// are cached.
buffer_type next_header_buffer() {
return next_buffer_impl(header_bufs_);
}
/// Returns the next cached payload_buffer or creates a new one if no buffers
/// are cached.
buffer_type next_payload_buffer() {
return next_buffer_impl(payload_bufs_);
}
......@@ -156,9 +202,8 @@ private:
// -- utility functions ------------------------------------------------------
static buffer_type next_buffer_impl(buffer_cache_type cache) {
if (cache.empty()) {
if (cache.empty())
return {};
}
auto buf = std::move(cache.back());
cache.pop_back();
return buf;
......@@ -173,9 +218,6 @@ protected:
buffer_type read_buf_;
// TODO implement retries using this member! Should this go into stream_trans?
// size_t max_consecutive_reads_;
endpoint_manager* manager_;
};
......
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