Commit 3a6b96a3 authored by Joseph Noir's avatar Joseph Noir

Begin generalization of UDP impl., breaks things

parent 3c85ec40
...@@ -24,9 +24,7 @@ set (LIBCAF_IO_SRCS ...@@ -24,9 +24,7 @@ set (LIBCAF_IO_SRCS
src/multiplexer.cpp src/multiplexer.cpp
src/uri.cpp src/uri.cpp
src/dgram_scribe.cpp src/dgram_scribe.cpp
src/dgram_doorman.cpp
src/dgram_stream_manager.cpp src/dgram_stream_manager.cpp
src/dgram_acceptor_manager.cpp
src/visitors.cpp src/visitors.cpp
# BASP files # BASP files
src/header.cpp src/header.cpp
......
...@@ -33,13 +33,11 @@ ...@@ -33,13 +33,11 @@
#include "caf/io/system_messages.hpp" #include "caf/io/system_messages.hpp"
#include "caf/io/connection_handle.hpp" #include "caf/io/connection_handle.hpp"
#include "caf/io/dgram_scribe_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/native_socket.hpp"
#include "caf/io/network/stream_manager.hpp" #include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp" #include "caf/io/network/acceptor_manager.hpp"
#include "caf/io/network/dgram_stream_manager.hpp" #include "caf/io/network/dgram_stream_manager.hpp"
#include "caf/io/network/dgram_acceptor_manager.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -87,7 +85,6 @@ public: ...@@ -87,7 +85,6 @@ public:
friend class scribe; friend class scribe;
friend class doorman; friend class doorman;
friend class dgram_scribe; friend class dgram_scribe;
friend class dgram_doorman;
// -- overridden modifiers of abstract_actor --------------------------------- // -- overridden modifiers of abstract_actor ---------------------------------
...@@ -162,11 +159,6 @@ public: ...@@ -162,11 +159,6 @@ public:
/// Enables or disables write notifications for given datagram socket. /// Enables or disables write notifications for given datagram socket.
void ack_writes(dgram_scribe_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(dgram_doorman_handle hdl, size_t buf_size);
/// Modifies the buffer for received datagrams. /// Modifies the buffer for received datagrams.
/// @param hdl Identifies the affected socket. /// @param hdl Identifies the affected socket.
/// @param buf_size Size of the receiver buffer for the next datagram. /// @param buf_size Size of the receiver buffer for the next datagram.
...@@ -233,34 +225,15 @@ public: ...@@ -233,34 +225,15 @@ public:
/// from the `multiplexer` to this broker. /// from the `multiplexer` to this broker.
expected<void> assign_dgram_scribe(dgram_scribe_handle hdl); expected<void> assign_dgram_scribe(dgram_scribe_handle hdl);
/// Creates and assigns a new `dgram_scribe` from given native socked `fd`.
expected<dgram_scribe_handle> add_dgram_scribe(network::native_socket fd);
/// Assigns a detached `dgram_scribe` instance identified by `hdl` /// Assigns a detached `dgram_scribe` instance identified by `hdl`
/// from the `multiplexer` to this broker and assigns it to handle the /// from the `multiplexer` to this broker and assigns it to handle the
/// remote endpoint `host` and `port`. /// remote endpoint `host` and `port`.
expected<void> assign_dgram_scribe(dgram_scribe_handle hdl, expected<void> assign_dgram_scribe(dgram_scribe_handle hdl,
const std::string& host, uint16_t port); const std::string& host, uint16_t port);
/// 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 `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 `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 `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 `dgram_doorman` instance identified by `hdl`
/// from the `multiplexer` to this broker.
expected<void> assign_dgram_doorman(dgram_doorman_handle hdl);
/// 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` /// Returns the remote address associated to `hdl`
/// or empty string if `hdl` is invalid. /// or empty string if `hdl` is invalid.
std::string remote_addr(connection_handle hdl); std::string remote_addr(connection_handle hdl);
...@@ -291,9 +264,6 @@ public: ...@@ -291,9 +264,6 @@ public:
/// or `0` if `hdl` is invalid. /// or `0` if `hdl` is invalid.
uint16_t local_port(dgram_scribe_handle hdl); uint16_t local_port(dgram_scribe_handle hdl);
/// Returns the local port associated to `hdl` or `0` if `hdl` is invalid.
uint16_t local_port(dgram_doorman_handle hdl);
/// Closes all connections and acceptors. /// Closes all connections and acceptors.
void close_all(); void close_all();
...@@ -355,9 +325,6 @@ protected: ...@@ -355,9 +325,6 @@ protected:
using dgram_scribe_map = std::unordered_map<dgram_scribe_handle, using dgram_scribe_map = std::unordered_map<dgram_scribe_handle,
intrusive_ptr<dgram_scribe>>; intrusive_ptr<dgram_scribe>>;
using dgram_doorman_map = std::unordered_map<dgram_doorman_handle,
intrusive_ptr<dgram_doorman>>;
/// @cond PRIVATE /// @cond PRIVATE
// meta programming utility // meta programming utility
...@@ -375,11 +342,6 @@ protected: ...@@ -375,11 +342,6 @@ protected:
return dgram_scribes_; return dgram_scribes_;
} }
// meta programming utility
inline dgram_doorman_map& get_map(dgram_doorman_handle) {
return dgram_doormans_;
}
// meta programming utility (not implemented) // meta programming utility (not implemented)
static intrusive_ptr<doorman> ptr_of(accept_handle); static intrusive_ptr<doorman> ptr_of(accept_handle);
...@@ -389,15 +351,12 @@ protected: ...@@ -389,15 +351,12 @@ protected:
// meta programming utility (not implemented) // meta programming utility (not implemented)
static intrusive_ptr<dgram_scribe> ptr_of(dgram_scribe_handle); static intrusive_ptr<dgram_scribe> ptr_of(dgram_scribe_handle);
// meta programming utility (not implemented)
static intrusive_ptr<dgram_doorman> ptr_of(dgram_doorman_handle);
/// @endcond /// @endcond
/// Returns the `multiplexer` running this broker. /// Returns the `multiplexer` running this broker.
network::multiplexer& backend(); network::multiplexer& backend();
/// Returns a `scribe`, `doorman`, `dgram_scribe` or `dgram_doorman` /// Returns a `scribe`, `doorman` or `dgram_scribe`
/// identified by `hdl`. /// identified by `hdl`.
template <class Handle> template <class Handle>
auto by_id(Handle hdl) -> optional<decltype(*ptr_of(hdl))> { auto by_id(Handle hdl) -> optional<decltype(*ptr_of(hdl))> {
...@@ -408,8 +367,8 @@ protected: ...@@ -408,8 +367,8 @@ protected:
return *(i->second); return *(i->second);
} }
/// Returns an intrusive pointer to a `scribe`, `doorman`, `dgram_scribe` /// Returns an intrusive pointer to a `scribe`, `doorman` or `dgram_scribe`
/// or `dgram_doorman` identified by `hdl` and remove it from this broker. /// identified by `hdl` and remove it from this broker.
template <class Handle> template <class Handle>
auto take(Handle hdl) -> decltype(ptr_of(hdl)) { auto take(Handle hdl) -> decltype(ptr_of(hdl)) {
using std::swap; using std::swap;
...@@ -427,7 +386,6 @@ private: ...@@ -427,7 +386,6 @@ private:
scribe_map scribes_; scribe_map scribes_;
doorman_map doormen_; doorman_map doormen_;
dgram_scribe_map dgram_scribes_; dgram_scribe_map dgram_scribes_;
dgram_doorman_map dgram_doormans_;
detail::intrusive_partitioned_list<mailbox_element, detail::disposer> cache_; detail::intrusive_partitioned_list<mailbox_element, detail::disposer> cache_;
std::vector<char> dummy_wr_buf_; std::vector<char> dummy_wr_buf_;
}; };
......
...@@ -30,7 +30,6 @@ ...@@ -30,7 +30,6 @@
#include "caf/io/scribe.hpp" #include "caf/io/scribe.hpp"
#include "caf/io/doorman.hpp" #include "caf/io/doorman.hpp"
#include "caf/io/dgram_doorman.hpp"
#include "caf/io/abstract_broker.hpp" #include "caf/io/abstract_broker.hpp"
#include "caf/io/dgram_scribe.hpp" #include "caf/io/dgram_scribe.hpp"
......
...@@ -29,8 +29,7 @@ ...@@ -29,8 +29,7 @@
namespace caf { namespace caf {
namespace io { namespace io {
/// Base class for `scribe` and `doorman` as well as `dgram_doorman` /// Base class for `scribe`, `doorman` and `dgram_scribe`.
/// and `dgram_scribe`.
/// @ingroup Broker /// @ingroup Broker
template <class Base, class Handle, class SysMsgType> template <class Base, class Handle, class SysMsgType>
class broker_servant : public Base { class broker_servant : public Base {
...@@ -105,11 +104,7 @@ protected: ...@@ -105,11 +104,7 @@ protected:
typename std::conditional< typename std::conditional<
std::is_same<Handle, accept_handle>::value, std::is_same<Handle, accept_handle>::value,
acceptor_passivated_msg, acceptor_passivated_msg,
typename std::conditional< dgram_scribe_passivated_msg
std::is_same<Handle, dgram_scribe_handle>::value,
dgram_scribe_passivated_msg,
dgram_doorman_passivated_msg
>::type
>::type >::type
>::type; >::type;
using tmp_t = mailbox_element_vals<passiv_t>; using tmp_t = mailbox_element_vals<passiv_t>;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_DGRAM_DOORMAN_HPP
#define CAF_IO_DGRAM_DOORMAN_HPP
#include <vector>
#include "caf/message.hpp"
#include "caf/io/broker_servant.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/dgram_doorman_handle.hpp"
#include "caf/io/network/dgram_acceptor_manager.hpp"
namespace caf {
namespace io {
using dgram_doorman_base = broker_servant<network::dgram_acceptor_manager,
dgram_doorman_handle,
dgram_dummy_msg>;
/// Manages reading from a datagram source
/// @ingroup Broker
class dgram_doorman : public dgram_doorman_base {
public:
dgram_doorman(abstract_broker* parent, dgram_doorman_handle hdl);
~dgram_doorman();
/// Configure buffer size for next accepted datagram.
virtual void configure_datagram_size(size_t buf_size) = 0;
/// Returns the current input buffer.
virtual std::vector<char>& rd_buf() = 0;
/// Returns the local port
virtual uint16_t local_port() const = 0;
void io_failure(execution_unit* ctx, network::operation op) override;
using dgram_doorman_base::new_endpoint;
bool new_endpoint(execution_unit* ctx, dgram_scribe_handle endpoint,
const void* buf, size_t num_bytes);
// needs to be launched explicitly
virtual void launch() = 0;
protected:
message detach_message() override;
};
} // namespace io
} // namespace caf
#endif // CAF_IO_DGRAM_DOORMAN_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_DGRAM_DOORMAN_HANDLE_HPP
#define CAF_IO_DGRAM_DOORMAN_HANDLE_HPP
#include <functional>
#include "caf/error.hpp"
#include "caf/io/handle.hpp"
#include "caf/meta/type_name.hpp"
namespace caf {
namespace io {
struct invalid_dgram_doorman_handle_t {
constexpr invalid_dgram_doorman_handle_t() {
// nop
}
};
constexpr invalid_dgram_doorman_handle_t invalid_dgram_doorman_handle
= invalid_dgram_doorman_handle_t{};
/// Generic type for identifying a datagram source.
class dgram_doorman_handle : public handle<dgram_doorman_handle,
invalid_dgram_doorman_handle_t> {
public:
friend class handle<dgram_doorman_handle, invalid_dgram_doorman_handle_t>;
using super = handle<dgram_doorman_handle, invalid_dgram_doorman_handle_t>;
constexpr dgram_doorman_handle() {
// nop
}
constexpr dgram_doorman_handle(const invalid_dgram_doorman_handle_t&) {
// nop
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
dgram_doorman_handle& x) {
return f(meta::type_name("dgram_doorman_handle"), x.id_);
}
private:
inline dgram_doorman_handle(int64_t handle_id) : super(handle_id) {
// nop
}
};
} // namespace io
} // namespace caf
namespace std {
template<>
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_DGRAM_DOORMAN_HANDLE_HPP
...@@ -43,17 +43,17 @@ constexpr invalid_dgram_scribe_handle_t invalid_dgram_scribe_handle ...@@ -43,17 +43,17 @@ constexpr invalid_dgram_scribe_handle_t invalid_dgram_scribe_handle
/// Generic type for identifying datagram sink. /// Generic type for identifying datagram sink.
class dgram_scribe_handle : public handle<dgram_scribe_handle, class dgram_scribe_handle : public handle<dgram_scribe_handle,
invalid_dgram_scribe_handle_t> { invalid_dgram_scribe_handle_t> {
public: public:
friend class handle<dgram_scribe_handle, invalid_dgram_scribe_handle_t>; friend class handle<dgram_scribe_handle, invalid_dgram_scribe_handle_t>;
using super = handle<dgram_scribe_handle, invalid_dgram_scribe_handle_t>; using super = handle<dgram_scribe_handle, invalid_dgram_scribe_handle_t>;
dgram_scribe_handle() : port_{0} { dgram_scribe_handle() {
// nop // nop
} }
dgram_scribe_handle(const invalid_dgram_scribe_handle_t&) : port_{0} { dgram_scribe_handle(const invalid_dgram_scribe_handle_t&) {
// nop // nop
} }
...@@ -63,28 +63,10 @@ public: ...@@ -63,28 +63,10 @@ public:
return f(meta::type_name("dgram_scribe_handle"), x.id_); return f(meta::type_name("dgram_scribe_handle"), x.id_);
} }
const std::string& host() const {
return host_;
}
void set_host(std::string host) {
host_ = move(host);
}
uint16_t port() const {
return port_;
}
void set_port(uint16_t port) {
port_ = port;
}
private: private:
inline dgram_scribe_handle(int64_t handle_id) : super(handle_id) { inline dgram_scribe_handle(int64_t handle_id) : super(handle_id) {
// nop // nop
} }
std::string host_;
uint16_t port_;
}; };
} // namespace io } // namespace io
......
...@@ -29,7 +29,6 @@ class doorman; ...@@ -29,7 +29,6 @@ class doorman;
class middleman; class middleman;
class basp_broker; class basp_broker;
class dgram_scribe; class dgram_scribe;
class dgram_doorman;
class receive_policy; class receive_policy;
class abstract_broker; class abstract_broker;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_DATAGRAM_HEADER_HPP
#define CAF_IO_NETWORK_DATAGRAM_HEADER_HPP
#include <cstdint>
#include "caf/meta/type_name.hpp"
namespace caf {
namespace io {
namespace network {
/// This header adds a sequence number and response port to
/// datagram messages carrying BASP messages.
struct datagram_header {
uint32_t sequence_number;
uint16_t response_port;
inline datagram_header(uint32_t seq, uint16_t port)
: sequence_number(seq),
response_port(port) {
// nop
}
datagram_header() = default;
};
/// @relates datagram_header
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, datagram_header& hdr) {
return f(meta::type_name("datagram_header"), hdr.sequence_number,
hdr.response_port);
}
/// @relates datagram_header
bool operator==(const datagram_header& lhs, const datagram_header& rhs);
/// @relates datagram_header
inline bool operator!=(const datagram_header& lhs, const datagram_header& rhs) {
return !(lhs == rhs);
}
/// Size of the datagram header in serialized form
constexpr size_t datagram_header_size = sizeof(uint32_t) + sizeof(uint16_t);
} // namespace network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_DATAGRAM_HEADER_HPP
...@@ -39,7 +39,6 @@ ...@@ -39,7 +39,6 @@
#include "caf/io/network/stream_manager.hpp" #include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp" #include "caf/io/network/acceptor_manager.hpp"
#include "caf/io/network/dgram_stream_manager.hpp" #include "caf/io/network/dgram_stream_manager.hpp"
#include "caf/io/network/dgram_acceptor_manager.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
...@@ -335,19 +334,6 @@ public: ...@@ -335,19 +334,6 @@ public:
const std::string& host, const std::string& host,
uint16_t port) override; uint16_t port) override;
expected<std::pair<dgram_doorman_handle, uint16_t>>
new_dgram_doorman(uint16_t port, const char* in, bool rflag) override;
expected<void> assign_dgram_doorman(abstract_broker* ptr,
dgram_doorman_handle hdl) override;
dgram_doorman_handle add_dgram_doorman(abstract_broker* ptr,
native_socket fd) override;
expected<std::pair<dgram_doorman_handle, uint16_t>>
add_dgram_doorman(abstract_broker* ptr, uint16_t port , const char* in,
bool rflag) override;
void exec_later(resumable* ptr) override; void exec_later(resumable* ptr) override;
explicit default_multiplexer(actor_system* sys); explicit default_multiplexer(actor_system* sys);
...@@ -439,10 +425,6 @@ inline dgram_scribe_handle dg_sink_hdl_from_socket(native_socket fd) { ...@@ -439,10 +425,6 @@ inline dgram_scribe_handle dg_sink_hdl_from_socket(native_socket fd) {
return dgram_scribe_handle::from_int(int64_from_native_socket(fd)); return dgram_scribe_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 /// A stream capable of both reading and writing. The stream's input
/// data is forwarded to its {@link stream_manager manager}. /// data is forwarded to its {@link stream_manager manager}.
class stream : public event_handler { class stream : public event_handler {
...@@ -635,7 +617,7 @@ public: ...@@ -635,7 +617,7 @@ public:
void removed_from_loop(operation op) override; void removed_from_loop(operation op) override;
void handle_event(operation op) override; void handle_event(operation op) override;
void set_endpoint_addr(const sockaddr_storage& sa, size_t len, void set_endpoint_addr(const sockaddr_storage& sa, size_t len,
bool is_tmp_endpoint); bool is_tmp_endpoint);
...@@ -670,74 +652,6 @@ private: ...@@ -670,74 +652,6 @@ private:
uint16_t port_; uint16_t port_;
}; };
/// A dgram_acceptor is responsible for receiving datagrams on an endpoint.
class dgram_acceptor: public event_handler {
public:
/// 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<dgram_acceptor_manager>;
/// A buffer class providing a compatible
/// interface to `std::vector`.
using buffer_type = std::vector<char>;
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
/// once the stream has been started.
void configure_datagram_size(size_t buf_size);
/// Returns the read buffer of this datagram receiver.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
inline buffer_type& rd_buf() {
return rd_buf_;
}
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 dgram_acceptor.
void activate(manager_type* mgr);
/// 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;
std::pair<const sockaddr_storage&,size_t> last_sender();
private:
void prepare_next_read();
// reader state
manager_ptr mgr_;
size_t dgram_size_;
buffer_type rd_buf_;
size_t bytes_read_;
// 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, expected<native_socket> new_tcp_connection(const std::string& host,
uint16_t port, uint16_t port,
optional<protocol> preferred = none); optional<protocol> preferred = none);
......
...@@ -34,7 +34,6 @@ ...@@ -34,7 +34,6 @@
#include "caf/io/accept_handle.hpp" #include "caf/io/accept_handle.hpp"
#include "caf/io/connection_handle.hpp" #include "caf/io/connection_handle.hpp"
#include "caf/io/dgram_scribe_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/protocol.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
...@@ -138,30 +137,6 @@ public: ...@@ -138,30 +137,6 @@ public:
add_dgram_scribe(abstract_broker*, const std::string& host, add_dgram_scribe(abstract_broker*, const std::string& host,
uint16_t port) = 0; 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<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_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 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<dgram_doorman_handle, uint16_t>>
add_dgram_doorman(abstract_broker* ptr, uint16_t port,
const char* in = nullptr, bool reuse_addr = false) = 0;
/// Simple wrapper for runnables /// Simple wrapper for runnables
class runnable : public resumable, public ref_counted { class runnable : public resumable, public ref_counted {
public: public:
......
...@@ -17,29 +17,40 @@ ...@@ -17,29 +17,40 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_IO_NETWORK_DGRAM_ACCEPTOR_MANGER_HPP #ifndef CAF_IO_REMOTE_ENDPOINT_HPP
#define CAF_IO_NETWORK_DGRAM_ACCEPTOR_MANGER_HPP #define CAF_IO_REMOTE_ENDPOINT_HPP
#include "caf/io/network/manager.hpp" #include <deque>
#include <vector>
namespace caf { namespace caf {
namespace io { namespace io {
namespace network { namespace network {
/// A datagram source manager provides callbacks for incoming class remote_endpoint {
/// datagrams as well as for error handling. /// A buffer class providing a compatible
class dgram_acceptor_manager : public manager { /// interface to `std::vector`.
public: using buffer_type = std::vector<char>;
dgram_acceptor_manager(abstract_broker* ptr);
private:
~dgram_acceptor_manager(); // state for receiving
size_t dgram_size_;
virtual bool new_endpoint(const void* buf, size_t num_bytes) = 0; buffer_type rd_buf_;
size_t bytes_read_;
// state for sending
bool ack_writes_;
bool writing_;
buffer_type wr_buf_;
std::deque<buffer_type> wr_offline_buf_;
// endpoint info
struct sockaddr_storage remote_endpoint_addr_;
socklen_t remote_endpoint_addr_len_;
}; };
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
#endif // CAF_IO_NETWORK_DGRAM_ACCEPTOR_MANGER_HPP #endif // CAF_IO_REMOTE_ENDPOINT_HPP
...@@ -31,7 +31,6 @@ ...@@ -31,7 +31,6 @@
#include "caf/io/accept_handle.hpp" #include "caf/io/accept_handle.hpp"
#include "caf/io/connection_handle.hpp" #include "caf/io/connection_handle.hpp"
#include "caf/io/dgram_scribe_handle.hpp" #include "caf/io/dgram_scribe_handle.hpp"
#include "caf/io/dgram_doorman_handle.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -140,51 +139,6 @@ inspect(Inspector& f, dgram_scribe_closed_msg& x) { ...@@ -140,51 +139,6 @@ inspect(Inspector& f, dgram_scribe_closed_msg& x) {
return f(meta::type_name("dgram_scribe_closed_msg"), x.handle); return f(meta::type_name("dgram_scribe_closed_msg"), x.handle);
} }
/// Signalizes that a {@link broker} dgram_acceptor has been closed.
struct dgram_acceptor_closed_msg {
/// Handle to the closed connection.
dgram_doorman_handle handle;
};
/// @relates connection_closed_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f,
dgram_acceptor_closed_msg& x) {
return f(meta::type_name("dgram_acceptor_closed_msg"), x.handle);
}
/// Signalizes that a {@link broker} datagram source was closed.
struct dgram_doorman_closed_msg {
dgram_doorman_handle handle;
};
/// @relates dgram_doorman_closed_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, dgram_doorman_closed_msg& x) {
return f(meta::type_name("dgram_doorman_closed_msg"), x.handle);
}
/// Signalizes newly discovered remote endpoint
/// and the responsible scribe to a {@link broker}.
struct new_endpoint_msg {
// Handle to the related datagram endpoint.
dgram_doorman_handle source;
// Buffer containing the received data.
std::vector<char> buf;
// Handle to new dgram_scribe
dgram_scribe_handle handle;
// Port of the addressed actor
uint16_t port;
};
/// @relates new_endpoint_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, new_endpoint_msg& x) {
return f(meta::type_name("new_endpoint_msg"), x.source, x.buf,
x.handle, x.port);
}
/// Signalizes that a datagram with a certain size has been sent. /// Signalizes that a datagram with a certain size has been sent.
struct new_datagram_msg { struct new_datagram_msg {
// Handle to the endpoint used. // Handle to the endpoint used.
...@@ -225,31 +179,6 @@ inspect(Inspector& f, dgram_scribe_passivated_msg& x) { ...@@ -225,31 +179,6 @@ inspect(Inspector& f, dgram_scribe_passivated_msg& x) {
return f(meta::type_name("dgram_scribe_passivated_msg"), x.handle); 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 dgram_doorman_passivated_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, dgram_doorman_passivated_msg& x) {
return f(meta::type_name("dgram_doorman_passivated_msg"), x.handle);
}
/// Dummy message for the dgram_doorman implementation
struct dgram_dummy_msg {
dgram_doorman_handle handle;
std::vector<char> buffer;
};
/// @relates dgram_dummy_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, dgram_dummy_msg& x) {
return f(meta::type_name("dgram_dummy_msg"), x.handle, x.buffer);
}
} // namespace io } // namespace io
} // namespace caf } // namespace caf
......
...@@ -117,14 +117,6 @@ void abstract_broker::ack_writes(dgram_scribe_handle hdl, bool enable) { ...@@ -117,14 +117,6 @@ void abstract_broker::ack_writes(dgram_scribe_handle hdl, bool enable) {
x->ack_writes(enable); x->ack_writes(enable);
} }
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);
if (x)
x->configure_datagram_size(buf_size);
}
void abstract_broker::configure_datagram_size(dgram_scribe_handle hdl, void abstract_broker::configure_datagram_size(dgram_scribe_handle hdl,
size_t buf_size) { size_t buf_size) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(buf_size)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(buf_size));
...@@ -239,32 +231,6 @@ abstract_broker::add_dgram_scribe(network::native_socket fd) { ...@@ -239,32 +231,6 @@ abstract_broker::add_dgram_scribe(network::native_socket fd) {
return backend().add_dgram_scribe(this, fd); return backend().add_dgram_scribe(this, fd);
} }
void
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<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_dgram_doorman(this, port, in, reuse_addr);
}
expected<void>
abstract_broker::assign_dgram_doorman(dgram_doorman_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
return backend().assign_dgram_doorman(this, hdl);
}
expected<dgram_doorman_handle>
abstract_broker::add_dgram_doorman(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return backend().add_dgram_doorman(this, fd);
}
std::string abstract_broker::remote_addr(connection_handle hdl) { std::string abstract_broker::remote_addr(connection_handle hdl) {
auto i = scribes_.find(hdl); auto i = scribes_.find(hdl);
return i != scribes_.end() ? i->second->addr() : std::string{}; return i != scribes_.end() ? i->second->addr() : std::string{};
...@@ -300,11 +266,6 @@ uint16_t abstract_broker::local_port(dgram_scribe_handle hdl) { ...@@ -300,11 +266,6 @@ uint16_t abstract_broker::local_port(dgram_scribe_handle hdl) {
return i != dgram_scribes_.end() ? i->second->local_port() : 0; return i != dgram_scribes_.end() ? i->second->local_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->local_port() : 0;
}
accept_handle abstract_broker::hdl_by_port(uint16_t port) { accept_handle abstract_broker::hdl_by_port(uint16_t port) {
for (auto& kvp : doormen_) for (auto& kvp : doormen_)
if (kvp.second->port() == port) if (kvp.second->port() == port)
...@@ -322,10 +283,6 @@ void abstract_broker::close_all() { ...@@ -322,10 +283,6 @@ void abstract_broker::close_all() {
// stop_reading will remove the scribe from scribes_ // stop_reading will remove the scribe from scribes_
scribes_.begin()->second->stop_reading(); scribes_.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()) { while (!dgram_scribes_.empty()) {
// stop_reading will remove the scribe from dgram_scribes_ // stop_reading will remove the scribe from dgram_scribes_
dgram_scribes_.begin()->second->stop_reading(); dgram_scribes_.begin()->second->stop_reading();
......
...@@ -717,6 +717,7 @@ behavior basp_broker::make_behavior() { ...@@ -717,6 +717,7 @@ behavior basp_broker::make_behavior() {
auto port = local_port(msg.handle); auto port = local_port(msg.handle);
state.instance.remove_published_actor(port); state.instance.remove_published_actor(port);
}, },
/*
// received from underlying broker implementation // received from underlying broker implementation
[=](new_endpoint_msg& msg) { [=](new_endpoint_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
...@@ -733,6 +734,7 @@ behavior basp_broker::make_behavior() { ...@@ -733,6 +734,7 @@ behavior basp_broker::make_behavior() {
configure_datagram_size(msg.handle, 1500); configure_datagram_size(msg.handle, 1500);
} }
}, },
*/
// received from underlying broker implementation // received from underlying broker implementation
[=](const dgram_scribe_closed_msg& msg) { [=](const dgram_scribe_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
...@@ -746,11 +748,6 @@ behavior basp_broker::make_behavior() { ...@@ -746,11 +748,6 @@ behavior basp_broker::make_behavior() {
CAF_ASSERT(nid == none CAF_ASSERT(nid == none
|| !state.instance.tbl().reachable(nid)); || !state.instance.tbl().reachable(nid));
}, },
[=](const dgram_doorman_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
auto port = local_port(msg.handle);
state.instance.remove_published_actor(port);
},
// received from middleman actor // received from middleman actor
[=](publish_atom, accept_handle hdl, uint16_t port, [=](publish_atom, accept_handle hdl, uint16_t port,
const strong_actor_ptr& whom, std::set<std::string>& sigs) { const strong_actor_ptr& whom, std::set<std::string>& sigs) {
...@@ -769,6 +766,7 @@ behavior basp_broker::make_behavior() { ...@@ -769,6 +766,7 @@ behavior basp_broker::make_behavior() {
<< CAF_ARG(whom)); << CAF_ARG(whom));
} }
}, },
/*
[=](publish_atom, dgram_doorman_handle hdl, uint16_t port, [=](publish_atom, dgram_doorman_handle hdl, uint16_t port,
const strong_actor_ptr& whom, std::set<std::string>& sigs) { const strong_actor_ptr& whom, std::set<std::string>& sigs) {
CAF_LOG_TRACE(CAF_ARG(hdl.id()) << CAF_ARG(whom) << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(hdl.id()) << CAF_ARG(whom) << CAF_ARG(port));
...@@ -786,6 +784,7 @@ behavior basp_broker::make_behavior() { ...@@ -786,6 +784,7 @@ behavior basp_broker::make_behavior() {
<< CAF_ARG(whom)); << CAF_ARG(whom));
} }
}, },
*/
// received from middleman actor (delegated) // received from middleman actor (delegated)
[=](connect_atom, connection_handle hdl, uint16_t port) { [=](connect_atom, connection_handle hdl, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hdl.id())); CAF_LOG_TRACE(CAF_ARG(hdl.id()));
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_header.hpp"
#include <sstream>
namespace caf {
namespace io {
namespace network {
bool operator==(const datagram_header& lhs, const datagram_header& rhs) {
return lhs.sequence_number == rhs.sequence_number
&& lhs.response_port == rhs.response_port;
}
} // namespace network
} // namespace io
} // namespace caf
...@@ -982,87 +982,6 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd, ...@@ -982,87 +982,6 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd,
return ptr->hdl(); return ptr->hdl();
} }
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 dgram_doorman {
public:
impl(abstract_broker* ptr, default_multiplexer& mx, native_socket 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("");
acceptor_.configure_datagram_size(buf_size);
}
bool new_endpoint(const void* buf, size_t num_bytes) override {
CAF_LOG_TRACE("");
// TODO: this currently ignores the payload of the datagram
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();
//auto endpoint_info = acceptor_.last_sender();
auto fd = new_dgram_scribe_impl(acceptor_.host(),
acceptor_.port());
if (!fd) {
CAF_LOG_ERROR(CAF_ARG(fd.error()));
return false;
// return std::move(fd.error());
}
auto endpoint_info = acceptor_.last_sender();
auto hdl = dm.add_dgram_scribe(parent(), *fd,
endpoint_info.first, endpoint_info.second,
false);
return dgram_doorman::new_endpoint(&dm, hdl, buf, num_bytes);
}
void stop_reading() override {
CAF_LOG_TRACE("");
acceptor_.stop_reading();
detach(&acceptor_.backend(), false);
}
std::vector<char>& rd_buf() override {
return acceptor_.rd_buf();
}
std::string addr() const override {
return acceptor_.host();
}
uint16_t port() const override {
return acceptor_.port();
}
uint16_t local_port() const override {
auto x = local_port_of_fd(acceptor_.fd());
if (!x)
return 0;
return *x;
}
void launch() override {
CAF_LOG_TRACE("");
CAF_ASSERT(!launched_);
launched_ = true;
acceptor_.start(this);
}
void add_to_loop() override {
acceptor_.activate(this);
}
void remove_from_loop() override {
acceptor_.passivate();
}
private:
bool launched_;
network::dgram_acceptor acceptor_;
};
auto ptr = make_counted<impl>(self, *this, fd);
self->add_dgram_doorman(ptr);
return ptr->hdl();
}
expected<connection_handle> expected<connection_handle>
default_multiplexer::new_tcp_scribe(const std::string& host, uint16_t port) { default_multiplexer::new_tcp_scribe(const std::string& host, uint16_t port) {
auto fd = new_tcp_connection(host, port); auto fd = new_tcp_connection(host, port);
...@@ -1154,35 +1073,6 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, ...@@ -1154,35 +1073,6 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self,
return add_dgram_scribe(self, *fd); return add_dgram_scribe(self, *fd);
} }
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_dgram_doorman_impl(port, in, reuse_addr);
if (!res)
return std::move(res.error());
return std::make_pair(dgram_doorman_handle::from_int(
int64_from_native_socket(res->first)),
res->second);
}
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_dgram_doorman(self, static_cast<native_socket>(hdl.id()));
return unit;
}
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_dgram_doorman_impl(port, in, reuse_addr);
if (!res)
return std::move(res.error());
auto bound_port = res->second;
return std::make_pair(add_dgram_doorman(self, res->first), bound_port);
}
/****************************************************************************** /******************************************************************************
* platform-independent implementations (finally) * * platform-independent implementations (finally) *
******************************************************************************/ ******************************************************************************/
...@@ -1732,93 +1622,6 @@ void dgram_stream::set_endpoint_addr(const sockaddr_storage& sa, ...@@ -1732,93 +1622,6 @@ void dgram_stream::set_endpoint_addr(const sockaddr_storage& sa,
std::tie(host_, port_) = sender_from_sockaddr(sa, len); std::tie(host_, port_) = sender_from_sockaddr(sa, len);
} }
dgram_acceptor::dgram_acceptor(default_multiplexer& backend_ref,
native_socket sockfd)
: event_handler(backend_ref, sockfd),
dgram_size_(1500),
sockaddr_len_(0) {
// TODO: Set reasonable default for datagram_size
configure_datagram_size(1500);
// nop
}
void dgram_acceptor::configure_datagram_size(size_t size) {
dgram_size_ = size;
}
void dgram_acceptor::start(manager_type* mgr) {
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
void dgram_acceptor::activate(manager_type* mgr) {
if (!mgr_) {
mgr_.reset(mgr);
event_handler::activate();
prepare_next_read();
}
}
void dgram_acceptor::stop_reading() {
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
void dgram_acceptor::removed_from_loop(operation op) {
if (op == operation::read)
mgr_.reset();
}
void dgram_acceptor::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(fd()) << CAF_ARG(op));
switch (op) {
case operation::read: {
// TODO: some idenfification to prevent arbitrary
// message from creating endpoints?
size_t rb;
if (!receive_datagram(rb, fd(), rd_buf_.data(), rd_buf_.size(),
sockaddr_, sockaddr_len_)) {
mgr_->io_failure(&backend(), operation::read);
passivate();
return;
}
bytes_read_ = rb;
if (rb > 0) {
std::tie(host_,port_) = sender_from_sockaddr(sockaddr_, sockaddr_len_);
// TODO: Add this once messages are only received once
auto res = mgr_->new_endpoint(rd_buf_.data(), rb);
if (!res) {
// What is the right way to propagate this?
CAF_LOG_DEBUG("Failure during creation of new udp endpoint");
// std::cerr << "[DA] Failure during creation of new udp endpoint"
// << std::endl;;
}
}
prepare_next_read();
break;
}
case operation::write: {
// This should not happen ...
break;
}
case operation::propagate_error:
if (mgr_)
mgr_->io_failure(&backend(), operation::read);
// backend will delete this handler anyway,
// no need to call backend().del() here
break;
}
}
std::pair<const sockaddr_storage&,size_t> dgram_acceptor::last_sender() {
return std::make_pair(std::ref(sockaddr_), sockaddr_len_);
}
void dgram_acceptor::prepare_next_read() {
rd_buf_.resize(dgram_size_);
}
class socket_guard { class socket_guard {
public: public:
explicit socket_guard(native_socket fd) : fd_(fd) { explicit socket_guard(native_socket fd) : fd_(fd) {
...@@ -2066,43 +1869,6 @@ expected<native_socket> new_dgram_scribe_impl(const std::string& host, ...@@ -2066,43 +1869,6 @@ expected<native_socket> new_dgram_scribe_impl(const std::string& host,
return sguard.release(); return sguard.release();
} }
expected<std::pair<native_socket, uint16_t>>
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) {
auto addrs = interfaces::native_address(addr);
if (!addrs) {
//CAF_LOG_INFO("no such host");
//return make_error(sec::cannot_connect_to_node, "no such host",
// addr, port);
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());
// ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(p));
//std::cerr << "[NDDI] Bound to " << *p << std::endl;
return std::make_pair(sguard.release(), *p);
}
expected<std::string> local_addr_of_fd(native_socket fd) { expected<std::string> local_addr_of_fd(native_socket fd) {
sockaddr_storage st; sockaddr_storage st;
socklen_t st_len = sizeof(st); socklen_t st_len = sizeof(st);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/dgram_acceptor_manager.hpp"
namespace caf {
namespace io {
namespace network {
dgram_acceptor_manager::dgram_acceptor_manager(abstract_broker* ptr)
: manager(ptr) {
// nop
}
dgram_acceptor_manager::~dgram_acceptor_manager() {
// nop
}
} // namespace network
} // 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/dgram_doorman.hpp"
#include "caf/logger.hpp"
namespace caf {
namespace io {
dgram_doorman::dgram_doorman(abstract_broker* parent, dgram_doorman_handle hdl)
: dgram_doorman_base(parent, hdl) {
// nop
}
dgram_doorman::~dgram_doorman() {
CAF_LOG_TRACE("");
}
message dgram_doorman::detach_message() {
return make_message(dgram_doorman_closed_msg{hdl()});
}
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,
dgram_scribe_handle endpoint,
const void*, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(endpoint) << CAF_ARG(num_bytes));
if (detached())
return false;
using endpoint_t = new_endpoint_msg;
using tmp_t = mailbox_element_vals<endpoint_t>;
auto guard = parent_;
auto& buf = rd_buf();
CAF_ASSERT(buf.size() >= num_bytes);
buf.resize(num_bytes);
tmp_t tmp{strong_actor_ptr{}, message_id::make(),
mailbox_element::forwarding_stack{},
endpoint_t{hdl(), std::move(buf), endpoint,
parent()->local_port(hdl())}};
invoke_mailbox_element_impl(ctx,tmp);
return true;
}
} // namespace io
} // namespace caf
...@@ -258,6 +258,7 @@ private: ...@@ -258,6 +258,7 @@ private:
std::move(whom), std::move(sigs)); std::move(whom), std::move(sigs));
} else if (std::distance(u.scheme().first, u.scheme().second) >= 3 && } else if (std::distance(u.scheme().first, u.scheme().second) >= 3 &&
equal(u.scheme().first, u.scheme().second, std::begin(udp))) { equal(u.scheme().first, u.scheme().second, std::begin(udp))) {
/*
// UDP // UDP
auto res auto res
= system().middleman().backend().new_dgram_doorman(u.port_as_int(), = system().middleman().backend().new_dgram_doorman(u.port_as_int(),
...@@ -268,6 +269,7 @@ private: ...@@ -268,6 +269,7 @@ private:
actual_port = res->second; actual_port = res->second;
anon_send(broker_, publish_atom::value, hdl, actual_port, anon_send(broker_, publish_atom::value, hdl, actual_port,
std::move(whom), std::move(sigs)); std::move(whom), std::move(sigs));
*/
} else { } else {
return sec::unsupported_protocol; return sec::unsupported_protocol;
} }
......
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