Commit d1324ce7 authored by Joseph Noir's avatar Joseph Noir

Adds endpoint impl., not included in the logic yet

parent d4020bfe
......@@ -23,6 +23,8 @@ set (LIBCAF_IO_SRCS
src/acceptor_manager.cpp
src/multiplexer.cpp
src/uri.cpp
src/endpoint.cpp
src/endpoint_manager.cpp
src/datagram_sink_manager.cpp
src/datagram_source_manager.cpp
src/datagram_sink.cpp
......
......@@ -176,6 +176,20 @@ public:
/// Writes `data` into the buffer of a given sink.
void write(datagram_sink_handle hdl, size_t data_size, const void* data);
// Enables or disables write notifications for given datagram endpoint.
void ack_writes(endpoint_handle hdl, bool enable);
/// Modifies the buffer for received datagrams.
/// @param hdl Identifies the affected socket.
/// @param buf_size Size of the receiver buffer for the next datagram.
void configure_datagram_size(endpoint_handle hdl, size_t buf_size);
/// Returns write buffer for given sink.
std::vector<char>& wr_buf(endpoint_handle hdl);
/// Writes `data` into the buffer of a given datagram endpoint.
void write(endpoint_handle hdl, size_t data_size, const void* data);
/// Returns the middleman instance this broker belongs to.
inline middleman& parent() {
return system().middleman();
......@@ -250,6 +264,24 @@ public:
expected<datagram_source_handle>
add_datagram_source(network::native_socket fd);
/// TODO
void add_endpoint(const intrusive_ptr<endpoint>& ptr);
/// TODO
expected<void> assign_endpoint(endpoint_handle hdl);
/// TODO
expected<endpoint_handle> add_remote_endpoint(const std::string& host,
uint16_t port);
/// TODO
expected<std::pair<endpoint_handle, uint16_t>>
add_local_endpoint(uint16_t port = 0, const char* in = nullptr,
bool reuse_addr = false);
/// TODO
expected<endpoint_handle> add_endpoint(network::native_socket fd);
/// Returns the remote address associated to `hdl`
/// or empty string if `hdl` is invalid.
std::string remote_addr(connection_handle hdl);
......@@ -279,6 +311,23 @@ public:
/// Returns the local port associated to `hdl` or `0` if `hdl` is invalid.
uint16_t local_port(datagram_source_handle hdl);
/* TODO: Do we need this?
/// Returns the address associated with `hdl`
/// or empy string if `hdl` is invalid.
std::string addr(endpoint_handle hdl);
/// Returns the remote port associated to `hdl`
/// or `0` if `hdl` is invalid or not "connected".
uint16_t remote_port(endpoint_handle hdl);
/// Return the local port associated to `hdl`
/// or `0` if `hdl` is invalid.
uint16_t local_port(endpoint_handle hdl);
/// Return the endpoint handle associated to fiven local `port` or `none`.
endpoint_handle endpoint_by_port(uint16_t port);
*/
/// Closes all connections and acceptors.
void close_all();
......@@ -367,6 +416,12 @@ protected:
inline datagram_source_map& get_map(datagram_source_handle) {
return datagram_sources_;
}
// meta programming utility
inline endpoint_map& get_map(endpoint_handle) {
return endpoints_;
}
// meta programming utility (not implemented)
static intrusive_ptr<doorman> ptr_of(accept_handle);
......@@ -379,6 +434,9 @@ protected:
// meta programming utility (not implemented)
static intrusive_ptr<datagram_source> ptr_of(datagram_source_handle);
// meta programming utility (not implemented)
static intrusive_ptr<endpoint> ptr_of(endpoint_handle);
/// @endcond
/// Returns the `multiplexer` running this broker.
......@@ -413,6 +471,7 @@ protected:
private:
scribe_map scribes_;
doorman_map doormen_;
endpoint_map endpoints_;
datagram_sink_map datagram_sinks_;
datagram_source_map datagram_sources_;
detail::intrusive_partitioned_list<mailbox_element, detail::disposer> cache_;
......
......@@ -30,6 +30,7 @@
#include "caf/io/scribe.hpp"
#include "caf/io/doorman.hpp"
#include "caf/io/endpoint.hpp"
#include "caf/io/datagram_sink.hpp"
#include "caf/io/abstract_broker.hpp"
#include "caf/io/datagram_source.hpp"
......
......@@ -105,12 +105,16 @@ protected:
typename std::conditional<
std::is_same<Handle, accept_handle>::value,
acceptor_passivated_msg,
typename std::conditional<
std::is_same<Handle, endpoint_handle>::value,
endpoint_passivated_msg,
typename std::conditional<
std::is_same<Handle, datagram_sink_handle>::value,
datagram_sink_passivated_msg,
datagram_source_passivated_msg
>::type
>::type
>::type
>::type;
using tmp_t = mailbox_element_vals<passiv_t>;
tmp_t tmp{strong_actor_ptr{}, message_id::make(),
......
......@@ -34,7 +34,7 @@ namespace io {
using datagram_sink_base = broker_servant<network::datagram_sink_manager,
datagram_sink_handle,
datagram_sent_msg>;
datagram_sink_msg>;
/// Manages writing to a datagram sink.
/// @ingroup Broker
......
......@@ -20,6 +20,8 @@
#ifndef CAF_IO_ENDPOINT_HPP
#define CAF_IO_ENDPOINT_HPP
#include <vector>
#include "caf/message.hpp"
#include "caf/io/broker_servant.hpp"
......@@ -30,15 +32,14 @@
namespace caf {
namespace io {
using endpoint_base = broker_servant<network::endpoint_manager,
endpoint_handle,
datagram_sent_msg>;
using endpoint_base = broker_servant<network::endpoint_manager, endpoint_handle,
datagram_msg>;
/// Manages writing and reading on a datagram endpoint.
/// @ingroup Broker
class endpoint : public endpoint_base {
public:
endpoint(abstract_broker* parent, datagram_sink_handle hdl);
endpoint(abstract_broker* parent, endpoint_handle hdl);
~endpoint();
......@@ -46,6 +47,7 @@ public:
virtual void ack_writes(bool enable) = 0;
/// Configure buffer size for next accepted datagram.
/// Implicitly starts the read loop on first call.
virtual void configure_datagram_size(size_t buf_size) = 0;
/// Returns the current write buffer.
......
......@@ -66,7 +66,7 @@ private:
}
};
} // namespace ios
} // namespace io
} // namespace caf
namespace std {
......
......@@ -26,6 +26,7 @@ namespace io {
class scribe;
class broker;
class doorman;
class endpoint;
class middleman;
class basp_broker;
class datagram_sink;
......
......@@ -38,6 +38,7 @@
#include "caf/io/network/multiplexer.hpp"
#include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp"
#include "caf/io/network/endpoint_manager.hpp"
#include "caf/io/network/datagram_sink_manager.hpp"
#include "caf/io/network/datagram_source_manager.hpp"
......@@ -254,7 +255,7 @@ public:
friend class io::middleman; // disambiguate reference
friend class supervisor;
using endpoint = std::pair<std::string,uint16_t>;
using endpoint_addr = std::pair<std::string,uint16_t>;
struct event {
native_socket fd;
......@@ -326,6 +327,26 @@ public:
add_datagram_source(abstract_broker* ptr, uint16_t port , const char* in,
bool rflag) override;
expected<endpoint_handle>
new_remote_endpoint(const std::string& host, uint16_t port) override;
expected<std::pair<endpoint_handle, uint16_t>>
new_local_endpoint(uint16_t port, const char* in, bool reuse_addr) override;
expected<void> assign_endpoint(abstract_broker* ptr,
endpoint_handle hdl) override;
expected<endpoint_handle> add_remote_endpoint(abstract_broker* ptr,
const std::string& host,
uint16_t port) override;
expected<std::pair<endpoint_handle, uint16_t>>
add_local_endpoint(abstract_broker* ptr, uint16_t port, const char* in,
bool reuse_addr) override;
endpoint_handle add_endpoint(abstract_broker* ptr,
network::native_socket fd) override;
void exec_later(resumable* ptr) override;
explicit default_multiplexer(actor_system* sys);
......@@ -343,7 +364,7 @@ public:
/// @cond PRIVATE
// Used by datagram senders and receivers to manage known endpoints
std::map<endpoint, datagram_sink_handle>& endpoints();
std::map<endpoint_addr, endpoint_handle>& endpoints();
/// @endcond
......@@ -408,7 +429,7 @@ private:
pipe_reader pipe_reader_;
// TODO: is this the right place?
// How to maintain endpoints if they close?
std::map<endpoint, datagram_sink_handle> remote_endpoints_;
std::map<endpoint_addr, endpoint_handle> remote_endpoints_;
};
inline connection_handle conn_hdl_from_socket(native_socket fd) {
......@@ -427,6 +448,10 @@ inline datagram_source_handle dg_source_hdl_from_socket(native_socket fd) {
return datagram_source_handle::from_int(int64_from_native_socket(fd));
}
inline endpoint_handle endpoint_hdl_from_socket(native_socket fd) {
return endpoint_handle::from_int(int64_from_native_socket(fd));
}
/// A stream capable of both reading and writing. The stream's input
/// data is forwarded to its {@link stream_manager manager}.
class stream : public event_handler {
......@@ -544,6 +569,87 @@ private:
native_socket sock_;
};
class datagram_hdlr : public event_handler {
public:
/// A manager type providing the TODO
using manager_type = endpoint_manager;
/// A smart pointer to an endpoint_manager.
using manager_ptr = intrusive_ptr<endpoint_manager>;
/// A buffer class providing a compatible
/// interface to `std::vector`.
using buffer_type = std::vector<char>;
datagram_hdlr(default_multiplexer& backend_ref, native_socket sockfd);
void ack_writes(bool x);
/// Copies data to the write buffer.
/// @warning Not thread safe.
void write(const void* buf, size_t num_bytes);
/// Configures how much buffer will be provided for the next datagram.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void configure_datagram_size(size_t buf_size);
/// Returns the write buffer of this datagram sender.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline buffer_type& wr_buf() {
return wr_offline_buf_;
}
/// Returns the read buffer of this datagram receiver.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline buffer_type& rd_buf() {
return rd_buf_;
}
std::pair<std::string,uint16_t> get_sender();
/// Sends the content of the write buffer, calling the `io_failure`
/// member function of `mgr` in case of an error.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void flush(const manager_ptr& mgr);
/// Starts forwarding incoming data to `mgr`.
void start(manager_type* mgr);
/// Activates the datagram_sender.
void activate(manager_type* mgr);
/// Closes the network connection and removes this handler from its parent.
void stop_reading();
void removed_from_loop(operation op) override;
void handle_event(operation op) override;
private:
void prepare_next_read();
void prepare_next_write();
// state for receiving
manager_ptr reader_;
size_t buf_size_;
size_t packet_size_;
buffer_type rd_buf_;
struct sockaddr_storage last_sender;
socklen_t sender_len;
// state for sending
manager_ptr writer_;
bool ack_writes_;
bool writing_;
buffer_type wr_buf_;
buffer_type wr_offline_buf_;
};
/// A datagram_sender is responsible for sending datagrams to an endpoint.
class datagram_sender: public event_handler {
public:
......@@ -668,6 +774,13 @@ new_datagram_sink_impl(const std::string& host, uint16_t port,
expected<std::pair<native_socket, uint16_t>>
new_datagram_source_impl(uint16_t port, const char* addr, bool reuse_addr);
expected<native_socket>
new_remote_endpoint_impl(const std::string& host, uint16_t port,
optional<protocol> preferred = none);
expected<std::pair<native_socket, uint16_t>>
new_local_endpoint_impl(uint16_t port, const char* addr, bool reuse_addr);
} // namespace network
} // namespace io
} // namespace caf
......
......@@ -32,6 +32,7 @@
#include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/endpoint_handle.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/datagram_sink_handle.hpp"
#include "caf/io/datagram_source_handle.hpp"
......@@ -147,6 +148,34 @@ public:
add_datagram_source(abstract_broker* ptr, uint16_t port,
const char* in = nullptr, bool reuse_addr = false) = 0;
/// TODO
virtual expected<endpoint_handle>
new_remote_endpoint(const std::string& host, uint16_t port) = 0;
/// TODO
virtual expected<std::pair<endpoint_handle, uint16_t>>
new_local_endpoint(uint16_t port = 0, const char* in = nullptr,
bool reuse_addr = false) = 0;
/// TODO
virtual expected<void> assign_endpoint(abstract_broker* ptr,
endpoint_handle hdl) = 0;
/// TODO
virtual expected<endpoint_handle> add_remote_endpoint(abstract_broker* ptr,
const std::string& host,
uint16_t port) = 0;
/// TODO
expected<std::pair<endpoint_handle, uint16_t>>
virtual add_local_endpoint(abstract_broker* ptr, uint16_t port = 0,
const char* in = nullptr,
bool reuse_addr = false) = 0;
/// TODO
virtual endpoint_handle add_endpoint(abstract_broker* ptr,
network::native_socket fd) = 0;
/// Simple wrapper for runnables
class runnable : public resumable, public ref_counted {
public:
......
......@@ -85,6 +85,27 @@ public:
add_datagram_source(abstract_broker* ptr, uint16_t port,
const char* in, bool reuse_addr) override;
expected<endpoint_handle>
new_remote_endpoint(const std::string& host, uint16_t port) override;
expected<std::pair<endpoint_handle, uint16_t>>
new_local_endpoint(uint16_t port, const char* in, bool reuse_addr) override;
expected<void> assign_endpoint(abstract_broker* ptr,
endpoint_handle hdl) override;
expected<endpoint_handle> add_remote_endpoint(abstract_broker* ptr,
const std::string& host,
uint16_t port) override;
expected<std::pair<endpoint_handle, uint16_t>>
add_local_endpoint(abstract_broker* ptr, uint16_t port, const char* in,
bool reuse_addr) override;
endpoint_handle add_endpoint(abstract_broker* ptr,
network::native_socket fd) override;
supervisor_ptr make_supervisor() override;
void run() override;
......@@ -111,12 +132,18 @@ public:
/// Returns the output buffer of the datagram source identified by `hdl`.
buffer_type& output_buffer(datagram_sink_handle hdl);
/// Returns the output buffer of the datagram source identified by `hdl`.
buffer_type& output_buffer(endpoint_handle hdl);
/// Returns the input buffer of the scribe identified by `hdl`.
buffer_type& input_buffer(connection_handle hdl);
/// Returns the input buffer of the datagram source identified by `hdl`.
buffer_type& input_buffer(datagram_source_handle hdl);
/// Returns the input buffer of the datagram source identified by `hdl`.
buffer_type& input_buffer(endpoint_handle hdl);
/// Returns the configured read policy of the scribe identified by `hdl`.
receive_policy::config& read_config(connection_handle hdl);
......@@ -161,6 +188,25 @@ public:
size_t& buffer_size(datagram_source_handle hdl);
uint16_t& local_port(endpoint_handle hdl);
uint16_t& remote_port(endpoint_handle hdl);
intrusive_ptr<endpoint>& impl_ptr(endpoint_handle hdl);
/// Returns `true` if this handle has been closed
/// for reading, `false` otherwise.
bool& stopped_reading(endpoint_handle hdl);
/// Returns `true` if this handle is inactive, otherwise `false`.
bool& passive_mode(endpoint_handle hdl);
/// Returns whether the scribe identified by `hdl` receives write ACKs.
bool& ack_writes(endpoint_handle hdl);
/// Returns size of the receive buffer for the next datagram.
size_t& buffer_size(endpoint_handle hdl);
/// Returns `true` if this handle has been closed
/// for reading, `false` otherwise.
bool& stopped_reading(accept_handle hdl);
......@@ -189,10 +235,20 @@ public:
bool has_pending_datagram_sink(std::string host, uint16_t port);
using pending_datagram_sources_map = std::map<uint16_t, datagram_source_handle>;
using pending_datagram_sources_map
= std::map<uint16_t, datagram_source_handle>;
bool has_pending_datagram_source(uint16_t port);
using pending_local_endpoints_map = std::map<uint16_t, endpoint_handle>;
bool has_pending_local_endpoint(std::string host, uint16_t port);
using pending_remote_endpoints_map
= std::map<std::pair<std::string, uint16_t>, endpoint_handle>;
bool has_pending_remote_endpoint(std::string host, uint16_t port);
/// Accepts a pending connect on `hdl`.
bool accept_connection(accept_handle hdl);
......@@ -260,16 +316,32 @@ private:
intrusive_ptr<datagram_source> ptr;
};
struct endpoint_data {
uint16_t local_port;
uint16_t remote_port;
buffer_type xbuf;
buffer_type re_buf;
buffer_type wr_buf;
size_t re_buf_size;
bool stopped_reading = false;
bool passive_mode = false;
intrusive_ptr<endpoint> ptr;
bool ack_writes = false;
};
// guards resumables_ and scribes_
std::mutex mx_;
std::condition_variable cv_;
std::list<resumable_ptr> resumables_;
pending_scribes_map scribes_;
pending_local_endpoints_map local_endpoints_;
pending_remote_endpoints_map remote_endpoints_;
pending_datagram_sinks_map datagram_sinks_;
pending_datagram_sources_map datagram_sources_;
std::unordered_map<uint16_t, accept_handle> doormen_;
std::unordered_map<connection_handle, scribe_data> scribe_data_;
std::unordered_map<accept_handle, doorman_data> doorman_data_;
std::unordered_map<endpoint_handle, endpoint_data> endpoint_data_;
std::unordered_map<datagram_sink_handle, datagram_sink_data> datagram_sink_data_;
std::unordered_map<datagram_source_handle, datagram_source_data> datagram_source_data_;
pending_connects_map pending_connects_;
......
......@@ -153,20 +153,6 @@ inspect(Inspector& f, datagram_source_closed_msg& x) {
return f(meta::type_name("datagram_source_closed_msg"), x.handle);
}
/// Signalizes that a datagram with a certain size has been sent.
struct datagram_sent_msg {
// Handle to the endpoint used.
datagram_sink_handle handle;
// number of bytes written.
uint64_t written;
};
/// @relates datagram_sent_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, datagram_sent_msg& x) {
return f(meta::type_name("datagram_sent_msg"), x.handle, x.written);
}
/// Signalizes newly arrived datagram for a {@link broker}.
struct new_datagram_msg {
/// Handle to the related datagram endpoint.
......@@ -181,6 +167,20 @@ typename Inspector::result_type inspect(Inspector& f, new_datagram_msg& x) {
return f(meta::type_name("new_datagram_msg"), x.handle, x.buf);
}
/// Signalizes that a datagram with a certain size has been sent.
struct datagram_sink_msg {
// Handle to the endpoint used.
datagram_sink_handle handle;
// number of bytes written.
uint64_t written;
};
/// @relates datagram_sink_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, datagram_sink_msg& x) {
return f(meta::type_name("datagram_sink_msg"), x.handle, x.written);
}
/// Signalizes that a datagram sink has entered passive mode.
struct datagram_sink_passivated_msg {
datagram_sink_handle handle;
......@@ -216,6 +216,47 @@ typename Inspector::result_type
inspect(Inspector& f, endpoint_closed_msg& x) {
return f(meta::type_name("endpoint_closed_msg"), x.handle);
}
// Signalizes newly arrived datagram for a {@link broker}.
struct datagram_msg {
/// Handle to the related datagram endpoint.
endpoint_handle handle;
/// Buffer containing the received data.
std::vector<char> buf;
};
/// @relates datagram_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, datagram_msg& x) {
return f(meta::type_name("datagram_msg"), x.handle, x.buf);
}
/// Signalizes that a datagram with a certain size has been sent.
struct datagram_sent_msg {
// Handle to the endpoint used.
endpoint_handle handle;
// number of bytes written.
uint64_t written;
};
/// @relates datagram_sent_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, datagram_sent_msg& x) {
return f(meta::type_name("datagram_sent_msg"), x.handle, x.written);
}
/// Signalizes that a endpoint has entered passive mode.
struct endpoint_passivated_msg {
endpoint_handle handle;
};
/// @relates endpoint_passivated_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, endpoint_passivated_msg& x) {
return f(meta::type_name("endpoint_passivated_msg"), x.handle);
}
} // namespace io
} // namespace caf
......
......@@ -128,7 +128,8 @@ void abstract_broker::configure_datagram_size(datagram_source_handle hdl,
std::vector<char>& abstract_broker::wr_buf(datagram_sink_handle hdl) {
auto x = by_id(hdl);
if (!x) {
CAF_LOG_ERROR("tried to access wr_buf() of an unknown connection_handle");
CAF_LOG_ERROR("tried to access wr_buf() of an unknown"
"datagram_sink_handle");
return dummy_wr_buf_;
}
return x->wr_buf();
......@@ -142,6 +143,37 @@ void abstract_broker::write(datagram_sink_handle hdl, size_t bs,
out.insert(out.end(), first, last);
}
void abstract_broker::ack_writes(endpoint_handle hdl, bool enable) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(enable));
auto x = by_id(hdl);
if (x)
x->ack_writes(enable);
}
void abstract_broker::configure_datagram_size(endpoint_handle hdl,
size_t buf_size) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(buf_size));
auto x = by_id(hdl);
if (x)
x->configure_datagram_size(buf_size);
}
std::vector<char>& abstract_broker::wr_buf(endpoint_handle hdl) {
auto x = by_id(hdl);
if (!x) {
CAF_LOG_ERROR("tried to access wr_buf() of an unknown endpoint_handle");
return dummy_wr_buf_;
}
return x->wr_buf();
}
void abstract_broker::write(endpoint_handle hdl, size_t bs, const void* buf) {
auto& out = wr_buf(hdl);
auto first = reinterpret_cast<const char*>(buf);
auto last = first + bs;
out.insert(out.end(), first, last);
}
std::vector<connection_handle> abstract_broker::connections() const {
std::vector<connection_handle> result;
result.reserve(scribes_.size());
......@@ -243,6 +275,35 @@ abstract_broker::add_datagram_source(network::native_socket fd) {
return backend().add_datagram_source(this, fd);
}
void abstract_broker::add_endpoint(const intrusive_ptr<endpoint>& ptr) {
endpoints_.emplace(ptr->hdl(), ptr);
ptr->launch(); // TODO: Is this required?
}
expected<void> abstract_broker::assign_endpoint(endpoint_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
return backend().assign_endpoint(this, hdl);
}
expected<endpoint_handle>
abstract_broker::add_remote_endpoint(const std::string& host, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(host) << CAF_ARG(port));
return backend().add_remote_endpoint(this, host, port);
}
expected<std::pair<endpoint_handle, uint16_t>>
abstract_broker::add_local_endpoint(uint16_t port, const char* in,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in) << CAF_ARG(reuse_addr));
return backend().add_local_endpoint(this, port, in, reuse_addr);
}
expected<endpoint_handle>
abstract_broker::add_endpoint(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return backend().add_endpoint(this, fd);
}
std::string abstract_broker::remote_addr(connection_handle hdl) {
auto i = scribes_.find(hdl);
return i != scribes_.end() ? i->second->addr() : std::string{};
......@@ -295,13 +356,9 @@ void abstract_broker::close_all() {
// stop_reading will remove the scribe from scribes_
scribes_.begin()->second->stop_reading();
}
while (!datagram_sinks_.empty()) {
// stop_reading will remove the datagram_sink from datagram_sinks_
datagram_sinks_.begin()->second->stop_reading();
}
while (!datagram_sources_.empty()) {
// stop_reading will remove the datgram_source from datgram_sources_
datagram_sources_.begin()->second->stop_reading();
while (!endpoints_.empty()) {
// stop_reading will remove the enpoint from endpoints_
endpoints_.begin()->second->stop_reading();
}
}
......
......@@ -465,30 +465,6 @@ void basp_broker_state::set_context(connection_handle hdl) {
this_context = &i->second;
}
void basp_broker_state::set_context(datagram_source_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
/*
auto i = udp_ctx.find(hdl);
if (i == udp_ctx.end()) {
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
},
none, hdl,
none,
0, 0,
0, 0,
none
}
).first;
}
// TODO: set some context?
*/
}
/******************************************************************************
* basp_broker *
******************************************************************************/
......@@ -546,11 +522,13 @@ behavior basp_broker::make_behavior() {
}
},
// received from underlying broker implementation
[=](new_datagram_msg& msg) {
[=](datagram_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
CAF_LOG_DEBUG("Received new_datagram_msg: " << CAF_ARG(msg));
auto& hdl = msg.handle;
// state.instance.handle(context(), msg,
// TODO: implement this
// look for existing context or create a new one
// handle messge
static_cast<void>(msg);
},
// received from proxy instances
[=](forward_atom, strong_actor_ptr& src,
......@@ -684,17 +662,18 @@ behavior basp_broker::make_behavior() {
rp.deliver(std::move(res.error()));
}
},
[=](connect_atom, datagram_sink_handle hdl, uint16_t port) {
[=](connect_atom, endpoint_handle hdl, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hdl.id()));
static_cast<void>(hdl);
static_cast<void>(port);
/*
auto rp = make_response_promise();
auto res = assign_datagram_sink(hdl);
if (res) {
auto& ctx = state.udp_ctx[hdl];
/*
ctx.sink = hdl;
ctx.remote_port = port;
ctx.callback = rp;
*/
// TODO: Start handshake with server as there is no way for
// the server to initiate this.
} else {
......@@ -702,6 +681,7 @@ behavior basp_broker::make_behavior() {
<< CAF_ARG(res));
rp.deliver(std::move(res.error()));
}
*/
},
[=](delete_atom, const node_id& nid, actor_id aid) {
CAF_LOG_TRACE(CAF_ARG(nid) << ", " << CAF_ARG(aid));
......
......@@ -37,8 +37,9 @@ message datagram_sink::detach_message() {
return make_message(datagram_sink_closed_msg{hdl()});
}
void datagram_sink::datagram_sent(execution_unit* ctx, size_t written) {
void datagram_sink::datagram_sent(execution_unit*, size_t) {
CAF_LOG_TRACE(CAF_ARG(written));
/*
if (detached())
return;
using sent_t = datagram_sent_msg;
......@@ -47,6 +48,7 @@ void datagram_sink::datagram_sent(execution_unit* ctx, size_t written) {
mailbox_element::forwarding_stack{},
sent_t{hdl(), written}};
invoke_mailbox_element_impl(ctx, tmp);
*/
}
void datagram_sink::io_failure(execution_unit* ctx, network::operation op) {
......
......@@ -622,7 +622,7 @@ void default_multiplexer::del(operation op, native_socket fd,
new_event(del_flag, op, fd, ptr);
}
std::map<default_multiplexer::endpoint, datagram_sink_handle>&
std::map<default_multiplexer::endpoint_addr, endpoint_handle>&
default_multiplexer::endpoints() {
return remote_endpoints_;
}
......@@ -862,6 +862,84 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
return ptr->hdl();
}
endpoint_handle default_multiplexer::add_endpoint(abstract_broker* self,
network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
CAF_ASSERT(fd != network::invalid_native_socket);
class impl : public endpoint {
public:
impl(abstract_broker* ptr, default_multiplexer& mx, native_socket sockfd)
: endpoint(ptr, network::endpoint_hdl_from_socket(sockfd)),
handler_(mx, sockfd) {
// nop
}
void ack_writes(bool enable) override {
handler_.ack_writes(enable);
}
void configure_datagram_size(size_t buf_size) override {
CAF_LOG_TRACE("");
handler_.configure_datagram_size(buf_size);
if (!launched_)
launch();
}
void stop_reading() override {
CAF_LOG_TRACE("");
handler_.stop_reading();
detach(&handler_.backend(), false);
}
std::vector<char>& wr_buf() override {
return handler_.wr_buf();
}
std::vector<char>& rd_buf() override {
return handler_.rd_buf();
}
void launch() override {
CAF_LOG_TRACE("");
handler_.start(this);
}
std::string addr() const override {
auto x = remote_addr_of_fd(handler_.fd());
if (!x)
return "";
return std::move(*x);
}
std::string local_addr() const {
auto x = local_addr_of_fd(handler_.fd());
if (!x)
return "";
return std::move(*x);
}
uint16_t port() const override {
auto x = remote_port_of_fd(handler_.fd());
if (!x)
return 0;
return *x;
}
uint16_t local_port() const {
auto x = local_port_of_fd(handler_.fd());
if (!x)
return 0;
return *x;
}
void flush() {
CAF_LOG_TRACE("");
handler_.flush(this);
}
void add_to_loop() override {
handler_.activate(this);
}
void remove_from_loop() override {
handler_.passivate();
}
private:
bool launched_;
network::datagram_hdlr handler_;
};
auto ptr = make_counted<impl>(self, *this, fd);
self->add_endpoint(ptr);
return ptr->hdl();
}
datagram_sink_handle
default_multiplexer::add_datagram_sink(abstract_broker* self,
native_socket fd) {
......@@ -1072,15 +1150,67 @@ expected<void> default_multiplexer::assign_datagram_source(abstract_broker* self
expected<std::pair<datagram_source_handle, uint16_t>>
default_multiplexer::add_datagram_source(abstract_broker* self, uint16_t port ,
const char* host, bool reuse_addr) {
auto res = new_datagram_source_impl(port, host, reuse_addr);
default_multiplexer::add_datagram_source(abstract_broker* self, uint16_t port,
const char* in, bool reuse_addr) {
auto res = new_datagram_source_impl(port, in, reuse_addr);
if (!res)
return std::move(res.error());
auto bound_port = res->second;
return std::make_pair(add_datagram_source(self, res->first), bound_port);
}
expected<endpoint_handle>
default_multiplexer::new_remote_endpoint(const std::string& host,
uint16_t port) {
auto fd = new_remote_endpoint_impl(host, port);
if (!fd)
return std::move(fd.error());
// TODO: add endpoint to endpoints vector of middleman!
// Maybe this should be done in the new_remote_enpoint_impl
return endpoint_handle::from_int(int64_from_native_socket(*fd));
}
expected<std::pair<endpoint_handle, uint16_t>>
default_multiplexer::new_local_endpoint(uint16_t port, const char* in,
bool reuse_addr) {
auto res = new_local_endpoint_impl(port, in, reuse_addr);
if (!res)
return std::move(res.error());
return std::make_pair(endpoint_handle::from_int(
int64_from_native_socket(res->first)),
res->second);
}
expected<void> default_multiplexer::assign_endpoint(abstract_broker* self,
endpoint_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(self->id()) << CAF_ARG(hdl));
add_endpoint(self, static_cast<native_socket>(hdl.id()));
return unit;
}
expected<endpoint_handle>
default_multiplexer::add_remote_endpoint(abstract_broker* self,
const std::string& host,
uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(self->id()) << CAF_ARG(host) << CAF_ARG(port));
auto fd = new_remote_endpoint_impl(host, port);
if (!fd)
return std::move(fd.error());
// TODO: add endpoint to endpoints vector of middleman!
// Maybe this should be done in the new_remote_enpoint_impl
return add_endpoint(self, *fd);
}
expected<std::pair<endpoint_handle, uint16_t>>
default_multiplexer::add_local_endpoint(abstract_broker* self, uint16_t port,
const char* in, bool reuse_addr) {
auto res = new_local_endpoint_impl(port, in, reuse_addr);
if (!res)
return std::move(res.error());
auto bound_port = res->second;
return std::make_pair(add_endpoint(self, res->first), bound_port);
}
/******************************************************************************
* platform-independent implementations (finally) *
******************************************************************************/
......@@ -1616,21 +1746,6 @@ void datagram_receiver::handle_event(operation op) {
}
if (rb == 0)
return;
auto& endpoints = backend().endpoints();
auto sender = get_sender();
auto endpoint = endpoints.find(sender);
if (endpoint == endpoints.end()) {
auto new_endpoint = backend().add_datagram_sink(reader_->parent(),
sender.first,
sender.second);
if (!new_endpoint) {
// TODO: error handling
CAF_LOG_DEBUG("Could not create endpoint for new sender.");
return;
}
endpoint = endpoints.emplace(sender, *new_endpoint).first;
}
auto hdl = endpoint->second;
auto res = reader_->consume(&backend(), rd_buf_.data(), rb);
packet_size_ = rb;
prepare_next_read();
......@@ -1685,6 +1800,185 @@ void datagram_receiver::prepare_next_read() {
rd_buf_.resize(buf_size_);
}
datagram_hdlr::datagram_hdlr(default_multiplexer& backend_ref,
native_socket sockfd)
: event_handler(backend_ref, sockfd),
buf_size_(0),
ack_writes_(false),
writing_(false) {
// TODO: Set some reasonable default.
configure_datagram_size(1500);
}
void datagram_hdlr::ack_writes(bool x) {
ack_writes_ = x;
}
void datagram_hdlr::write(const void* buf, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(num_bytes));
auto first = reinterpret_cast<const char*>(buf);
auto last = first + num_bytes;
wr_offline_buf_.insert(wr_offline_buf_.end(), first, last);
}
void datagram_hdlr::configure_datagram_size(size_t buf_size) {
buf_size_ = buf_size;
}
std::pair<std::string,uint16_t> datagram_hdlr::get_sender() {
char addr[INET6_ADDRSTRLEN];
std::string host;
uint16_t port = 0;
switch(last_sender.ss_family) {
case AF_INET:
port = ntohs(reinterpret_cast<sockaddr_in*>(&last_sender)->sin_port);
inet_ntop(AF_INET,
&reinterpret_cast<sockaddr_in*>(&last_sender)->sin_addr,
addr, INET_ADDRSTRLEN);
host.insert(std::begin(host), std::begin(addr),
std::begin(addr) + INET_ADDRSTRLEN);
break;
case AF_INET6:
port = ntohs(reinterpret_cast<sockaddr_in6*>(&last_sender)->sin6_port);
inet_ntop(AF_INET6,
&reinterpret_cast<sockaddr_in*>(&last_sender)->sin_addr,
addr, INET6_ADDRSTRLEN);
host.insert(std::begin(host), std::begin(addr),
std::begin(addr) + INET6_ADDRSTRLEN);
break;
default:
CAF_LOG_DEBUG("Received data with unknown protocol family.");
// nop
break;
}
return std::make_pair(std::move(host), port);
}
void datagram_hdlr::flush(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr);
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
if (!wr_offline_buf_.empty() && !writing_) {
backend().add(operation::write, fd(), this);
writer_ = mgr;
writing_ = true;
prepare_next_write();
}
}
void datagram_hdlr::start(manager_type* mgr) {
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void datagram_hdlr::activate(manager_type* mgr) {
if (!reader_) {
reader_.reset(mgr);
event_handler::activate();
prepare_next_read();
}
}
void datagram_hdlr::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
void datagram_hdlr::removed_from_loop(operation op) {
switch (op) {
case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break;
case operation::propagate_error: break;
}
}
void datagram_hdlr::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case operation::read: {
// Read next datagram
size_t rb;
if (!receive_datagram(rb, fd(), rd_buf_.data(), rd_buf_.size(),
last_sender, sender_len)) {
reader_->io_failure(&backend(), operation::read);
passivate();
return;
}
if (rb == 0)
return;
// TODO: find responsible (remote) endpoint to deliver datagram
/*
auto& endpoints = backend().endpoints();
auto sender = get_sender();
auto endpoint = endpoints.find(sender);
if (endpoint == endpoints.end()) {
auto new_endpoint = backend().add_datagram_sink(reader_->parent(),
sender.first,
sender.second);
if (!new_endpoint) {
// TODO: error handling
CAF_LOG_DEBUG("Could not create endpoint for new sender.");
return;
}
endpoint = endpoints.emplace(sender, *new_endpoint).first;
}
auto hdl = endpoint->second;
*/
auto res = reader_->consume(&backend(), rd_buf_.data(), rb);
packet_size_ = rb;
prepare_next_read();
if (!res) {
passivate();
return;
}
break;
}
case operation::write: {
size_t wb; // written bytes
if (!send_datagram(wb, fd(), wr_buf_.data(), wr_buf_.size())) {
writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, fd(), this);
} else if (wb > 0) {
CAF_ASSERT(wb == wr_buf_.size());
if (ack_writes_)
writer_->datagram_sent(&backend(), wb);
prepare_next_write();
} else {
// TODO: remove this if sure that datagrams are either written
// as a whole or not at all
std::cerr << "Partial datagram wrtten: " << wb
<< " of " << wr_buf_.size() << std::endl;
if (writer_)
writer_->io_failure(&backend(), operation::write);
}
break;
break;
}
case operation::propagate_error:
if (reader_)
reader_->io_failure(&backend(), operation::read);
if (writer_)
writer_->io_failure(&backend(), operation::write);
// backend will delete this handler anyway,
// no need to call backend().del() here
break;
}
}
void datagram_hdlr::prepare_next_read() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.size()) << CAF_ARG(wr_offline_buf_.size()));
wr_buf_.clear();
if (wr_offline_buf_.empty()) {
backend().del(operation::write, fd(), this);
} else {
wr_buf_.swap(wr_offline_buf_);
}
}
void datagram_hdlr::prepare_next_write() {
rd_buf_.resize(buf_size_);
}
class socket_guard {
public:
explicit socket_guard(native_socket fd) : fd_(fd) {
......@@ -1918,6 +2212,40 @@ expected<native_socket> new_datagram_sink_impl(const std::string& host,
return sguard.release();
}
expected<native_socket> new_remote_endpoint_impl(const std::string& host,
uint16_t port,
optional<protocol> preferred) {
CAF_LOG_TRACE(CAF_ARG(host) << CAF_ARG(port) << CAF_ARG(preferred));
CAF_LOG_INFO("trying to create endpoint for:" << CAF_ARG(host)
<< CAF_ARG(port));
auto res = interfaces::native_address(host, preferred);
if (!res) {
CAF_LOG_INFO("no such host");
return make_error(sec::cannot_connect_to_node, "no such host", host, port);
}
auto proto = res->second;
CAF_ASSERT(proto == ipv4 || proto == ipv6);
CALL_CFUN(fd, cc_valid_socket, "socket",
socket(proto == ipv4 ? AF_INET : AF_INET6, SOCK_DGRAM, 0));
socket_guard sguard(fd);
if (proto == ipv6) {
if (ip_connect<AF_INET6>(fd, res->first, port)) {
CAF_LOG_INFO("successfully connected to host via IPv6");
return sguard.release();
}
sguard.close();
// IPv4 fallback
return new_datagram_sink_impl(host, port, ipv4);
}
if (!ip_connect<AF_INET>(fd, res->first, port)) {
CAF_LOG_INFO("could not connect to:" << CAF_ARG(host) << CAF_ARG(port));
return make_error(sec::cannot_connect_to_node,
"ip_connect failed", host, port);
}
CAF_LOG_INFO("successfully connected to host via IPv4");
return sguard.release();
}
expected<std::pair<native_socket, uint16_t>>
new_datagram_source_impl(uint16_t port, const char* addr, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
......@@ -1953,6 +2281,40 @@ new_datagram_source_impl(uint16_t port, const char* addr, bool reuse_addr) {
return std::make_pair(sguard.release(), *p);
}
expected<std::pair<native_socket, uint16_t>>
new_local_endpoint_impl(uint16_t port, const char* addr, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
protocol proto = ipv6;
if (addr) {
auto addrs = interfaces::native_address(addr);
if (!addrs)
return make_error(sec::cannot_open_port, "Invalid ADDR", addr);
proto = addrs->second;
CAF_ASSERT(proto == ipv4 || proto == ipv6);
}
CALL_CFUN(fd, cc_valid_socket, "socket",
socket(proto == ipv4 ? AF_INET : AF_INET6, SOCK_DGRAM, 0));
// sguard closes the socket in case of exception
socket_guard sguard(fd);
if (reuse_addr) {
int on = 1;
CALL_CFUN(tmp1, cc_zero, "setsockopt",
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<setsockopt_ptr>(&on),
static_cast<socklen_t>(sizeof(on))));
}
// TODO: Change this to something UDP specific?
auto p = proto == ipv4 ? new_ip_acceptor_impl<AF_INET>(fd, port, addr)
: new_ip_acceptor_impl<AF_INET6>(fd, port, addr);
if (!p)
return std::move(p.error());
// Not needed: CALL_CFUN(tmp2, cc_zero, "listen", listen(fd, SOMAXCONN));
// Should there be some handshake between nodes?
// TODO: Remove comments
// ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(p));
return std::make_pair(sguard.release(), *p);
}
expected<std::string> local_addr_of_fd(native_socket fd) {
sockaddr_storage st;
......
......@@ -25,7 +25,7 @@
namespace caf {
namespace io {
endpoint::endpoint(abstract_broker* parent, datagram_sink_handle hdl)
endpoint::endpoint(abstract_broker* parent, endpoint_handle hdl)
: endpoint_base(parent, hdl) {
// nop
}
......@@ -38,5 +38,49 @@ message endpoint::detach_message() {
return make_message(endpoint_closed_msg{hdl()});
}
bool endpoint::consume(execution_unit* ctx, const void*, size_t num_bytes) {
CAF_ASSERT(ctx != nullptr);
CAF_LOG_TRACE(CAF_ARG(num_bytes));
if (detached())
// we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an I/O event causes
// the broker to call close_all() while the pollset contained
// further activities for the broker
return false;
// keep a strong reference to our parent until we leave scope
// to avoid UB when becoming detached during invocation
auto guard = parent_;
auto& buf = rd_buf();
CAF_ASSERT(buf.size() >= num_bytes);
// make sure size is correct, swap into message, and then call client
buf.resize(num_bytes);
auto& msg_buf = msg().buf;
msg_buf.swap(buf);
auto result = invoke_mailbox_element(ctx);
// swap buffer back to stream and implicitly flush wr_buf()
msg_buf.swap(buf);
// flush(); // <-- TODO: here from scribe, not sure why?
return result;
}
void endpoint::datagram_sent(execution_unit* ctx, size_t written) {
CAF_LOG_TRACE(CAF_ARG(written));
if (detached())
return;
using sent_t = datagram_sent_msg;
using tmp_t = mailbox_element_vals<datagram_sent_msg>;
tmp_t tmp{strong_actor_ptr{}, message_id::make(),
mailbox_element::forwarding_stack{},
sent_t{hdl(), written}};
invoke_mailbox_element_impl(ctx, tmp);
}
void endpoint::io_failure(execution_unit* ctx, network::operation op) {
CAF_LOG_TRACE(CAF_ARG(hdl()) << CAF_ARG(op));
// keep compiler happy when compiling w/o logging
static_cast<void>(op);
detach(ctx, true);
}
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#include "caf/io/network/endpoint_manager.hpp"
namespace caf {
namespace io {
namespace network {
endpoint_manager::endpoint_manager(abstract_broker* ptr)
: manager(ptr) {
// nop
}
endpoint_manager::~endpoint_manager() {
// nop
}
} // namespace network
} // namespace io
} // namespace caf
......@@ -23,6 +23,7 @@
#include "caf/io/scribe.hpp"
#include "caf/io/doorman.hpp"
#include "caf/io/endpoint.hpp"
#include "caf/io/datagram_sink.hpp"
#include "caf/io/datagram_source.hpp"
......@@ -214,6 +215,120 @@ test_multiplexer::add_tcp_doorman(abstract_broker* ptr, uint16_t prt,
return result;
}
expected<endpoint_handle>
test_multiplexer::new_remote_endpoint(const std::string& host,
uint16_t port_hint) {
guard_type guard{mx_};
endpoint_handle result;
auto i = remote_endpoints_.find(std::make_pair(host, port_hint));
if (i != remote_endpoints_.end()) {
result = i->second;
remote_endpoints_.erase(i);
}
return result;
}
expected<std::pair<endpoint_handle, uint16_t>>
test_multiplexer::new_local_endpoint(uint16_t desired_prt, const char*, bool) {
endpoint_handle result;
auto i = local_endpoints_.find(desired_prt);
if (i != local_endpoints_.end()) {
result = i->second;
local_endpoints_.erase(i);
}
return std::make_pair(result, desired_prt);
}
expected<void> test_multiplexer::assign_endpoint(abstract_broker* ptr,
endpoint_handle hdl) {
class impl : public endpoint {
public:
impl(abstract_broker* self, endpoint_handle eh, test_multiplexer* mpx)
: endpoint(self, eh),
mpx_(mpx) {
// nop
}
void configure_datagram_size(size_t buf_size) override {
mpx_->buffer_size(hdl()) = buf_size;
}
void ack_writes(bool enable) override {
mpx_->ack_writes(hdl()) = enable;
}
std::vector<char>& wr_buf() override {
return mpx_->output_buffer(hdl());
}
std::vector<char>& rd_buf() override {
return mpx_->input_buffer(hdl());
}
void launch() override {
// nop
}
void stop_reading() override {
mpx_->stopped_reading(hdl()) = true;
detach(mpx_, false);
}
std::string addr() const override {
return "test";
}
uint16_t port() const override {
return static_cast<uint16_t>(hdl().id());
}
void add_to_loop() override {
mpx_->passive_mode(hdl()) = false;
}
void remove_from_loop() override {
mpx_->passive_mode(hdl()) = true;
}
private:
test_multiplexer* mpx_;
};
CAF_LOG_TRACE(CAF_ARG(hdl));
auto sptr = make_counted<impl>(ptr, hdl, this);
impl_ptr(hdl) = sptr;
ptr->add_endpoint(sptr);
return unit;
}
expected<endpoint_handle>
test_multiplexer::add_remote_endpoint(abstract_broker* ptr,
const std::string& host, uint16_t port) {
auto hdl = new_remote_endpoint(host, port);
if (!hdl)
return std::move(hdl.error());
assign_endpoint(ptr, *hdl);
return hdl;
}
expected<std::pair<endpoint_handle, uint16_t>>
test_multiplexer::add_local_endpoint(abstract_broker* ptr, uint16_t port,
const char* in, bool reuse_addr) {
auto result = new_local_endpoint(port, in, reuse_addr);
if (!result)
return std::move(result.error());
local_port(result->first) = port;
assign_endpoint(ptr, result->first);
return result;
}
endpoint_handle test_multiplexer::add_endpoint(abstract_broker*,
network::native_socket) {
std::cerr << "test_multiplexer::add_endpoint called with native socket"
<< std::endl;
abort();
}
expected<datagram_sink_handle>
test_multiplexer::new_datagram_sink(const std::string& host,
uint16_t port_hint) {
......@@ -486,6 +601,14 @@ size_t& test_multiplexer::buffer_size(datagram_source_handle hdl) {
return datagram_source_data_[hdl].buffer_size;
}
uint16_t& test_multiplexer::local_port(endpoint_handle hdl) {
return endpoint_data_[hdl].local_port;
}
uint16_t& test_multiplexer::remote_port(endpoint_handle hdl) {
return endpoint_data_[hdl].remote_port;
}
bool& test_multiplexer::stopped_reading(accept_handle hdl) {
return doorman_data_[hdl].stopped_reading;
}
......@@ -498,6 +621,36 @@ intrusive_ptr<doorman>& test_multiplexer::impl_ptr(accept_handle hdl) {
return doorman_data_[hdl].ptr;
}
test_multiplexer::buffer_type&
test_multiplexer::output_buffer(endpoint_handle hdl) {
return endpoint_data_[hdl].wr_buf;
}
test_multiplexer::buffer_type&
test_multiplexer::input_buffer(endpoint_handle hdl) {
return endpoint_data_[hdl].re_buf;
}
intrusive_ptr<endpoint>& test_multiplexer::impl_ptr(endpoint_handle hdl) {
return endpoint_data_[hdl].ptr;
}
bool& test_multiplexer::stopped_reading(endpoint_handle hdl) {
return endpoint_data_[hdl].stopped_reading;
}
bool& test_multiplexer::passive_mode(endpoint_handle hdl) {
return endpoint_data_[hdl].passive_mode;
}
bool& test_multiplexer::ack_writes(endpoint_handle hdl) {
return endpoint_data_[hdl].ack_writes;
}
size_t& test_multiplexer::buffer_size(endpoint_handle hdl) {
return endpoint_data_[hdl].re_buf_size;
}
void test_multiplexer::add_pending_connect(accept_handle src,
connection_handle hdl) {
pending_connects_.emplace(src, hdl);
......
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