Unverified Commit 0a4f2108 authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #43

Followup for #41
parents ba13aad9 81204d8c
......@@ -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
......@@ -26,7 +26,14 @@
namespace caf::net {
/// Implements base class for transports.
/// Implements a base class for transports.
/// @tparam Transport The derived type of the transport implementation.
/// @tparam NextLayer The Following Layer. Either `transport_worker` or
/// `transport_worker_dispatcher`
/// @tparam Handle The type of the related socket_handle.
/// @tparam Application The type of the application used in this stack.
/// @tparam IdType The id type of the derived transport, must match the IdType
/// of `NextLayer`.
template <class Transport, class NextLayer, class Handle, class Application,
class IdType>
class transport_base {
......@@ -50,37 +57,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 -------------------------------------------------------------
/// Returns the socket_handle of this transport.
handle_type handle() const noexcept {
return handle_;
}
/// Returns a reference to the `actor_system` of this transport.
/// @pre `init` must be called before calling this getter.
actor_system& system() {
return manager().system();
}
/// Returns a reference to the application of this transport.
application_type& application() {
return next_layer_.application();
}
/// Returns a reference to this transport.
transport_type& transport() {
return *reinterpret_cast<transport_type*>(this);
}
/// Returns a 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.
/// @returns `error` on failure, none on success.
virtual error init(endpoint_manager& parent) {
CAF_LOG_TRACE("");
manager_ = &parent;
......@@ -96,6 +111,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 The `actor_handle` which the result 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 +128,77 @@ public:
f(next_layer_);
}
/// Gets called by an actor proxy after creation.
/// @param peer The `node_id` of the remote node.
/// @param id The id of the remote actor.
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));
}
/// Notifies the transport that the timeout identified by `value` plus `id`
/// was triggered.
/// @param value The `atom_value` of the timeout.
/// @param id The timeout id of the 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 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 -----------------------------------------------
virtual void configure_read(receive_policy::config){
/// Configures this transport for the next read event.
virtual void configure_read(receive_policy::config) {
// nop
};
}
/// Called by the endpoint manager when the transport can read data from its
/// socket.
virtual bool handle_read_event(endpoint_manager&) = 0;
/// Called by the endpoint manager when the transport can write data to its
/// socket.
virtual bool handle_write_event(endpoint_manager& parent) = 0;
/// Queues a packet scattered across multiple buffers to be sent via this
/// transport.
/// @param id The id of the destination endpoint.
/// @param buffers Pointers to the buffers that make up the packet content.
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 +207,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 +223,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