Commit 211c30f4 authored by Joseph Noir's avatar Joseph Noir

Reworking datagram management types

parent 9971d016
......@@ -23,12 +23,10 @@ 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
src/datagram_source.cpp
src/dgram_scribe.cpp
src/dgram_doorman.cpp
src/dgram_acceptor_manager.cpp
src/dgram_communicator_manager.cpp
# BASP files
src/header.cpp
src/message_type.cpp
......@@ -72,4 +70,3 @@ include_directories(. ${INCLUDE_DIRS})
if(NOT WIN32)
install(DIRECTORY caf/ DESTINATION include/caf FILES_MATCHING PATTERN "*.hpp")
endif()
......@@ -30,18 +30,16 @@
#include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/endpoint_handle.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/datagram_sink_handle.hpp"
#include "caf/io/datagram_source_handle.hpp"
#include "caf/io/dgram_scribe_handle.hpp"
#include "caf/io/dgram_doorman_handle.hpp"
#include "caf/io/network/native_socket.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"
#include "caf/io/network/dgram_communicator_manager.hpp"
#include "caf/io/network/dgram_acceptor_manager.hpp"
namespace caf {
namespace io {
......@@ -88,9 +86,8 @@ public:
// even brokers need friends
friend class scribe;
friend class doorman;
friend class endpoint;
friend class datagram_sink;
friend class datagram_source;
friend class dgram_scribe;
friend class dgram_doorman;
// -- overridden modifiers of abstract_actor ---------------------------------
......@@ -163,32 +160,18 @@ public:
void flush(connection_handle hdl);
/// Enables or disables write notifications for given datagram socket.
void ack_writes(datagram_sink_handle hdl, bool enable);
void ack_writes(dgram_scribe_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(datagram_source_handle hdl, size_t buf_size);
void configure_datagram_size(dgram_doorman_handle hdl, size_t buf_size);
/// Returns write buffer for given sink.
std::vector<char>& wr_buf(datagram_sink_handle hdl);
std::vector<char>& wr_buf(dgram_scribe_handle hdl);
/// 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);
void write(dgram_scribe_handle hdl, size_t data_size, const void* data);
/// Returns the middleman instance this broker belongs to.
inline middleman& parent() {
......@@ -229,58 +212,40 @@ public:
/// Creates and assigns a new `doorman` from given native socked `fd`.
expected<accept_handle> add_tcp_doorman(network::native_socket fd);
/// Adds a `datagram_sink` instance to this broker.
void add_datagram_sink(const intrusive_ptr<datagram_sink>& ptr);
/// Adds a `dgram_scribe` instance to this broker.
void add_dgram_scribe(const intrusive_ptr<dgram_scribe>& ptr);
/// Tries to create a datgram sink for `host` on given `port` and creates
/// a new datagram sink describing the endpoint afterwards.
/// @returns The handle of the new `datagram_sink` on success.
expected<datagram_sink_handle> add_datagram_sink(const std::string& host,
/// @returns The handle of the new `dgram_scribe` on success.
expected<dgram_scribe_handle> add_dgram_scribe(const std::string& host,
uint16_t port);
/// Assigns a detached `datagram_sink` instance identified by `hdl`
/// Assigns a detached `dgram_scribe` instance identified by `hdl`
/// from the `multiplexer` to this broker.
expected<void> assign_datagram_sink(datagram_sink_handle hdl);
expected<void> assign_dgram_scribe(dgram_scribe_handle hdl);
/// Creates and assigns a new `datagram_sink` from given native socked `fd`.
expected<datagram_sink_handle> add_datagram_sink(network::native_socket fd);
/// Creates and assigns a new `dgram_scribe` from given native socked `fd`.
expected<dgram_scribe_handle> add_dgram_scribe(network::native_socket fd);
/// Adds a `datagram_source` instance to this broker.
void add_datagram_source(const intrusive_ptr<datagram_source>& ptr);
/// Adds a `dgram_doorman` instance to this broker.
void add_dgram_doorman(const intrusive_ptr<dgram_doorman>& ptr);
/// Tries to open a local port and creates a `datagram_source` managing
/// Tries to open a local port and creates a `dgram_doorman` managing
/// it on success. If `port == 0`, then the broker will ask
/// the operating system to pick a random port.
/// @returns The handle of the new `datagram_source` and the assigned port.
expected<std::pair<datagram_source_handle, uint16_t>>
add_datagram_source(uint16_t port = 0, const char* in = nullptr,
/// @returns The handle of the new `dgram_doorman` and the assigned port.
expected<std::pair<dgram_doorman_handle, uint16_t>>
add_dgram_doorman(uint16_t port = 0, const char* in = nullptr,
bool reuse_addr = false);
/// Assigns a detached `datagram_source` instance identified by `hdl`
/// Assigns a detached `dgram_doorman` instance identified by `hdl`
/// from the `multiplexer` to this broker.
expected<void> assign_datagram_source(datagram_source_handle hdl);
/// Creates and assigns a new `datagram_source` from given native socked `fd`.
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);
expected<void> assign_dgram_doorman(dgram_doorman_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);
/// Creates and assigns a new `dgram_doorman` from given native socked `fd`.
expected<dgram_doorman_handle>
add_dgram_doorman(network::native_socket fd);
/// Returns the remote address associated to `hdl`
/// or empty string if `hdl` is invalid.
......@@ -302,31 +267,14 @@ public:
/// Returns the remote address associated to `hdl`
/// or empty string if `hdl` is invalid.
std::string remote_addr(datagram_sink_handle hdl);
std::string remote_addr(dgram_scribe_handle hdl);
/// Returns the remote port associated to `hdl`
/// or `0` if `hdl` is invalid.
uint16_t remote_port(datagram_sink_handle hdl);
uint16_t remote_port(dgram_scribe_handle hdl);
/// 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);
*/
uint16_t local_port(dgram_doorman_handle hdl);
/// Closes all connections and acceptors.
void close_all();
......@@ -386,14 +334,11 @@ protected:
using scribe_map = std::unordered_map<connection_handle,
intrusive_ptr<scribe>>;
using datagram_sink_map = std::unordered_map<datagram_sink_handle,
intrusive_ptr<datagram_sink>>;
using dgram_scribe_map = std::unordered_map<dgram_scribe_handle,
intrusive_ptr<dgram_scribe>>;
using datagram_source_map = std::unordered_map<datagram_source_handle,
intrusive_ptr<datagram_source>>;
using endpoint_map = std::unordered_map<endpoint_handle,
intrusive_ptr<endpoint>>;
using dgram_doorman_map = std::unordered_map<dgram_doorman_handle,
intrusive_ptr<dgram_doorman>>;
/// @cond PRIVATE
......@@ -408,18 +353,13 @@ protected:
}
// meta programming utility
inline datagram_sink_map& get_map(datagram_sink_handle) {
return datagram_sinks_;
}
// meta programming utility
inline datagram_source_map& get_map(datagram_source_handle) {
return datagram_sources_;
inline dgram_scribe_map& get_map(dgram_scribe_handle) {
return dgram_scribes_;
}
// meta programming utility
inline endpoint_map& get_map(endpoint_handle) {
return endpoints_;
inline dgram_doorman_map& get_map(dgram_doorman_handle) {
return dgram_doormans_;
}
// meta programming utility (not implemented)
......@@ -429,20 +369,17 @@ protected:
static intrusive_ptr<scribe> ptr_of(connection_handle);
// meta programming utility (not implemented)
static intrusive_ptr<datagram_sink> ptr_of(datagram_sink_handle);
// meta programming utility (not implemented)
static intrusive_ptr<datagram_source> ptr_of(datagram_source_handle);
static intrusive_ptr<dgram_scribe> ptr_of(dgram_scribe_handle);
// meta programming utility (not implemented)
static intrusive_ptr<endpoint> ptr_of(endpoint_handle);
static intrusive_ptr<dgram_doorman> ptr_of(dgram_doorman_handle);
/// @endcond
/// Returns the `multiplexer` running this broker.
network::multiplexer& backend();
/// Returns a `scribe`, `doorman`, `datagram_sink` or `datagram_source`
/// Returns a `scribe`, `doorman`, `dgram_scribe` or `dgram_doorman`
/// identified by `hdl`.
template <class Handle>
auto by_id(Handle hdl) -> optional<decltype(*ptr_of(hdl))> {
......@@ -453,8 +390,8 @@ protected:
return *(i->second);
}
/// Returns an intrusive pointer to a `scribe`, `doorman`, `datagram_sink`
/// or `datagram_source` identified by `hdl` and remove it from this broker.
/// Returns an intrusive pointer to a `scribe`, `doorman`, `dgram_scribe`
/// or `dgram_doorman` identified by `hdl` and remove it from this broker.
template <class Handle>
auto take(Handle hdl) -> decltype(ptr_of(hdl)) {
using std::swap;
......@@ -471,9 +408,8 @@ protected:
private:
scribe_map scribes_;
doorman_map doormen_;
endpoint_map endpoints_;
datagram_sink_map datagram_sinks_;
datagram_source_map datagram_sources_;
dgram_scribe_map dgram_scribes_;
dgram_doorman_map dgram_doormans_;
detail::intrusive_partitioned_list<mailbox_element, detail::disposer> cache_;
std::vector<char> dummy_wr_buf_;
};
......
......@@ -115,7 +115,7 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// our current processed BSAP header
basp::header hdr;
// local source
endpoint_handle hdl;
dgram_scribe_handle hdl;
// network-agnositc node identifier
node_id id;
// ports
......@@ -127,7 +127,6 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
};
void set_context(connection_handle hdl);
void set_context(endpoint_handle hdl);
// pointer to ourselves
broker* self;
......@@ -137,11 +136,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// keeps context information for all open connections
std::unordered_map<connection_handle, connection_context> tcp_ctx;
std::unordered_map<endpoint_handle, endpoint_context> udp_ctx;
// points to the current context for callbacks such as `make_proxy`
connection_context* this_context = nullptr;
endpoint_context* udp_context = nullptr;
// stores handles to spawn servers for other nodes; these servers
// are spawned whenever the broker learns a new node ID and try to
......
......@@ -30,10 +30,9 @@
#include "caf/io/scribe.hpp"
#include "caf/io/doorman.hpp"
#include "caf/io/endpoint.hpp"
#include "caf/io/datagram_sink.hpp"
#include "caf/io/dgram_doorman.hpp"
#include "caf/io/abstract_broker.hpp"
#include "caf/io/datagram_source.hpp"
#include "caf/io/dgram_scribe.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp"
......
......@@ -29,8 +29,8 @@
namespace caf {
namespace io {
/// Base class for `scribe` and `doorman` as well as `datagram_source`
/// and `datagram_sink`.
/// Base class for `scribe` and `doorman` as well as `dgram_doorman`
/// and `dgram_scribe`.
/// @ingroup Broker
template <class Base, class Handle, class SysMsgType>
class broker_servant : public Base {
......@@ -106,13 +106,9 @@ protected:
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
std::is_same<Handle, dgram_scribe_handle>::value,
dgram_scribe_passivated_msg,
dgram_doorman_passivated_msg
>::type
>::type
>::type;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#ifndef CAF_IO_DATAGRAM_SINK_HPP
#define CAF_IO_DATAGRAM_SINK_HPP
#include <vector>
#include "caf/message.hpp"
#include "caf/io/broker_servant.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/datagram_sink_handle.hpp"
#include "caf/io/network/datagram_sink_manager.hpp"
namespace caf {
namespace io {
using datagram_sink_base = broker_servant<network::datagram_sink_manager,
datagram_sink_handle,
datagram_sink_msg>;
/// Manages writing to a datagram sink.
/// @ingroup Broker
class datagram_sink : public datagram_sink_base {
public:
datagram_sink(abstract_broker* parent, datagram_sink_handle hdl);
~datagram_sink();
/// Enables or disables write notifications.
virtual void ack_writes(bool enable) = 0;
/// Returns the current write buffer.
virtual std::vector<char>& wr_buf() = 0;
void datagram_sent(execution_unit* ctx, size_t num_bytes) override;
void io_failure(execution_unit* ctx, network::operation op) override;
protected:
message detach_message() override;
};
} // namespace io
} // namespace caf
#endif // CAF_IO_DATAGRAM_SINK_HPP
......@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_DATAGRAM_SOURCE_HPP
#define CAF_IO_DATAGRAM_SOURCE_HPP
#ifndef CAF_IO_DGRAM_DOORMAN_HPP
#define CAF_IO_DGRAM_DOORMAN_HPP
#include <vector>
......@@ -26,23 +26,23 @@
#include "caf/io/broker_servant.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/datagram_source_handle.hpp"
#include "caf/io/network/datagram_source_manager.hpp"
#include "caf/io/dgram_doorman_handle.hpp"
#include "caf/io/network/dgram_communicator_manager.hpp"
namespace caf {
namespace io {
using datagram_source_base = broker_servant<network::datagram_source_manager,
datagram_source_handle,
new_datagram_msg>;
using dgram_doorman_base = broker_servant<network::dgram_acceptor_manager,
dgram_doorman_handle,
new_endpoint_msg>;
/// Manages reading from a datagram source
/// @ingroup Broker
class datagram_source : public datagram_source_base {
class dgram_doorman : public dgram_doorman_base {
public:
datagram_source(abstract_broker* parent, datagram_source_handle hdl);
dgram_doorman(abstract_broker* parent, dgram_doorman_handle hdl);
~datagram_source();
~dgram_doorman();
/// Configure buffer size for next accepted datagram.
virtual void configure_datagram_size(size_t buf_size) = 0;
......@@ -50,10 +50,12 @@ public:
/// Returns the current input buffer.
virtual std::vector<char>& rd_buf() = 0;
bool consume(execution_unit* ctx, const void* buf, size_t besize) override;
void io_failure(execution_unit* ctx, network::operation op) override;
using dgram_doorman_base::new_endpoint;
bool new_endpoint(execution_unit* ctx, const void*, size_t num_bytes);
// needs to be launched explicitly
virtual void launch() = 0;
......@@ -64,5 +66,4 @@ protected:
} // namespace io
} // namespace caf
#endif // CAF_IO_DATAGRAM_SOURCE_HPP
#endif // CAF_IO_DGRAM_DOORMAN_HPP
......@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_DATAGRAM_SOURCE_HANDLE_HPP
#define CAF_IO_DATAGRAM_SOURCE_HANDLE_HPP
#ifndef CAF_IO_DGRAM_DOORMAN_HANDLE_HPP
#define CAF_IO_DGRAM_DOORMAN_HANDLE_HPP
#include <functional>
......@@ -31,40 +31,40 @@
namespace caf {
namespace io {
struct invalid_datagram_source_handle_t {
constexpr invalid_datagram_source_handle_t() {
struct invalid_dgram_doorman_handle_t {
constexpr invalid_dgram_doorman_handle_t() {
// nop
}
};
constexpr invalid_datagram_source_handle_t invalid_datagram_source_handle
= invalid_datagram_source_handle_t{};
constexpr invalid_dgram_doorman_handle_t invalid_dgram_doorman_handle
= invalid_dgram_doorman_handle_t{};
/// Generic type for identifying a datagram source.
class datagram_source_handle : public handle<datagram_source_handle,
invalid_datagram_source_handle_t> {
class dgram_doorman_handle : public handle<dgram_doorman_handle,
invalid_dgram_doorman_handle_t> {
public:
friend class handle<datagram_source_handle, invalid_datagram_source_handle_t>;
friend class handle<dgram_doorman_handle, invalid_dgram_doorman_handle_t>;
using super = handle<datagram_source_handle, invalid_datagram_source_handle_t>;
using super = handle<dgram_doorman_handle, invalid_dgram_doorman_handle_t>;
constexpr datagram_source_handle() {
constexpr dgram_doorman_handle() {
// nop
}
constexpr datagram_source_handle(const invalid_datagram_source_handle_t&) {
constexpr dgram_doorman_handle(const invalid_dgram_doorman_handle_t&) {
// nop
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
datagram_source_handle& x) {
return f(meta::type_name("datagram_source_handle"), x.id_);
dgram_doorman_handle& x) {
return f(meta::type_name("dgram_doorman_handle"), x.id_);
}
private:
inline datagram_source_handle(int64_t handle_id) : super(handle_id) {
inline dgram_doorman_handle(int64_t handle_id) : super(handle_id) {
// nop
}
};
......@@ -75,12 +75,12 @@ private:
namespace std {
template<>
struct hash<caf::io::datagram_source_handle> {
size_t operator()(const caf::io::datagram_source_handle& hdl) const {
struct hash<caf::io::dgram_doorman_handle> {
size_t operator()(const caf::io::dgram_doorman_handle& hdl) const {
hash<int64_t> f;
return f(hdl.id());
}
};
} // namespace std
#endif // CAF_IO_DATAGRAM_SOURCE_HANDLE_HPP
#endif // CAF_IO_DGRAM_DOORMAN_HANDLE_HPP
......@@ -17,61 +17,60 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_ENDPOINT_HPP
#define CAF_IO_ENDPOINT_HPP
#ifndef CAF_IO_DGRAM_SCRIBE_HPP
#define CAF_IO_DGRAM_SCRIBE_HPP
#include <vector>
#include "caf/message.hpp"
#include "caf/io/broker_servant.hpp"
#include "caf/io/endpoint_handle.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/network/endpoint_manager.hpp"
#include "caf/io/dgram_scribe_handle.hpp"
#include "caf/io/network/dgram_acceptor_manager.hpp"
namespace caf {
namespace io {
using endpoint_base = broker_servant<network::endpoint_manager, endpoint_handle,
datagram_msg>;
using dgram_scribe_base = broker_servant<network::dgram_communicator_manager,
dgram_scribe_handle,
new_datagram_msg>;
/// Manages writing and reading on a datagram endpoint.
/// Manages writing to a datagram sink.
/// @ingroup Broker
class endpoint : public endpoint_base {
class dgram_scribe : public dgram_scribe_base {
public:
endpoint(abstract_broker* parent, endpoint_handle hdl);
dgram_scribe(abstract_broker* parent, dgram_scribe_handle hdl);
~endpoint();
/// Enables or disables write notifications.
virtual void ack_writes(bool enable) = 0;
~dgram_scribe();
/// 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.
/// Enables or disables write notifications.
virtual void ack_writes(bool enable) = 0;
/// Returns the current output buffer.
virtual std::vector<char>& wr_buf() = 0;
/// Returns the current input buffer.
virtual std::vector<char>& rd_buf() = 0;
bool consume(execution_unit* ctx, const void* buf, size_t besize) override;
void datagram_sent(execution_unit* ctx, size_t num_bytes) override;
/// Flushes the output buffer, i.e., sends the
/// content of the buffer via the network.
virtual void flush() = 0;
void io_failure(execution_unit* ctx, network::operation op) override;
// needs to be launched explicitly, TODO: Does it?
// Can't configure_datagram_size do that?
virtual void launch() = 0;
bool consume(execution_unit*, const void*, size_t) override;
void datagram_sent(execution_unit*, size_t) override;
protected:
message detach_message() override;
};
} // namespace io
} // namespace caf
#endif // CAF_IO_ENDPOINT_HPP
#endif // CAF_IO_DGRAM_SCRIBE_HPP
......@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_DATAGRAM_SINK_HANDLE_HPP
#define CAF_IO_DATAGRAM_SINK_HANDLE_HPP
#ifndef CAF_IO_DGRAM_SCRIBE_HANDLE_HPP
#define CAF_IO_DGRAM_SCRIBE_HANDLE_HPP
#include <functional>
......@@ -31,36 +31,36 @@
namespace caf {
namespace io {
struct invalid_datagram_sink_handle_t {
constexpr invalid_datagram_sink_handle_t() {
struct invalid_dgram_scribe_handle_t {
constexpr invalid_dgram_scribe_handle_t() {
// nop
}
};
constexpr invalid_datagram_sink_handle_t invalid_datagram_sink_handle
= invalid_datagram_sink_handle_t{};
constexpr invalid_dgram_scribe_handle_t invalid_dgram_scribe_handle
= invalid_dgram_scribe_handle_t{};
/// Generic type for identifying datagram sink.
class datagram_sink_handle : public handle<datagram_sink_handle,
invalid_datagram_sink_handle_t> {
class dgram_scribe_handle : public handle<dgram_scribe_handle,
invalid_dgram_scribe_handle_t> {
public:
friend class handle<datagram_sink_handle, invalid_datagram_sink_handle_t>;
friend class handle<dgram_scribe_handle, invalid_dgram_scribe_handle_t>;
using super = handle<datagram_sink_handle, invalid_datagram_sink_handle_t>;
using super = handle<dgram_scribe_handle, invalid_dgram_scribe_handle_t>;
datagram_sink_handle() : port_{0} {
dgram_scribe_handle() : port_{0} {
// nop
}
datagram_sink_handle(const invalid_datagram_sink_handle_t&) : port_{0} {
dgram_scribe_handle(const invalid_dgram_scribe_handle_t&) : port_{0} {
// nop
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
datagram_sink_handle& x) {
return f(meta::type_name("datagram_sink_handle"), x.id_);
dgram_scribe_handle& x) {
return f(meta::type_name("dgram_scribe_handle"), x.id_);
}
const std::string& host() const {
......@@ -80,7 +80,7 @@ public:
}
private:
inline datagram_sink_handle(int64_t handle_id) : super(handle_id) {
inline dgram_scribe_handle(int64_t handle_id) : super(handle_id) {
// nop
}
std::string host_;
......@@ -93,12 +93,12 @@ private:
namespace std {
template<>
struct hash<caf::io::datagram_sink_handle> {
size_t operator()(const caf::io::datagram_sink_handle& hdl) const {
struct hash<caf::io::dgram_scribe_handle> {
size_t operator()(const caf::io::dgram_scribe_handle& hdl) const {
hash<int64_t> f;
return f(hdl.id());
}
};
} // namespace std
#endif // CAF_IO_DATAGRAM_SINK_HANDLE_HPP
#endif // CAF_IO_DGRAM_SCRIBE_HANDLE_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#ifndef CAF_IO_ENDPOINT_HANDLE_HPP
#define CAF_IO_ENDPOINT_HANDLE_HPP
#include "caf/error.hpp"
#include "caf/io/handle.hpp"
#include "caf/meta/type_name.hpp"
namespace caf {
namespace io {
struct invalid_endpoint_handle_t {
constexpr invalid_endpoint_handle_t() {
// nop
}
};
constexpr invalid_endpoint_handle_t invalid_endpoint_handle
= invalid_endpoint_handle_t{};
/// Generic handle type for managing local and remote datagram endpoints.
class endpoint_handle : public handle<endpoint_handle,
invalid_endpoint_handle_t> {
public:
friend class handle<endpoint_handle, invalid_endpoint_handle_t>;
using super = handle<endpoint_handle,invalid_endpoint_handle_t>;
constexpr endpoint_handle() {
// nop
}
constexpr endpoint_handle(const invalid_endpoint_handle_t&) {
// nop
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
endpoint_handle& x) {
return f(meta::type_name("endpoint_handle"), x.id_);
}
private:
inline endpoint_handle(int64_t handle_id) : super(handle_id) {
// nop
}
};
} // namespace io
} // namespace caf
namespace std {
template<>
struct hash<caf::io::endpoint_handle> {
size_t operator()(const caf::io::endpoint_handle& hdl) const {
hash<int64_t> f;
return f(hdl.id());
}
};
} // namespace std
#endif // CAF_IO_ENDPOINT_HANDLE_HPP
......@@ -26,13 +26,12 @@ namespace io {
class scribe;
class broker;
class doorman;
class endpoint;
class middleman;
class basp_broker;
class datagram_sink;
class dgram_doorman;
class receive_policy;
class abstract_broker;
class datagram_source;
class dgram_scribe;
template <class... Sigs>
class typed_broker;
......
......@@ -38,9 +38,8 @@
#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"
#include "caf/io/network/dgram_communicator_manager.hpp"
#include "caf/io/network/dgram_acceptor_manager.hpp"
#include "caf/io/network/native_socket.hpp"
......@@ -255,8 +254,6 @@ public:
friend class io::middleman; // disambiguate reference
friend class supervisor;
using endpoint_addr = std::pair<std::string,uint16_t>;
struct event {
native_socket fd;
int mask;
......@@ -301,52 +298,32 @@ public:
add_tcp_doorman(abstract_broker* ptr, uint16_t port, const char* in,
bool rflag) override;
expected<datagram_sink_handle> new_datagram_sink(const std::string& host,
expected<dgram_scribe_handle> new_dgram_scribe(const std::string& host,
uint16_t port) override;
expected<void> assign_datagram_sink(abstract_broker* ptr,
datagram_sink_handle hdl) override;
expected<void> assign_dgram_scribe(abstract_broker* ptr,
dgram_scribe_handle hdl) override;
datagram_sink_handle add_datagram_sink(abstract_broker* ptr,
dgram_scribe_handle add_dgram_scribe(abstract_broker* ptr,
native_socket fd) override;
expected<datagram_sink_handle> add_datagram_sink(abstract_broker* ptr,
expected<dgram_scribe_handle> add_dgram_scribe(abstract_broker* ptr,
const std::string& host,
uint16_t port) override;
expected<std::pair<datagram_source_handle, uint16_t>>
new_datagram_source(uint16_t port, const char* in, bool rflag) override;
expected<std::pair<dgram_doorman_handle, uint16_t>>
new_dgram_doorman(uint16_t port, const char* in, bool rflag) override;
expected<void> assign_datagram_source(abstract_broker* ptr,
datagram_source_handle hdl) override;
expected<void> assign_dgram_doorman(abstract_broker* ptr,
dgram_doorman_handle hdl) override;
datagram_source_handle add_datagram_source(abstract_broker* ptr,
dgram_doorman_handle add_dgram_doorman(abstract_broker* ptr,
native_socket fd) override;
expected<std::pair<datagram_source_handle, uint16_t>>
add_datagram_source(abstract_broker* ptr, uint16_t port , const char* in,
expected<std::pair<dgram_doorman_handle, uint16_t>>
add_dgram_doorman(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);
......@@ -361,13 +338,6 @@ public:
void del(operation op, native_socket fd, event_handler* ptr);
/// @cond PRIVATE
// Used by datagram senders and receivers to manage known endpoints
std::map<endpoint_addr, endpoint_handle>& endpoints();
/// @endcond
private:
// platform-dependent additional initialization code
void init();
......@@ -427,9 +397,6 @@ private:
multiplexer_poll_shadow_data shadow_;
std::pair<native_socket, native_socket> pipe_;
pipe_reader pipe_reader_;
// TODO: is this the right place?
// How to maintain endpoints if they close?
std::map<endpoint_addr, endpoint_handle> remote_endpoints_;
};
inline connection_handle conn_hdl_from_socket(native_socket fd) {
......@@ -440,16 +407,12 @@ inline accept_handle accept_hdl_from_socket(native_socket fd) {
return accept_handle::from_int(int64_from_native_socket(fd));
}
inline datagram_sink_handle dg_sink_hdl_from_socket(native_socket fd) {
return datagram_sink_handle::from_int(int64_from_native_socket(fd));
inline dgram_scribe_handle dg_sink_hdl_from_socket(native_socket fd) {
return dgram_scribe_handle::from_int(int64_from_native_socket(fd));
}
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));
inline dgram_doorman_handle dg_source_hdl_from_socket(native_socket fd) {
return dgram_doorman_handle::from_int(int64_from_native_socket(fd));
}
/// A stream capable of both reading and writing. The stream's input
......@@ -569,20 +532,36 @@ private:
native_socket sock_;
};
class datagram_handler : public event_handler {
/// A dgram_communicator is responsible for sending datagrams to an endpoint.
class dgram_communicator: public event_handler {
public:
/// A manager type providing the TODO
using manager_type = endpoint_manager;
/// A manger providing the TODO
using manager_type = dgram_communicator_manager;
/// A smart pointer to an endpoint_manager.
using manager_ptr = intrusive_ptr<endpoint_manager>;
/// A smart pointer to a datagram sink manger
using manager_ptr = intrusive_ptr<dgram_communicator_manager>;
/// A buffer class providing a compatible
/// interface to `std::vector`.
using buffer_type = std::vector<char>;
datagram_handler(default_multiplexer& backend_ref, native_socket sockfd);
dgram_communicator(default_multiplexer& backend_ref, native_socket sockfd);
/// 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);
/// Starts reading data from the socket, forwarding incoming data to `mgr`.
void start(manager_type* mgr);
/// Activates the stream.
void activate(manager_type* mgr);
/// Configures how much data will be provided for the next `consume` callback.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void configure_read(receive_policy::config config);
void ack_writes(bool x);
......@@ -590,26 +569,27 @@ public:
/// @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.
/// Returns the write buffer of this stream.
/// @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.
/// Returns the read buffer of this stream.
/// @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();
inline const std::string& host() const {
return host_;
}
inline uint16_t port() const {
return port_;
}
/// Sends the content of the write buffer, calling the `io_failure`
/// member function of `mgr` in case of an error.
......@@ -617,109 +597,55 @@ public:
/// 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.
/// Closes the read channel of the underlying socket and removes
/// this handler from its parent.
void stop_reading();
void removed_from_loop(operation op) override;
void handle_event(operation op) override;
void sender_from_sockaddr(sockaddr_storage& sa, size_t len);
private:
void prepare_next_read();
void prepare_next_write();
// state for receiving
// state for reading
manager_ptr reader_;
size_t buf_size_;
size_t packet_size_;
size_t dgram_size_;
buffer_type rd_buf_;
struct sockaddr_storage last_sender;
socklen_t sender_len;
// state for sending
// state for writing
manager_ptr writer_;
bool ack_writes_;
bool writing_;
size_t written_;
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:
/// A manger providing the TODO
using manager_type = datagram_sink_manager;
/// A smart pointer to a datagram sink manger
using manager_ptr = intrusive_ptr<datagram_sink_manager>;
/// A buffer class providing a compatible
/// interface to `std::vector`.
using buffer_type = std::vector<char>;
datagram_sender(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);
/// 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_;
}
/// 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_write();
manager_ptr writer_;
bool ack_writes_;
buffer_type wr_buf_;
buffer_type wr_offline_buf_;
// general state
struct sockaddr_storage remote_endpoint_;
socklen_t sockaddr_len_;
std::string host_;
uint16_t port_;
};
/// A datagram_receiver is responsible for receiving datagrams on an endpoint.
class datagram_receiver: public event_handler {
/// A dgram_acceptor is responsible for receiving datagrams on an endpoint.
class dgram_acceptor: public event_handler {
public:
/// A manger providing the TODO
using manager_type = datagram_source_manager;
/// A manger providing the new_endpoint function
using manager_type = dgram_acceptor_manager;
/// A smart pointer to a datagram sink manger
using manager_ptr = intrusive_ptr<datagram_source_manager>;
using manager_ptr = intrusive_ptr<dgram_acceptor_manager>;
/// A buffer class providing a compatible
/// interface to `std::vector`.
using buffer_type = std::vector<char>;
datagram_receiver(default_multiplexer& backend_ref, native_socket sockfd);
dgram_acceptor(default_multiplexer& backend_ref, native_socket sockfd);
/// Configures how much buffer will be provided for the next datagram.
/// @warning Must not be called outside the IO multiplexers event loop
......@@ -733,10 +659,18 @@ public:
return rd_buf_;
}
inline const std::string& host() const {
return host_;
}
inline uint16_t port() const {
return port_;
}
/// Starts reading data from the socket.
void start(manager_type* mgr);
/// Activates the datagram_receiver.
/// Activates the dgram_acceptor.
void activate(manager_type* mgr);
/// Closes the read channel of the underlying socket and removes
......@@ -747,17 +681,22 @@ public:
void handle_event(operation op) override;
std::pair<std::string,uint16_t> get_sender();
void sender_from_sockaddr(sockaddr_storage& sa, size_t len);
private:
void prepare_next_read();
// reader state
manager_ptr reader_;
size_t buf_size_;
size_t packet_size_;
size_t dgram_size_;
buffer_type rd_buf_;
struct sockaddr_storage last_sender;
socklen_t sender_len;
// general state
struct sockaddr_storage sockaddr_;
socklen_t sockaddr_len_;
std::string host_;
uint16_t port_;
native_socket sock_; // TODO: Do I need this?
};
expected<native_socket> new_tcp_connection(const std::string& host,
......@@ -768,18 +707,11 @@ expected<std::pair<native_socket, uint16_t>>
new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr);
expected<native_socket>
new_datagram_sink_impl(const std::string& host, uint16_t port,
optional<protocol> preferred = none);
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,
new_dgram_scribe_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);
new_dgram_doorman_impl(uint16_t port, const char* addr, bool reuse_addr);
} // namespace network
} // namespace io
......
......@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_NETWORK_DATAGRAM_SOURCE_MANGER_HPP
#define CAF_IO_NETWORK_DATAGRAM_SOURCE_MANGER_HPP
#ifndef CAF_IO_NETWORK_DGRAM_ACCEPTOR_MANGER_HPP
#define CAF_IO_NETWORK_DGRAM_ACCEPTOR_MANGER_HPP
#include "caf/io/network/manager.hpp"
......@@ -28,17 +28,18 @@ namespace network {
/// A datagram source manager provides callbacks for incoming
/// datagrams as well as for error handling.
class datagram_source_manager : public manager {
class dgram_acceptor_manager : public manager {
public:
datagram_source_manager(abstract_broker* ptr);
dgram_acceptor_manager(abstract_broker* ptr);
~datagram_source_manager();
~dgram_acceptor_manager();
virtual bool new_endpoint() = 0;
};
} // namespace network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_DATAGRAM_SOURCE_MANGER_HPP
#endif // CAF_IO_NETWORK_DGRAM_ACCEPTOR_MANGER_HPP
......@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_NETWORK_DATAGRAM_SINK_MANGER_HPP
#define CAF_IO_NETWORK_DATAGRAM_SINK_MANGER_HPP
#ifndef CAF_IO_NETWORK_DGRAM_COMMUNICATOR_MANGER_HPP
#define CAF_IO_NETWORK_DGRAM_COMMUNICATOR_MANGER_HPP
#include "caf/io/network/manager.hpp"
......@@ -28,22 +28,22 @@ namespace network {
/// A datagram manager provides callbacks for outgoing
/// datagrams as well as for error handling.
class datagram_sink_manager : public manager {
class dgram_communicator_manager : public manager {
public:
datagram_sink_manager(abstract_broker* ptr);
dgram_communicator_manager(abstract_broker* ptr);
~datagram_sink_manager();
~dgram_communicator_manager();
/// Called by the underlying I/O device whenever it received data.
/// @returns `true` if the manager accepts further reads, otherwise `false`.
virtual bool consume(execution_unit* ctx, const void* buf, size_t besize) = 0;
virtual bool consume(execution_unit*, const void*, size_t) = 0;
/// Called by the underlying I/O device whenever it sent a datagram.
virtual void datagram_sent(execution_unit* ctx, size_t num_bytes) = 0;
/// Called by the underlying I/O device whenever it sent data.
virtual void datagram_sent(execution_unit*, size_t) = 0;
};
} // namespace network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_DATAGRAM_SINK_MANGER_HPP
#endif // CAF_IO_NETWORK_DGRAM_COMMUNICATOR_MANGER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#ifndef CAF_IO_NETWORK_ENDPOINT_MANAGER_HPP
#define CAF_IO_NETWORK_ENDPOINT_MANAGER_HPP
#include "caf/io/network/manager.hpp"
namespace caf {
namespace io {
namespace network {
/// An endpoint manager configures datagram endpoints and provides callbacks
/// for incoming data as well as for error handling.
class endpoint_manager : public manager {
public:
endpoint_manager(abstract_broker* ptr);
~endpoint_manager();
/// Called by the underlying I/O device whenever it received data.
/// @returns `true` if the manager accepts further reads, otherwise `false`.
virtual bool consume(execution_unit* ctx, const void* buf, size_t bsize) = 0;
/// Called by the underlying I/O device whenever it sent a datagram.
virtual void datagram_sent(execution_unit* ctx, size_t num_bytes) = 0;
};
} // namespace network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_ENDPOINT_MANAGER_HPP
......@@ -32,10 +32,9 @@
#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"
#include "caf/io/dgram_scribe_handle.hpp"
#include "caf/io/dgram_doorman_handle.hpp"
#include "caf/io/network/protocol.hpp"
#include "caf/io/network/native_socket.hpp"
......@@ -104,78 +103,51 @@ public:
/// Tries to create a datagram sink to sent messages to `host` on given
/// `port` and returns an unbound datagram sink handle on success.
/// @threadsafe
virtual expected<datagram_sink_handle>
new_datagram_sink(const std::string& host, uint16_t port) = 0;
virtual expected<dgram_scribe_handle>
new_dgram_scribe(const std::string& host, uint16_t port) = 0;
/// Assigns an unbound datagram sink identified by `hdl` to `ptr`.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<void> assign_datagram_sink(abstract_broker* ptr,
datagram_sink_handle hdl) = 0;
virtual expected<void>
assign_dgram_scribe(abstract_broker* ptr,
dgram_scribe_handle hdl) = 0;
/// Creates a new datagram sink from a native socket handle.
/// @warning Do not call from outside the multiplexer's event loop.
virtual datagram_sink_handle add_datagram_sink(abstract_broker* ptr,
native_socket fd) = 0;
virtual dgram_scribe_handle
add_dgram_scribe(abstract_broker* ptr, native_socket fd) = 0;
/// Tries to create a datagram sink to sent messages to `host` on `port`
/// and returns a new datagram sink managing the endpoint on success.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<datagram_sink_handle>
add_datagram_sink(abstract_broker*, const std::string& host,
virtual expected<dgram_scribe_handle>
add_dgram_scribe(abstract_broker*, const std::string& host,
uint16_t port) = 0;
/// Tries to create an unbound datagram source bound to `port`, optionally
/// accepting only messages from IP address `in`.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<std::pair<datagram_source_handle, uint16_t>>
new_datagram_source(uint16_t port, const char* in = nullptr,
virtual expected<std::pair<dgram_doorman_handle, uint16_t>>
new_dgram_doorman(uint16_t port, const char* in = nullptr,
bool reuse_addr = false) = 0;
/// Assigns an unbound datagram source identified by `hdl` to `ptr`.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<void> assign_datagram_source(abstract_broker* ptr,
datagram_source_handle) = 0;
virtual expected<void> assign_dgram_doorman(abstract_broker* ptr,
dgram_doorman_handle) = 0;
/// Creates a new datagram source from a native socket handle.
/// @warning Do not call from outside the multiplexer's event loop.
virtual datagram_source_handle add_datagram_source(abstract_broker* ptr,
virtual dgram_doorman_handle add_dgram_doorman(abstract_broker* ptr,
native_socket fd) = 0;
/// Tries to create a new datagram source on port `p`, optionally
/// accepting only messages from IP address `in`.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<std::pair<datagram_source_handle, uint16_t>>
add_datagram_source(abstract_broker* ptr, uint16_t port,
virtual expected<std::pair<dgram_doorman_handle, uint16_t>>
add_dgram_doorman(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:
......
......@@ -59,53 +59,32 @@ public:
add_tcp_doorman(abstract_broker* ptr, uint16_t prt,
const char* in, bool reuse_addr) override;
expected<datagram_sink_handle> new_datagram_sink(const std::string& host,
expected<dgram_scribe_handle> new_dgram_scribe(const std::string& host,
uint16_t port) override;
expected<void> assign_datagram_sink(abstract_broker* ptr,
datagram_sink_handle hdl) override;
expected<void> assign_dgram_scribe(abstract_broker* ptr,
dgram_scribe_handle hdl) override;
datagram_sink_handle add_datagram_sink(abstract_broker* ptr,
dgram_scribe_handle add_dgram_scribe(abstract_broker* ptr,
native_socket fd) override;
expected<datagram_sink_handle> add_datagram_sink(abstract_broker* ptr,
expected<dgram_scribe_handle> add_dgram_scribe(abstract_broker* ptr,
const std::string& host,
uint16_t port) override;
expected<std::pair<datagram_source_handle, uint16_t>>
new_datagram_source(uint16_t port, const char* in, bool reuse_addr) override;
expected<std::pair<dgram_doorman_handle, uint16_t>>
new_dgram_doorman(uint16_t port, const char* in, bool reuse_addr) override;
expected<void> assign_datagram_source(abstract_broker* ptr,
datagram_source_handle hdl) override;
expected<void> assign_dgram_doorman(abstract_broker* ptr,
dgram_doorman_handle hdl) override;
datagram_source_handle add_datagram_source(abstract_broker* ptr,
dgram_doorman_handle add_dgram_doorman(abstract_broker* ptr,
native_socket fd) override;
expected<std::pair<datagram_source_handle, uint16_t>>
add_datagram_source(abstract_broker* ptr, uint16_t port,
expected<std::pair<dgram_doorman_handle, uint16_t>>
add_dgram_doorman(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;
......@@ -114,10 +93,10 @@ public:
void provide_acceptor(uint16_t port, accept_handle hdl);
void provide_datagram_sink(std::string host, uint16_t port,
datagram_sink_handle hdl);
void provide_dgram_scribe(std::string host, uint16_t port,
dgram_scribe_handle hdl);
void provide_datagram_source(uint16_t port, datagram_source_handle hdl);
void provide_dgram_doorman(uint16_t port, dgram_doorman_handle hdl);
/// A buffer storing bytes.
using buffer_type = std::vector<char>;
......@@ -130,19 +109,13 @@ public:
buffer_type& output_buffer(connection_handle hdl);
/// 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);
buffer_type& output_buffer(dgram_scribe_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);
buffer_type& input_buffer(dgram_doorman_handle hdl);
/// Returns the configured read policy of the scribe identified by `hdl`.
receive_policy::config& read_config(connection_handle hdl);
......@@ -163,49 +136,32 @@ public:
/// Returns `true` if this handle has been closed
/// for reading, `false` otherwise.
bool& stopped_reading(datagram_sink_handle hdl);
bool& stopped_reading(dgram_scribe_handle hdl);
/// Returns whether the scribe identified by `hdl` receives write ACKs.
bool& ack_writes(datagram_sink_handle hdl);
bool& ack_writes(dgram_scribe_handle hdl);
/// Returns `true` if this handle is inactive, otherwise `false`.
bool& passive_mode(datagram_sink_handle hdl);
bool& passive_mode(dgram_scribe_handle hdl);
intrusive_ptr<datagram_sink>& impl_ptr(datagram_sink_handle hdl);
intrusive_ptr<dgram_scribe>& impl_ptr(dgram_scribe_handle hdl);
uint16_t& port(datagram_sink_handle hdl);
uint16_t& port(dgram_scribe_handle hdl);
/// Returns `true` if this handle has been closed
/// for reading, `false` otherwise.
bool& stopped_reading(datagram_source_handle hdl);
bool& stopped_reading(dgram_doorman_handle hdl);
/// Returns `true` if this handle is inactive, otherwise `false`.
bool& passive_mode(datagram_source_handle hdl);
intrusive_ptr<datagram_source>& impl_ptr(datagram_source_handle hdl);
bool& passive_mode(dgram_doorman_handle hdl);
uint16_t& port(datagram_source_handle hdl);
intrusive_ptr<dgram_doorman>& impl_ptr(dgram_doorman_handle hdl);
size_t& buffer_size(datagram_source_handle hdl);
uint16_t& port(dgram_doorman_handle hdl);
uint16_t& local_port(endpoint_handle hdl);
size_t& buffer_size(dgram_doorman_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);
size_t& buffer_size(dgram_scribe_handle hdl);
/// Returns `true` if this handle has been closed
/// for reading, `false` otherwise.
......@@ -230,24 +186,16 @@ public:
bool has_pending_scribe(std::string host, uint16_t port);
using pending_datagram_sinks_map = std::map<std::pair<std::string, uint16_t>,
datagram_sink_handle>;
using pending_dgram_scribes_map = std::map<std::pair<std::string, uint16_t>,
dgram_scribe_handle>;
bool has_pending_datagram_sink(std::string host, uint16_t port);
bool has_pending_dgram_scribe(std::string host, uint16_t port);
using pending_datagram_sources_map
= std::map<uint16_t, datagram_source_handle>;
using pending_dgram_doormans_map
= std::map<uint16_t, dgram_doorman_handle>;
bool has_pending_datagram_source(uint16_t port);
bool has_pending_dgram_doorman(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);
......@@ -297,36 +245,25 @@ private:
intrusive_ptr<doorman> ptr;
};
struct datagram_sink_data {
struct dgram_scribe_data {
uint16_t port;
buffer_type xbuf;
buffer_type rd_buf;
buffer_type wr_buf;
size_t buffer_size;
bool stopped_reading = false;
bool passive_mode = false;
intrusive_ptr<dgram_scribe> ptr;
bool ack_writes = false;
intrusive_ptr<datagram_sink> ptr;
};
struct datagram_source_data {
struct dgram_doorman_data {
uint16_t port;
buffer_type rd_buf;
size_t buffer_size;
bool stopped_reading = false;
bool passive_mode = false;
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;
intrusive_ptr<dgram_doorman> ptr;
};
// guards resumables_ and scribes_
......@@ -334,16 +271,13 @@ private:
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_;
pending_dgram_scribes_map dgram_scribes_;
pending_dgram_doormans_map dgram_doormans_;
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_;
std::unordered_map<dgram_scribe_handle, dgram_scribe_data> dgram_scribe_data_;
std::unordered_map<dgram_doorman_handle, dgram_doorman_data> dgram_doorman_data_;
pending_connects_map pending_connects_;
};
......
......@@ -29,10 +29,9 @@
#include "caf/io/handle.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"
#include "caf/io/dgram_scribe_handle.hpp"
#include "caf/io/dgram_doorman_handle.hpp"
namespace caf {
namespace io {
......@@ -130,111 +129,61 @@ inspect(Inspector& f, acceptor_passivated_msg& x) {
}
/// Signalizes that a {@link broker} datagram sink was closed.
struct datagram_sink_closed_msg {
datagram_sink_handle handle;
struct dgram_scribe_closed_msg {
dgram_scribe_handle handle;
};
/// @relates datagram_sink_closed_msg
/// @relates dgram_scribe_closed_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, datagram_sink_closed_msg& x) {
return f(meta::type_name("datagram_sink_closed_msg"), x.handle);
inspect(Inspector& f, dgram_scribe_closed_msg& x) {
return f(meta::type_name("dgram_scribe_closed_msg"), x.handle);
}
/// Signalizes that a {@link broker} datagram source was closed.
struct datagram_source_closed_msg {
datagram_source_handle handle;
struct dgram_doorman_closed_msg {
dgram_doorman_handle handle;
};
/// @relates datagram_source_closed_msg
/// @relates dgram_doorman_closed_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, datagram_source_closed_msg& x) {
return f(meta::type_name("datagram_source_closed_msg"), x.handle);
inspect(Inspector& f, dgram_doorman_closed_msg& x) {
return f(meta::type_name("dgram_doorman_closed_msg"), x.handle);
}
/// Signalizes newly arrived datagram for a {@link broker}.
struct new_datagram_msg {
struct new_endpoint_msg {
/// Handle to the related datagram endpoint.
datagram_source_handle handle;
dgram_doorman_handle handle;
/// Buffer containing the received data.
std::vector<char> buf;
};
/// @relates new_datagram_msg
/// @relates new_endpoint_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, new_datagram_msg& x) {
return f(meta::type_name("new_datagram_msg"), x.handle, x.buf);
typename Inspector::result_type inspect(Inspector& f, new_endpoint_msg& x) {
return f(meta::type_name("new_endpoint_msg"), x.handle, x.buf);
}
/// Signalizes that a datagram with a certain size has been sent.
struct datagram_sink_msg {
struct new_datagram_msg {
// Handle to the endpoint used.
datagram_sink_handle handle;
dgram_scribe_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;
};
/// @relates datagram_sink_passivated_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, datagram_sink_passivated_msg& x) {
return f(meta::type_name("datagram_sink_passivated_msg"), x.handle);
}
/// Signalizes that a datagram source has entered passive mode.
struct datagram_source_passivated_msg {
datagram_source_handle handle;
};
/// @relates datagram_source_passivated_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, datagram_source_passivated_msg& x) {
return f(meta::type_name("datagram_source_passivated_msg"), x.handle);
}
/// Signalizes that a {@link broker} datagram endpoint was closed.
struct endpoint_closed_msg {
endpoint_handle handle;
};
/// @relates endpoint_closed_msg
template <class Inspector>
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
/// @relates new_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);
typename Inspector::result_type inspect(Inspector& f, new_datagram_msg& x) {
return f(meta::type_name("new_datagram_msg"), x.handle, x.written);
}
/// Signalizes that a datagram with a certain size has been sent.
struct datagram_sent_msg {
// Handle to the endpoint used.
endpoint_handle handle;
dgram_scribe_handle handle;
// number of bytes written.
uint64_t written;
};
......@@ -245,16 +194,28 @@ 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;
/// Signalizes that a datagram sink has entered passive mode.
struct dgram_scribe_passivated_msg {
dgram_scribe_handle handle;
};
/// @relates dgram_scribe_passivated_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, dgram_scribe_passivated_msg& x) {
return f(meta::type_name("dgram_scribe_passivated_msg"), x.handle);
}
/// Signalizes that a datagram source has entered passive mode.
struct dgram_doorman_passivated_msg {
dgram_doorman_handle handle;
};
/// @relates endpoint_passivated_msg
/// @relates dgram_doorman_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);
inspect(Inspector& f, dgram_doorman_passivated_msg& x) {
return f(meta::type_name("dgram_doorman_passivated_msg"), x.handle);
}
} // namespace io
......
......@@ -110,14 +110,14 @@ void abstract_broker::flush(connection_handle hdl) {
x->flush();
}
void abstract_broker::ack_writes(datagram_sink_handle hdl, bool enable) {
void abstract_broker::ack_writes(dgram_scribe_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(datagram_source_handle hdl,
void abstract_broker::configure_datagram_size(dgram_doorman_handle hdl,
size_t buf_size) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(buf_size));
auto x = by_id(hdl);
......@@ -125,17 +125,17 @@ void abstract_broker::configure_datagram_size(datagram_source_handle hdl,
x->configure_datagram_size(buf_size);
}
std::vector<char>& abstract_broker::wr_buf(datagram_sink_handle hdl) {
std::vector<char>& abstract_broker::wr_buf(dgram_scribe_handle hdl) {
auto x = by_id(hdl);
if (!x) {
CAF_LOG_ERROR("tried to access wr_buf() of an unknown"
"datagram_sink_handle");
"dgram_scribe_handle");
return dummy_wr_buf_;
}
return x->wr_buf();
}
void abstract_broker::write(datagram_sink_handle hdl, size_t bs,
void abstract_broker::write(dgram_scribe_handle hdl, size_t bs,
const void* buf) {
auto& out = wr_buf(hdl);
auto first = reinterpret_cast<const char*>(buf);
......@@ -143,37 +143,6 @@ 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());
......@@ -228,80 +197,51 @@ abstract_broker::add_tcp_doorman(network::native_socket fd) {
return backend().add_tcp_doorman(this, fd);
}
void abstract_broker::add_datagram_sink(const intrusive_ptr<datagram_sink>& ptr) {
datagram_sinks_.emplace(ptr->hdl(), ptr);
void abstract_broker::add_dgram_scribe(const intrusive_ptr<dgram_scribe>& ptr) {
dgram_scribes_.emplace(ptr->hdl(), ptr);
}
expected<datagram_sink_handle>
abstract_broker::add_datagram_sink(const std::string& hostname, uint16_t port) {
expected<dgram_scribe_handle>
abstract_broker::add_dgram_scribe(const std::string& hostname, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hostname) << CAF_ARG(port));
return backend().add_datagram_sink(this, hostname, port);
return backend().add_dgram_scribe(this, hostname, port);
}
expected<void> abstract_broker::assign_datagram_sink(datagram_sink_handle hdl) {
expected<void> abstract_broker::assign_dgram_scribe(dgram_scribe_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
return backend().assign_datagram_sink(this, hdl);
return backend().assign_dgram_scribe(this, hdl);
}
expected<datagram_sink_handle>
abstract_broker::add_datagram_sink(network::native_socket fd) {
expected<dgram_scribe_handle>
abstract_broker::add_dgram_scribe(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return backend().add_datagram_sink(this, fd);
return backend().add_dgram_scribe(this, fd);
}
void
abstract_broker::add_datagram_source(const intrusive_ptr<datagram_source>& ptr) {
datagram_sources_.emplace(ptr->hdl(), ptr);
abstract_broker::add_dgram_doorman(const intrusive_ptr<dgram_doorman>& ptr) {
dgram_doormans_.emplace(ptr->hdl(), ptr);
ptr->launch();
// TODO: some launching of things?
}
expected<std::pair<datagram_source_handle, uint16_t>>
abstract_broker::add_datagram_source(uint16_t port, const char* in,
expected<std::pair<dgram_doorman_handle, uint16_t>>
abstract_broker::add_dgram_doorman(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_datagram_source(this, port, in, reuse_addr);
return backend().add_dgram_doorman(this, port, in, reuse_addr);
}
expected<void>
abstract_broker::assign_datagram_source(datagram_source_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
return backend().assign_datagram_source(this, hdl);
}
expected<datagram_source_handle>
abstract_broker::add_datagram_source(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(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) {
abstract_broker::assign_dgram_doorman(dgram_doorman_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);
return backend().assign_dgram_doorman(this, hdl);
}
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) {
expected<dgram_doorman_handle>
abstract_broker::add_dgram_doorman(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return backend().add_endpoint(this, fd);
return backend().add_dgram_doorman(this, fd);
}
std::string abstract_broker::remote_addr(connection_handle hdl) {
......@@ -324,19 +264,19 @@ uint16_t abstract_broker::local_port(accept_handle hdl) {
return i != doormen_.end() ? i->second->port() : 0;
}
std::string abstract_broker::remote_addr(datagram_sink_handle hdl) {
auto i = datagram_sinks_.find(hdl);
return i != datagram_sinks_.end() ? i->second->addr() : std::string{};
std::string abstract_broker::remote_addr(dgram_scribe_handle hdl) {
auto i = dgram_scribes_.find(hdl);
return i != dgram_scribes_.end() ? i->second->addr() : std::string{};
}
uint16_t abstract_broker::remote_port(datagram_sink_handle hdl) {
auto i = datagram_sinks_.find(hdl);
return i != datagram_sinks_.end() ? i->second->port() : 0;
uint16_t abstract_broker::remote_port(dgram_scribe_handle hdl) {
auto i = dgram_scribes_.find(hdl);
return i != dgram_scribes_.end() ? i->second->port() : 0;
}
uint16_t abstract_broker::local_port(datagram_source_handle hdl) {
auto i = datagram_sources_.find(hdl);
return i != datagram_sources_.end() ? i->second->port() : 0;
uint16_t abstract_broker::local_port(dgram_doorman_handle hdl) {
auto i = dgram_doormans_.find(hdl);
return i != dgram_doormans_.end() ? i->second->port() : 0;
}
accept_handle abstract_broker::hdl_by_port(uint16_t port) {
......@@ -356,9 +296,13 @@ void abstract_broker::close_all() {
// stop_reading will remove the scribe from scribes_
scribes_.begin()->second->stop_reading();
}
while (!endpoints_.empty()) {
// stop_reading will remove the enpoint from endpoints_
endpoints_.begin()->second->stop_reading();
while (!dgram_doormans_.empty()) {
// stop_reading will remove the scribe from dgram_doormans_
dgram_doormans_.begin()->second->stop_reading();
}
while (!dgram_scribes_.empty()) {
// stop_reading will remove the scribe from dgram_scribes_
dgram_scribes_.begin()->second->stop_reading();
}
}
......
......@@ -465,21 +465,6 @@ void basp_broker_state::set_context(connection_handle hdl) {
this_context = &i->second;
}
void basp_broker_state::set_context(endpoint_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
auto i = udp_ctx.find(hdl);
if (i == udp_ctx.end()) {
CAF_LOG_INFO("create new BASP context:" << CAF_ARG(hdl));
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},
hdl, none, 0, 0, none}).first;
}
udp_context = &i->second;
}
/******************************************************************************
* basp_broker *
******************************************************************************/
......@@ -537,10 +522,11 @@ behavior basp_broker::make_behavior() {
}
},
// received from underlying broker implementation
[=](datagram_msg& msg) {
[=](new_datagram_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
CAF_LOG_DEBUG("Received new_datagram_msg: " << CAF_ARG(msg));
state.set_context(msg.handle);
static_cast<void>(msg);
// state.set_context(msg.handle);
// auto& ctx = *state.udp_context;
//auto next = state.instance.handle(context(), msg, ctx.hdr, ...);
// TODO: implement this
......@@ -645,14 +631,14 @@ behavior basp_broker::make_behavior() {
<< CAF_ARG(whom));
}
},
[=](publish_atom, datagram_source_handle hdl, uint16_t port,
[=](publish_atom, dgram_doorman_handle hdl, uint16_t port,
const strong_actor_ptr& whom, std::set<std::string>& sigs) {
CAF_LOG_TRACE(CAF_ARG(hdl.id()) << CAF_ARG(whom) << CAF_ARG(port));
if (hdl.invalid()) {
CAF_LOG_WARNING("invalid handle");
return;
}
auto res = assign_datagram_source(hdl);
auto res = assign_dgram_doorman(hdl);
if (res) {
if (whom)
system().registry().put(whom->id(), whom);
......@@ -680,13 +666,13 @@ behavior basp_broker::make_behavior() {
rp.deliver(std::move(res.error()));
}
},
[=](connect_atom, endpoint_handle hdl, uint16_t port) {
[=](connect_atom, dgram_scribe_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);
auto res = assign_dgram_scribe(hdl);
if (res) {
auto& ctx = state.udp_ctx[hdl];
ctx.sink = hdl;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/datagram_source.hpp"
#include "caf/logger.hpp"
namespace caf {
namespace io {
datagram_source::datagram_source(abstract_broker* parent,
datagram_source_handle hdl)
: datagram_source_base(parent, hdl) {
// nop
}
datagram_source::~datagram_source() {
CAF_LOG_TRACE("");
}
message datagram_source::detach_message() {
return make_message(datagram_source_closed_msg{hdl()});
}
bool datagram_source::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 datagram_source::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/datagram_source_manager.hpp"
namespace caf {
namespace io {
namespace network {
datagram_source_manager::datagram_source_manager(abstract_broker* ptr)
: manager(ptr) {
// nop
}
datagram_source_manager::~datagram_source_manager() {
// nop
}
} // namespace network
} // namespace io
} // namespace caf
......@@ -622,11 +622,6 @@ void default_multiplexer::del(operation op, native_socket fd,
new_event(del_flag, op, fd, ptr);
}
std::map<default_multiplexer::endpoint_addr, endpoint_handle>&
default_multiplexer::endpoints() {
return remote_endpoints_;
}
void default_multiplexer::wr_dispatch_request(resumable* ptr) {
intptr_t ptrval = reinterpret_cast<intptr_t>(ptr);
// on windows, we actually have sockets, otherwise we have file handles
......@@ -862,115 +857,48 @@ 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) {
dgram_scribe_handle
default_multiplexer::add_dgram_scribe(abstract_broker* self,
native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
CAF_ASSERT(fd != network::invalid_native_socket);
class impl : public endpoint {
class impl : public dgram_scribe {
public:
impl(abstract_broker* ptr, default_multiplexer& mx, native_socket sockfd)
: endpoint(ptr, network::endpoint_hdl_from_socket(sockfd)),
handler_(mx, sockfd) {
: dgram_scribe(ptr, network::dg_sink_hdl_from_socket(sockfd)),
communicator_(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();
communicator_.configure_datagram_size(buf_size);
}
void stop_reading() override {
CAF_LOG_TRACE("");
handler_.stop_reading();
detach(&handler_.backend(), false);
void ack_writes(bool enable) override {
communicator_.ack_writes(enable);
}
std::vector<char>& wr_buf() override {
return handler_.wr_buf();
return communicator_.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_handler 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) {
CAF_LOG_TRACE(CAF_ARG(fd));
CAF_ASSERT(fd != network::invalid_native_socket);
class impl : public datagram_sink {
public:
impl(abstract_broker* ptr, default_multiplexer& mx, native_socket sockfd)
: datagram_sink(ptr, network::dg_sink_hdl_from_socket(sockfd)),
sender_(mx, sockfd) {
// nop
return communicator_.rd_buf();
}
void stop_reading() override {
CAF_LOG_TRACE("");
sender_.stop_reading();
detach(&sender_.backend(), false);
}
void ack_writes(bool enable) override {
sender_.ack_writes(enable);
communicator_.stop_reading();
detach(&communicator_.backend(), false);
}
std::vector<char>& wr_buf() override {
return sender_.wr_buf();
void flush() override {
CAF_LOG_TRACE("");
communicator_.flush(this);
}
std::string addr() const override {
auto x = remote_addr_of_fd(sender_.fd());
auto x = remote_addr_of_fd(communicator_.fd());
if (!x)
return "";
return *x;
}
uint16_t port() const override {
auto x = remote_port_of_fd(sender_.fd());
auto x = remote_port_of_fd(communicator_.fd());
if (!x)
return 0;
return *x;
......@@ -979,77 +907,89 @@ default_multiplexer::add_datagram_sink(abstract_broker* self,
CAF_LOG_TRACE("");
CAF_ASSERT(!launched_);
launched_ = true;
sender_.start(this);
communicator_.start(this);
}
void add_to_loop() override {
sender_.activate(this);
communicator_.activate(this);
}
void remove_from_loop() override {
sender_.passivate();
communicator_.passivate();
}
private:
bool launched_;
network::datagram_sender sender_;
network::dgram_communicator communicator_;
};
auto ptr = make_counted<impl>(self, *this, fd);
self->add_datagram_sink(ptr);
self->add_dgram_scribe(ptr);
return ptr->hdl();
}
datagram_source_handle
default_multiplexer::add_datagram_source(abstract_broker* self,
dgram_doorman_handle
default_multiplexer::add_dgram_doorman(abstract_broker* self,
native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
CAF_ASSERT(fd != network::invalid_native_socket);
class impl : public datagram_source {
class impl : public dgram_doorman {
public:
impl(abstract_broker* ptr, default_multiplexer& mx, native_socket sockfd)
: datagram_source(ptr, network::dg_source_hdl_from_socket(sockfd)),
receiver_(mx, sockfd) {
: dgram_doorman(ptr, network::dg_source_hdl_from_socket(sockfd)),
acceptor_(mx, sockfd) {
// nop
}
void configure_datagram_size(size_t buf_size) override {
CAF_LOG_TRACE("");
receiver_.configure_datagram_size(buf_size);
acceptor_.configure_datagram_size(buf_size);
}
bool new_endpoint() override {
CAF_LOG_TRACE("");
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;
auto& dm = acceptor_.backend();
static_cast<void>(dm);
/*
auto hdl = dm.add_tcp_scribe(parent(),
std::move(acceptor_.accepted_socket()));
return doorman::new_connection(&dm, hdl);
*/
// TODO: Implement me!
return false;
}
void stop_reading() override {
CAF_LOG_TRACE("");
receiver_.stop_reading();
detach(&receiver_.backend(), false);
acceptor_.stop_reading();
detach(&acceptor_.backend(), false);
}
std::vector<char>& rd_buf() override {
return receiver_.rd_buf();
return acceptor_.rd_buf();
}
std::string addr() const override {
auto x = remote_addr_of_fd(receiver_.fd());
if (!x)
return "";
return *x;
return acceptor_.host();
}
uint16_t port() const override {
auto x = remote_port_of_fd(receiver_.fd());
if (!x)
return 0;
return *x;
return acceptor_.port();
}
void launch() override {
CAF_LOG_TRACE("");
CAF_ASSERT(!launched_);
launched_ = true;
receiver_.start(this);
acceptor_.start(this);
}
void add_to_loop() override {
receiver_.activate(this);
acceptor_.activate(this);
}
void remove_from_loop() override {
receiver_.passivate();
acceptor_.passivate();
}
private:
bool launched_;
network::datagram_receiver receiver_;
network::dgram_acceptor acceptor_;
};
auto ptr = make_counted<impl>(self, *this, fd);
self->add_datagram_source(ptr);
self->add_dgram_doorman(ptr);
return ptr->hdl();
}
......@@ -1104,111 +1044,59 @@ default_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port,
return std::make_pair(add_tcp_doorman(self, res->first), bound_port);
}
expected<datagram_sink_handle>
default_multiplexer::new_datagram_sink(const std::string& host, uint16_t port) {
auto fd = new_datagram_sink_impl(host, port);
expected<dgram_scribe_handle>
default_multiplexer::new_dgram_scribe(const std::string& host, uint16_t port) {
auto fd = new_dgram_scribe_impl(host, port);
if (!fd)
return std::move(fd.error());
return datagram_sink_handle::from_int(int64_from_native_socket(*fd));
return dgram_scribe_handle::from_int(int64_from_native_socket(*fd));
}
expected<void> default_multiplexer::assign_datagram_sink(abstract_broker* self,
datagram_sink_handle hdl) {
expected<void> default_multiplexer::assign_dgram_scribe(abstract_broker* self,
dgram_scribe_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(self->id()) << CAF_ARG(hdl));
add_datagram_sink(self, static_cast<native_socket>(hdl.id()));
add_dgram_scribe(self, static_cast<native_socket>(hdl.id()));
return unit;
}
expected<datagram_sink_handle>
default_multiplexer::add_datagram_sink(abstract_broker* self,
expected<dgram_scribe_handle>
default_multiplexer::add_dgram_scribe(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_datagram_sink_impl(host, port);
auto fd = new_dgram_scribe_impl(host, port);
if (!fd)
return std::move(fd.error());
return add_datagram_sink(self, *fd);
return add_dgram_scribe(self, *fd);
}
expected<std::pair<datagram_source_handle, uint16_t>>
default_multiplexer::new_datagram_source(uint16_t port, const char* in,
expected<std::pair<dgram_doorman_handle, uint16_t>>
default_multiplexer::new_dgram_doorman(uint16_t port, const char* in,
bool reuse_addr) {
auto res = new_datagram_source_impl(port, in, reuse_addr);
auto res = new_dgram_doorman_impl(port, in, reuse_addr);
if (!res)
return std::move(res.error());
return std::make_pair(datagram_source_handle::from_int(
return std::make_pair(dgram_doorman_handle::from_int(
int64_from_native_socket(res->first)),
res->second);
}
expected<void> default_multiplexer::assign_datagram_source(abstract_broker* self,
datagram_source_handle hdl) {
expected<void> default_multiplexer::assign_dgram_doorman(abstract_broker* self,
dgram_doorman_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(self->id()) << CAF_ARG(hdl));
add_datagram_source(self, static_cast<native_socket>(hdl.id()));
add_dgram_doorman(self, static_cast<native_socket>(hdl.id()));
return unit;
}
expected<std::pair<datagram_source_handle, uint16_t>>
default_multiplexer::add_datagram_source(abstract_broker* self, uint16_t port,
expected<std::pair<dgram_doorman_handle, uint16_t>>
default_multiplexer::add_dgram_doorman(abstract_broker* self, uint16_t port,
const char* in, bool reuse_addr) {
auto res = new_datagram_source_impl(port, in, reuse_addr);
auto res = new_dgram_doorman_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());
auto hdl = endpoint_handle::from_int(int64_from_native_socket(*fd));
remote_endpoints_[std::make_pair(host, port)] = hdl;
return hdl;
}
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());
auto hdl = add_endpoint(self, *fd);
remote_endpoints_[std::make_pair(host, port)] = hdl;
return hdl;
}
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);
return std::make_pair(add_dgram_doorman(self, res->first), bound_port);
}
/******************************************************************************
......@@ -1599,25 +1487,32 @@ void acceptor::removed_from_loop(operation op) {
mgr_.reset();
}
datagram_sender::datagram_sender(default_multiplexer& backend_ref,
dgram_communicator::dgram_communicator(default_multiplexer& backend_ref,
native_socket sockfd)
: event_handler(backend_ref, sockfd),
ack_writes_(false) {
// nop
dgram_size_(0),
ack_writes_(false),
writing_(false) {
// TODO: Set some reasonable default.
configure_datagram_size(1500);
}
void datagram_sender::ack_writes(bool x) {
void dgram_communicator::ack_writes(bool x) {
ack_writes_ = x;
}
void datagram_sender::write(const void* buf, size_t num_bytes) {
void dgram_communicator::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_sender::flush(const manager_ptr& mgr) {
void dgram_communicator::configure_datagram_size(size_t size) {
dgram_size_ = size;
}
void dgram_communicator::flush(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr);
CAF_LOG_TRACE(CAF_ARG(wr_offline_buf_.size()));
if (wr_offline_buf_.empty()) {
......@@ -1627,31 +1522,31 @@ void datagram_sender::flush(const manager_ptr& mgr) {
}
}
void datagram_sender::start(manager_type* mgr) {
void dgram_communicator::start(manager_type* mgr) {
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void datagram_sender::activate(manager_type* mgr) {
void dgram_communicator::activate(manager_type* mgr) {
if (!writer_) {
writer_.reset(mgr);
event_handler::activate();
}
}
void datagram_sender::stop_reading() {
void dgram_communicator::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
void datagram_sender::removed_from_loop(operation op) {
void dgram_communicator::removed_from_loop(operation op) {
CAF_LOG_TRACE(CAF_ARG(fd()) << CAF_ARG(op));
if (op == operation::write)
writer_.reset();
}
void datagram_sender::handle_event(operation op) {
void dgram_communicator::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case operation::read: {
......@@ -1687,7 +1582,7 @@ void datagram_sender::handle_event(operation op) {
}
}
void datagram_sender::prepare_next_write() {
void dgram_communicator::prepare_next_write() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.size()) << CAF_ARG(wr_offline_buf_.size()));
wr_buf_.clear();
if (wr_offline_buf_.empty()) {
......@@ -1697,55 +1592,84 @@ void datagram_sender::prepare_next_write() {
}
}
datagram_receiver::datagram_receiver(default_multiplexer& backend_ref,
void dgram_communicator::sender_from_sockaddr(sockaddr_storage& sa, size_t) {
char addr[INET6_ADDRSTRLEN];
switch(sa.ss_family) {
case AF_INET:
port_ = ntohs(reinterpret_cast<sockaddr_in*>(&sa)->sin_port);
inet_ntop(AF_INET,
&reinterpret_cast<sockaddr_in*>(&sa)->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*>(&sa)->sin6_port);
inet_ntop(AF_INET6,
&reinterpret_cast<sockaddr_in*>(&sa)->sin_addr,
addr, INET6_ADDRSTRLEN);
host_.insert(std::begin(host_), std::begin(addr),
std::begin(addr) + INET6_ADDRSTRLEN);
break;
default:
// nop
break;
}
}
dgram_acceptor::dgram_acceptor(default_multiplexer& backend_ref,
native_socket sockfd)
: event_handler(backend_ref, sockfd),
buf_size_(1500) {
// TODO: Set reasonable default for buffer size
dgram_size_(1500),
sockaddr_len_(0) {
// TODO: Set reasonable default for datagram_size
configure_datagram_size(1500);
// nop
}
void datagram_receiver::configure_datagram_size(size_t buf_size) {
buf_size_ = buf_size;
void dgram_acceptor::configure_datagram_size(size_t size) {
dgram_size_ = size;
}
void datagram_receiver::start(manager_type* mgr) {
void dgram_acceptor::start(manager_type* mgr) {
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void datagram_receiver::activate(manager_type* mgr) {
void dgram_acceptor::activate(manager_type* mgr) {
if (!reader_) {
reader_.reset(mgr);
event_handler::activate();
}
}
void datagram_receiver::stop_reading() {
void dgram_acceptor::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
void datagram_receiver::removed_from_loop(operation op) {
void dgram_acceptor::removed_from_loop(operation op) {
if (op == operation::read)
reader_.reset();
}
void datagram_receiver::handle_event(operation op) {
void dgram_acceptor::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)) {
sockaddr_, sockaddr_len_)) {
reader_->io_failure(&backend(), operation::read);
passivate();
return;
}
sender_from_sockaddr(sockaddr_, sockaddr_len_);
if (rb == 0)
return;
/*
auto res = reader_->consume(&backend(), rd_buf_.data(), rb);
packet_size_ = rb;
prepare_next_read();
......@@ -1753,6 +1677,7 @@ void datagram_receiver::handle_event(operation op) {
passivate();
return;
}
*/
break;
}
case operation::write: {
......@@ -1768,223 +1693,32 @@ void datagram_receiver::handle_event(operation op) {
}
}
std::pair<std::string,uint16_t> datagram_receiver::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:
// nop
break;
}
return std::make_pair(std::move(host), port);
}
void datagram_receiver::prepare_next_read() {
rd_buf_.resize(buf_size_);
}
datagram_handler::datagram_handler(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_handler::ack_writes(bool x) {
ack_writes_ = x;
}
void datagram_handler::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_handler::configure_datagram_size(size_t buf_size) {
buf_size_ = buf_size;
}
std::pair<std::string,uint16_t> datagram_handler::get_sender() {
void dgram_acceptor::sender_from_sockaddr(sockaddr_storage& sa, size_t) {
char addr[INET6_ADDRSTRLEN];
std::string host;
uint16_t port = 0;
switch(last_sender.ss_family) {
switch(sa.ss_family) {
case AF_INET:
port = ntohs(reinterpret_cast<sockaddr_in*>(&last_sender)->sin_port);
port_ = ntohs(reinterpret_cast<sockaddr_in*>(&sa)->sin_port);
inet_ntop(AF_INET,
&reinterpret_cast<sockaddr_in*>(&last_sender)->sin_addr,
&reinterpret_cast<sockaddr_in*>(&sa)->sin_addr,
addr, INET_ADDRSTRLEN);
host.insert(std::begin(host), std::begin(addr),
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);
port_ = ntohs(reinterpret_cast<sockaddr_in6*>(&sa)->sin6_port);
inet_ntop(AF_INET6,
&reinterpret_cast<sockaddr_in*>(&last_sender)->sin_addr,
&reinterpret_cast<sockaddr_in*>(&sa)->sin_addr,
addr, INET6_ADDRSTRLEN);
host.insert(std::begin(host), std::begin(addr),
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_handler::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_handler::start(manager_type* mgr) {
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void datagram_handler::activate(manager_type* mgr) {
if (!reader_) {
reader_.reset(mgr);
event_handler::activate();
prepare_next_read();
}
}
void datagram_handler::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
void datagram_handler::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_handler::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;
// Find responsible (remote) endpoint to deliver datagram.
auto sender = get_sender();
bool consumed = false;
if (reader_->addr() == sender.first &&
reader_->port() == sender.second) {
// Our assigned endpoint is respoinsible
consumed = reader_->consume(&backend(), rd_buf_.data(), rb);
} else {
// Search for the responsible endpoint or create a new one.
auto& endpoints = backend().endpoints();
auto endpoint = endpoints.find(sender);
if (endpoint == endpoints.end()) {
auto new_endpoint = backend().add_remote_endpoint(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;
// TODO: requires the multiplexer to be single threaded!
reader_->consume_as(hdl, &backend(), rd_buf_.data(), rb);
}
packet_size_ = rb;
prepare_next_read();
if (!consumed) {
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
// Could the handler propagte the knowledge if it know how
// much it can actually write?
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_handler::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_handler::prepare_next_write() {
void dgram_acceptor::prepare_next_read() {
rd_buf_.resize(buf_size_);
}
......@@ -2187,7 +1921,7 @@ new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr) {
return std::make_pair(sguard.release(), *p);
}
expected<native_socket> new_datagram_sink_impl(const std::string& host,
expected<native_socket> new_dgram_scribe_impl(const std::string& host,
uint16_t port,
optional<protocol> preferred) {
CAF_LOG_TRACE(CAF_ARG(host) << CAF_ARG(port) << CAF_ARG(preferred));
......@@ -2210,7 +1944,7 @@ expected<native_socket> new_datagram_sink_impl(const std::string& host,
}
sguard.close();
// IPv4 fallback
return new_datagram_sink_impl(host, port, ipv4);
return new_dgram_scribe_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));
......@@ -2221,77 +1955,8 @@ 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"));
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::pair<native_socket, uint16_t>>
new_local_endpoint_impl(uint16_t port, const char* addr, bool reuse_addr) {
new_dgram_doorman_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) {
......
......@@ -17,18 +17,18 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/network/datagram_sink_manager.hpp"
#include "caf/io/network/dgram_acceptor_manager.hpp"
namespace caf {
namespace io {
namespace network {
datagram_sink_manager::datagram_sink_manager(abstract_broker* ptr)
dgram_acceptor_manager::dgram_acceptor_manager(abstract_broker* ptr)
: manager(ptr) {
// nop
}
datagram_sink_manager::~datagram_sink_manager() {
dgram_acceptor_manager::~dgram_acceptor_manager() {
// nop
}
......
......@@ -17,18 +17,18 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/network/endpoint_manager.hpp"
#include "caf/io/network/dgram_communicator_manager.hpp"
namespace caf {
namespace io {
namespace network {
endpoint_manager::endpoint_manager(abstract_broker* ptr)
dgram_communicator_manager::dgram_communicator_manager(abstract_broker* ptr)
: manager(ptr) {
// nop
}
endpoint_manager::~endpoint_manager() {
dgram_communicator_manager::~dgram_communicator_manager() {
// nop
}
......
......@@ -17,46 +17,39 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/datagram_sink.hpp"
#include "caf/io/dgram_doorman.hpp"
#include "caf/logger.hpp"
namespace caf {
namespace io {
datagram_sink::datagram_sink(abstract_broker* parent, datagram_sink_handle hdl)
: datagram_sink_base(parent, hdl) {
dgram_doorman::dgram_doorman(abstract_broker* parent,
dgram_doorman_handle hdl)
: dgram_doorman_base(parent, hdl) {
// nop
}
datagram_sink::~datagram_sink() {
dgram_doorman::~dgram_doorman() {
CAF_LOG_TRACE("");
}
message datagram_sink::detach_message() {
return make_message(datagram_sink_closed_msg{hdl()});
message dgram_doorman::detach_message() {
return make_message(dgram_doorman_closed_msg{hdl()});
}
void datagram_sink::datagram_sent(execution_unit*, size_t) {
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 datagram_sink::io_failure(execution_unit* ctx, network::operation op) {
void dgram_doorman::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);
}
bool dgram_doorman::new_endpoint(execution_unit* ctx, const void* buf,
size_t besize) {
// TODO: implement me!
return false;
}
} // namespace io
} // namespace caf
......@@ -17,28 +17,28 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/endpoint.hpp"
#include "caf/io/dgram_scribe.hpp"
#include "caf/logger.hpp"
namespace caf {
namespace io {
endpoint::endpoint(abstract_broker* parent, endpoint_handle hdl)
: endpoint_base(parent, hdl) {
dgram_scribe::dgram_scribe(abstract_broker* parent, dgram_scribe_handle hdl)
: dgram_scribe_base(parent, hdl) {
// nop
}
endpoint::~endpoint() {
dgram_scribe::~dgram_scribe() {
CAF_LOG_TRACE("");
}
message endpoint::detach_message() {
return make_message(endpoint_closed_msg{hdl()});
message dgram_scribe::detach_message() {
return make_message(dgram_scribe_closed_msg{hdl()});
}
bool endpoint::consume(execution_unit* ctx, const void*, size_t num_bytes) {
bool dgram_scribe::consume(execution_unit* ctx, const void*,
size_t num_bytes) {
CAF_ASSERT(ctx != nullptr);
CAF_LOG_TRACE(CAF_ARG(num_bytes));
if (detached())
......@@ -63,7 +63,7 @@ bool endpoint::consume(execution_unit* ctx, const void*, size_t num_bytes) {
return result;
}
void endpoint::datagram_sent(execution_unit* ctx, size_t written) {
void dgram_scribe::datagram_sent(execution_unit* ctx, size_t written) {
CAF_LOG_TRACE(CAF_ARG(written));
if (detached())
return;
......@@ -75,7 +75,7 @@ void endpoint::datagram_sent(execution_unit* ctx, size_t written) {
invoke_mailbox_element_impl(ctx, tmp);
}
void endpoint::io_failure(execution_unit* ctx, network::operation op) {
void dgram_scribe::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);
......
......@@ -23,9 +23,8 @@
#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"
#include "caf/io/dgram_scribe.hpp"
#include "caf/io/dgram_doorman.hpp"
namespace caf {
namespace io {
......@@ -215,142 +214,31 @@ 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,
expected<dgram_scribe_handle>
test_multiplexer::new_dgram_scribe(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()) {
dgram_scribe_handle result;
auto i = dgram_scribes_.find(std::make_pair(host, port_hint));
if (i != dgram_scribes_.end()) {
result = i->second;
remote_endpoints_.erase(i);
dgram_scribes_.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 {
expected<void> test_multiplexer::assign_dgram_scribe(abstract_broker* ptr,
dgram_scribe_handle hdl) {
class impl : public dgram_scribe {
public:
impl(abstract_broker* self, endpoint_handle eh, test_multiplexer* mpx)
: endpoint(self, eh),
impl(abstract_broker* self, dgram_scribe_handle dsh, test_multiplexer* mpx)
: dgram_scribe(self, dsh),
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) {
guard_type guard{mx_};
datagram_sink_handle result;
auto i = datagram_sinks_.find(std::make_pair(host, port_hint));
if (i != datagram_sinks_.end()) {
result = i->second;
datagram_sinks_.erase(i);
}
return result;
}
expected<void> test_multiplexer::assign_datagram_sink(abstract_broker* ptr,
datagram_sink_handle hdl) {
class impl : public datagram_sink {
public:
impl(abstract_broker* self, datagram_sink_handle dsh, test_multiplexer* mpx)
: datagram_sink(self, dsh),
mpx_(mpx) {
// nop
}
void stop_reading() override {
mpx_->stopped_reading(hdl()) = true;
detach(mpx_, false);
......@@ -380,46 +268,46 @@ expected<void> test_multiplexer::assign_datagram_sink(abstract_broker* ptr,
test_multiplexer* mpx_;
};
auto dsptr = make_counted<impl>(ptr, hdl, this);
ptr->add_datagram_sink(dsptr);
ptr->add_dgram_scribe(dsptr);
return unit;
}
datagram_sink_handle test_multiplexer::add_datagram_sink(abstract_broker*,
dgram_scribe_handle test_multiplexer::add_dgram_scribe(abstract_broker*,
native_socket) {
std::cerr << "test_multiplexer::add_datagram_sink called with native socket"
std::cerr << "test_multiplexer::add_dgram_scribe called with native socket"
<< std::endl;
abort();
}
expected<datagram_sink_handle>
test_multiplexer::add_datagram_sink(abstract_broker* ptr,
expected<dgram_scribe_handle>
test_multiplexer::add_dgram_scribe(abstract_broker* ptr,
const std::string& host, uint16_t prt) {
auto result = new_datagram_sink(host, prt);
auto result = new_dgram_scribe(host, prt);
if (!result)
return std::move(result.error());
// port(*result) = prt; // TODO: remove this?
assign_datagram_sink(ptr, *result);
assign_dgram_scribe(ptr, *result);
return result;
}
expected<std::pair<datagram_source_handle, uint16_t>>
test_multiplexer::new_datagram_source(uint16_t desired_port, const char*, bool) {
expected<std::pair<dgram_doorman_handle, uint16_t>>
test_multiplexer::new_dgram_doorman(uint16_t desired_port, const char*, bool) {
guard_type guard{mx_};
datagram_source_handle result;
auto i = datagram_sources_.find(desired_port);
if (i != datagram_sources_.end()) {
dgram_doorman_handle result;
auto i = dgram_doormans_.find(desired_port);
if (i != dgram_doormans_.end()) {
result = i->second;
datagram_sources_.erase(i);
dgram_doormans_.erase(i);
}
return std::make_pair(result, desired_port);
}
expected<void> test_multiplexer::assign_datagram_source(abstract_broker* ptr,
datagram_source_handle hdl) {
class impl : public datagram_source {
expected<void> test_multiplexer::assign_dgram_doorman(abstract_broker* ptr,
dgram_doorman_handle hdl) {
class impl : public dgram_doorman {
public:
impl(abstract_broker* self, datagram_source_handle dsh, test_multiplexer* mpx)
: datagram_source(self, dsh),
impl(abstract_broker* self, dgram_doorman_handle dsh, test_multiplexer* mpx)
: dgram_doorman(self, dsh),
mpx_(mpx) {
// nop
}
......@@ -452,25 +340,25 @@ expected<void> test_multiplexer::assign_datagram_source(abstract_broker* ptr,
test_multiplexer* mpx_;
};
auto dsptr = make_counted<impl>(ptr, hdl, this);
ptr->add_datagram_source(dsptr);
ptr->add_dgram_doorman(dsptr);
return unit;
}
datagram_source_handle test_multiplexer::add_datagram_source(abstract_broker*,
dgram_doorman_handle test_multiplexer::add_dgram_doorman(abstract_broker*,
native_socket) {
std::cerr << "test_multiplexer::add_datagram_source called with native socket"
std::cerr << "test_multiplexer::add_dgram_doorman called with native socket"
<< std::endl;
abort();
}
expected<std::pair<datagram_source_handle, uint16_t>>
test_multiplexer::add_datagram_source(abstract_broker* ptr, uint16_t prt,
expected<std::pair<dgram_doorman_handle, uint16_t>>
test_multiplexer::add_dgram_doorman(abstract_broker* ptr, uint16_t prt,
const char* in, bool reuse_addr) {
auto result = new_datagram_source(prt, in, reuse_addr);
auto result = new_dgram_doorman(prt, in, reuse_addr);
if (!result)
return std::move(result.error());
port(result->first) = prt;
assign_datagram_source(ptr, result->first);
assign_dgram_doorman(ptr, result->first);
return result;
}
......@@ -495,17 +383,17 @@ void test_multiplexer::provide_acceptor(uint16_t desired_port,
doorman_data_[hdl].port = desired_port;
}
void test_multiplexer::provide_datagram_sink(std::string host,
void test_multiplexer::provide_dgram_scribe(std::string host,
uint16_t desired_port,
datagram_sink_handle hdl) {
datagram_sinks_.emplace(std::make_pair(std::move(host), desired_port), hdl);
datagram_sink_data_[hdl].port = desired_port;
dgram_scribe_handle hdl) {
dgram_scribes_.emplace(std::make_pair(std::move(host), desired_port), hdl);
dgram_scribe_data_[hdl].port = desired_port;
}
void test_multiplexer::provide_datagram_source(uint16_t desired_port,
datagram_source_handle hdl) {
datagram_sources_.emplace(std::make_pair(desired_port, hdl));
datagram_source_data_[hdl].port = desired_port;
void test_multiplexer::provide_dgram_doorman(uint16_t desired_port,
dgram_doorman_handle hdl) {
dgram_doormans_.emplace(std::make_pair(desired_port, hdl));
dgram_doorman_data_[hdl].port = desired_port;
}
/// The external input buffer should be filled by
......@@ -550,63 +438,59 @@ uint16_t& test_multiplexer::port(accept_handle hdl) {
}
test_multiplexer::buffer_type&
test_multiplexer::output_buffer(datagram_sink_handle hdl) {
return datagram_sink_data_[hdl].wr_buf;
test_multiplexer::output_buffer(dgram_scribe_handle hdl) {
return dgram_scribe_data_[hdl].wr_buf;
}
bool& test_multiplexer::stopped_reading(datagram_sink_handle hdl) {
return datagram_sink_data_[hdl].stopped_reading;
bool& test_multiplexer::stopped_reading(dgram_scribe_handle hdl) {
return dgram_scribe_data_[hdl].stopped_reading;
}
bool& test_multiplexer::ack_writes(datagram_sink_handle hdl) {
return datagram_sink_data_[hdl].ack_writes;
bool& test_multiplexer::ack_writes(dgram_scribe_handle hdl) {
return dgram_scribe_data_[hdl].ack_writes;
}
bool& test_multiplexer::passive_mode(datagram_sink_handle hdl) {
return datagram_sink_data_[hdl].passive_mode;
bool& test_multiplexer::passive_mode(dgram_scribe_handle hdl) {
return dgram_scribe_data_[hdl].passive_mode;
}
intrusive_ptr<datagram_sink>&
test_multiplexer::impl_ptr(datagram_sink_handle hdl) {
return datagram_sink_data_[hdl].ptr;
intrusive_ptr<dgram_scribe>&
test_multiplexer::impl_ptr(dgram_scribe_handle hdl) {
return dgram_scribe_data_[hdl].ptr;
}
uint16_t& test_multiplexer::port(datagram_sink_handle hdl) {
return datagram_sink_data_[hdl].port;
uint16_t& test_multiplexer::port(dgram_scribe_handle hdl) {
return dgram_scribe_data_[hdl].port;
}
test_multiplexer::buffer_type&
test_multiplexer::input_buffer(datagram_source_handle hdl) {
return datagram_source_data_[hdl].rd_buf;
test_multiplexer::input_buffer(dgram_doorman_handle hdl) {
return dgram_doorman_data_[hdl].rd_buf;
}
bool& test_multiplexer::stopped_reading(datagram_source_handle hdl) {
return datagram_source_data_[hdl].stopped_reading;
bool& test_multiplexer::stopped_reading(dgram_doorman_handle hdl) {
return dgram_doorman_data_[hdl].stopped_reading;
}
bool& test_multiplexer::passive_mode(datagram_source_handle hdl) {
return datagram_source_data_[hdl].passive_mode;
bool& test_multiplexer::passive_mode(dgram_doorman_handle hdl) {
return dgram_doorman_data_[hdl].passive_mode;
}
intrusive_ptr<datagram_source>&
test_multiplexer::impl_ptr(datagram_source_handle hdl) {
return datagram_source_data_[hdl].ptr;
intrusive_ptr<dgram_doorman>&
test_multiplexer::impl_ptr(dgram_doorman_handle hdl) {
return dgram_doorman_data_[hdl].ptr;
}
uint16_t& test_multiplexer::port(datagram_source_handle hdl) {
return datagram_source_data_[hdl].port;
uint16_t& test_multiplexer::port(dgram_doorman_handle hdl) {
return dgram_doorman_data_[hdl].port;
}
size_t& test_multiplexer::buffer_size(datagram_source_handle hdl) {
return datagram_source_data_[hdl].buffer_size;
size_t& test_multiplexer::buffer_size(dgram_doorman_handle hdl) {
return dgram_doorman_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;
size_t& test_multiplexer::buffer_size(dgram_scribe_handle hdl) {
return dgram_scribe_data_[hdl].buffer_size;
}
bool& test_multiplexer::stopped_reading(accept_handle hdl) {
......@@ -621,36 +505,6 @@ 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);
......
......@@ -72,7 +72,7 @@ behavior make_pong_behavior() {
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(datagrams, fixture)
/*
CAF_TEST(test_remote_endpoint) {
auto& mp = client_side_mm.backend();
auto hdl = client_side_mm.named_broker<basp_broker>(atom("BASP"));
......@@ -134,5 +134,5 @@ CAF_TEST(test_datagram_remote_actor) {
anon_send_exit(pong, exit_reason::user_shutdown);
}
*/
CAF_TEST_FIXTURE_SCOPE_END()
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