Commit e9541de1 authored by Jakob Otto's avatar Jakob Otto

Merge branch 'master' into topic/datagram-transport

parents 7ec63266 47b45d8e
...@@ -8,10 +8,16 @@ file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp") ...@@ -8,10 +8,16 @@ file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
# list cpp files excluding platform-dependent files # list cpp files excluding platform-dependent files
set(LIBCAF_NET_SRCS set(LIBCAF_NET_SRCS
src/actor_proxy_impl.cpp src/actor_proxy_impl.cpp
src/application.cpp
src/connection_state.cpp
src/convert_ip_endpoint.cpp
src/datagram_socket.cpp src/datagram_socket.cpp
src/ec.cpp
src/endpoint_manager.cpp src/endpoint_manager.cpp
src/header.cpp
src/host.cpp src/host.cpp
src/ip.cpp src/ip.cpp
src/message_type.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/network_socket.cpp src/network_socket.cpp
src/pipe_socket.cpp src/pipe_socket.cpp
...@@ -19,10 +25,10 @@ set(LIBCAF_NET_SRCS ...@@ -19,10 +25,10 @@ set(LIBCAF_NET_SRCS
src/socket.cpp src/socket.cpp
src/socket_manager.cpp src/socket_manager.cpp
src/stream_socket.cpp src/stream_socket.cpp
src/convert_ip_endpoint.cpp
src/udp_datagram_socket.cpp
src/tcp_stream_socket.cpp
src/tcp_accept_socket.cpp src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
src/tcp_stream_socket.cpp
src/udp_datagram_socket.cpp
) )
add_custom_target(libcaf_net) add_custom_target(libcaf_net)
......
...@@ -24,9 +24,9 @@ ...@@ -24,9 +24,9 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
sockaddr_storage to_sockaddr(const ip_endpoint& ep); void convert(const ip_endpoint& src, sockaddr_storage& dst);
ip_endpoint to_ip_endpoint(const sockaddr_storage& addr); error convert(const sockaddr_storage& src, ip_endpoint& dst);
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <memory>
#include <unordered_map>
#include <unordered_set>
#include <vector>
#include "caf/actor_addr.hpp"
#include "caf/byte.hpp"
#include "caf/callback.hpp"
#include "caf/error.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/net/basp/message_type.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/node_id.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/response_promise.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
namespace caf {
namespace net {
namespace basp {
/// An implementation of BASP as an application layer protocol.
class application {
public:
// -- member types -----------------------------------------------------------
using buffer_type = std::vector<byte>;
using byte_span = span<const byte>;
using write_packet_callback = callback<byte_span, byte_span>;
using proxy_registry_ptr = std::shared_ptr<proxy_registry>;
// -- constructors, destructors, and assignment operators --------------------
explicit application(proxy_registry_ptr proxies);
// -- interface functions ----------------------------------------------------
template <class Parent>
error init(Parent& parent) {
// Initialize member variables.
system_ = &parent.system();
// Write handshake.
if (auto err = generate_handshake())
return err;
auto hdr = to_bytes(header{message_type::handshake,
static_cast<uint32_t>(buf_.size()), version});
parent.write_packet(hdr, buf_);
parent.transport().configure_read(receive_policy::exactly(header_size));
return none;
}
template <class Parent>
error write_message(Parent& parent,
std::unique_ptr<endpoint_manager::message> ptr) {
// TODO: avoid extra copy of the payload
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver;
if (dst == nullptr) {
// TODO: valid?
return none;
}
if (src != nullptr) {
if (auto err = sink(src->node(), src->id(), dst->id(), ptr->msg->stages))
return err;
} else {
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages))
return err;
}
buf_.insert(buf_.end(), ptr->payload.begin(), ptr->payload.end());
header hdr{message_type::actor_message, static_cast<uint32_t>(buf_.size()),
ptr->msg->mid.integer_value()};
auto bytes = to_bytes(hdr);
parent.write_packet(make_span(bytes), make_span(buf_));
return none;
}
template <class Parent>
error handle_data(Parent& parent, byte_span bytes) {
auto write_packet = make_callback([&](byte_span hdr, byte_span payload) {
parent.write_packet(hdr, payload);
return none;
});
size_t next_read_size = header_size;
if (auto err = handle(next_read_size, write_packet, bytes))
return err;
parent.transport().configure_read(receive_policy::exactly(next_read_size));
return none;
}
template <class Parent>
void resolve(Parent& parent, string_view path, actor listener) {
auto write_packet = make_callback([&](byte_span hdr, byte_span payload) {
parent.write_packet(hdr, payload);
return none;
});
resolve_remote_path(write_packet, path, listener);
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
// nop
}
void handle_error(sec) {
// nop
}
static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x);
// -- utility functions ------------------------------------------------------
strong_actor_ptr resolve_local_path(string_view path);
void resolve_remote_path(write_packet_callback& write_packet,
string_view path, actor listener);
// -- properties -------------------------------------------------------------
connection_state state() const noexcept {
return state_;
}
actor_system& system() const noexcept {
return *system_;
}
private:
// -- message handling -------------------------------------------------------
error handle(size_t& next_read_size, write_packet_callback& write_packet,
byte_span bytes);
error handle(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_handshake(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_actor_message(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_resolve_request(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_resolve_response(write_packet_callback& write_packet, header hdr,
byte_span payload);
/// Writes the handshake payload to `buf_`.
error generate_handshake();
// -- member variables -------------------------------------------------------
/// Stores a pointer to the parent actor system.
actor_system* system_ = nullptr;
/// Stores the expected type of the next incoming message.
connection_state state_ = connection_state::await_handshake_header;
/// Caches the last header while waiting for the matching payload.
header hdr_;
/// Re-usable buffer for storing payloads.
buffer_type buf_;
/// Stores our own ID.
node_id id_;
/// Stores the ID of our peer.
node_id peer_id_;
/// Tracks which local actors our peer monitors.
std::unordered_set<actor_addr> monitored_actors_;
/// Caches actor handles obtained via `resolve`.
std::unordered_map<uint64_t, response_promise> pending_resolves_;
/// Ascending ID generator for requests to our peer.
uint64_t next_request_id_ = 1;
/// Points to the factory object for generating proxies.
proxy_registry_ptr proxies_;
};
} // namespace basp
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <string>
namespace caf {
namespace net {
namespace basp {
/// @addtogroup BASP
/// Stores the state of a connection in a `basp::application`.
enum class connection_state {
/// Initial state for any connection to wait for the peer's handshake.
await_handshake_header,
/// Indicates that the header for the peer's handshake arrived and BASP
/// requires the payload next.
await_handshake_payload,
/// Indicates that a connection is established and this node is waiting for
/// the next BASP header.
await_header,
/// Indicates that this node has received a header with non-zero payload and
/// is waiting for the data.
await_payload,
/// Indicates that the connection is about to shut down.
shutdown,
};
/// @relates connection_state
std::string to_string(connection_state x);
/// @}
} // namespace basp
} // namespace net
} // namespace caf
...@@ -19,51 +19,23 @@ ...@@ -19,51 +19,23 @@
#pragma once #pragma once
#include <cstddef> #include <cstddef>
#include <limits> #include <cstdint>
#include "caf/net/socket_id.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
namespace basp {
template <class Derived> /// @addtogroup BASP
struct abstract_socket {
socket_id id;
constexpr abstract_socket() : id(invalid_socket_id) {
// nop
}
constexpr abstract_socket(socket_id id) : id(id) {
// nop
}
constexpr abstract_socket(const Derived& other) : id(other.id) {
// nop
}
abstract_socket& operator=(const Derived& other) {
id = other.id;
return *this;
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, Derived& x) {
return f(x.id);
}
friend constexpr bool operator==(Derived x, Derived y) { /// The current BASP version.
return x.id == y.id; /// @note BASP is not backwards compatible.
} constexpr uint64_t version = 1;
friend constexpr bool operator!=(Derived x, Derived y) { /// Size of a BASP header in serialized form.
return x.id != y.id; constexpr size_t header_size = 13;
}
friend constexpr bool operator<(Derived x, Derived y) { /// @}
return x.id < y.id;
}
};
} // namespace basp
} // namespace net } // namespace net
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <string>
#include "caf/fwd.hpp"
namespace caf {
namespace net {
namespace basp {
/// BASP-specific error codes.
enum class ec : uint8_t {
invalid_magic_number = 1,
unexpected_number_of_bytes,
unexpected_payload,
missing_payload,
illegal_state,
invalid_handshake,
missing_handshake,
unexpected_handshake,
version_mismatch,
unimplemented = 10,
app_identifiers_mismatch,
invalid_payload,
};
/// @relates ec
std::string to_string(ec x);
/// @relates ec
error make_error(ec x);
} // namespace basp
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <array>
#include <cstdint>
#include "caf/detail/comparable.hpp"
#include "caf/fwd.hpp"
#include "caf/meta/hex_formatted.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/message_type.hpp"
namespace caf {
namespace net {
namespace basp {
/// @addtogroup BASP
/// The header of a Binary Actor System Protocol (BASP) message.
struct header : detail::comparable<header> {
// -- constructors, destructors, and assignment operators --------------------
constexpr header() noexcept
: type(message_type::handshake), payload_len(0), operation_data(0) {
// nop
}
constexpr header(message_type type, uint32_t payload_len,
uint64_t operation_data) noexcept
: type(type), payload_len(payload_len), operation_data(operation_data) {
// nop
}
header(const header&) noexcept = default;
header& operator=(const header&) noexcept = default;
// -- factory functions ------------------------------------------------------
/// @pre `bytes.size() == header_size`
static header from_bytes(span<const byte> bytes);
// -- comparison -------------------------------------------------------------
int compare(header other) const noexcept;
// -- member variables -------------------------------------------------------
/// Denotes the BASP operation and how `operation_data` gets interpreted.
message_type type;
/// Stores the size in bytes for the payload that follows this header.
uint32_t payload_len;
/// Stores type-specific information such as the BASP version in handshakes.
uint64_t operation_data;
};
/// Serializes a header to a byte representation.
/// @relates header
std::array<byte, header_size> to_bytes(header x);
/// @relates header
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, header& x) {
return f(meta::type_name("basp::header"), x.type, x.payload_len,
x.operation_data);
}
/// @}
} // namespace basp
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <string>
namespace caf {
namespace net {
namespace basp {
/// @addtogroup BASP
/// Describes the first header field of a BASP message and determines the
/// interpretation of the other header fields.
enum class message_type : uint8_t {
/// Sends supported BASP version and node information to the server.
///
/// ![](client_handshake.png)
handshake = 0,
/// Transmits an actor-to-actor messages.
///
/// ![](direct_message.png)
actor_message = 1,
/// Tries to resolve a path on the receiving node.
///
/// ![](resolve_request.png)
resolve_request = 2,
/// Transmits the result of a path lookup.
///
/// ![](resolve_response.png)
resolve_response = 3,
/// Informs the receiving node that the sending node has created a proxy
/// instance for one of its actors. Causes the receiving node to attach a
/// functor to the actor that triggers a down_message on termination.
///
/// ![](monitor_message.png)
monitor_message = 4,
/// Informs the receiving node that it has a proxy for an actor that has been
/// terminated.
///
/// ![](down_message.png)
down_message = 5,
/// Used to generate periodic traffic between two nodes in order to detect
/// disconnects.
///
/// ![](heartbeat.png)
heartbeat = 6,
};
/// @relates message_type
std::string to_string(message_type);
/// @}
} // namespace basp
} // namespace net
} // namespace caf
...@@ -25,18 +25,10 @@ namespace caf { ...@@ -25,18 +25,10 @@ namespace caf {
namespace net { namespace net {
/// A datagram-oriented network communication endpoint. /// A datagram-oriented network communication endpoint.
struct datagram_socket : abstract_socket<datagram_socket> { struct datagram_socket : network_socket {
using super = abstract_socket<datagram_socket>; using super = network_socket;
using super::super; using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
constexpr operator network_socket() const noexcept {
return network_socket{id};
}
}; };
/// Enables or disables `SIO_UDP_CONNRESET` error on `x`. /// Enables or disables `SIO_UDP_CONNRESET` error on `x`.
......
...@@ -91,10 +91,14 @@ public: ...@@ -91,10 +91,14 @@ public:
/// Original message to a remote actor. /// Original message to a remote actor.
mailbox_element_ptr msg; mailbox_element_ptr msg;
/// ID of the receiving actor.
strong_actor_ptr receiver;
/// Serialized representation of of `msg->content()`. /// Serialized representation of of `msg->content()`.
std::vector<byte> payload; std::vector<byte> payload;
message(mailbox_element_ptr msg, std::vector<byte> payload); message(mailbox_element_ptr msg, strong_actor_ptr receiver,
std::vector<byte> payload);
}; };
struct message_policy { struct message_policy {
...@@ -136,7 +140,8 @@ public: ...@@ -136,7 +140,8 @@ public:
void resolve(std::string path, actor listener); void resolve(std::string path, actor listener);
/// Enqueues a message to the endpoint. /// Enqueues a message to the endpoint.
void enqueue(mailbox_element_ptr msg, std::vector<byte> payload); void enqueue(mailbox_element_ptr msg, strong_actor_ptr receiver,
std::vector<byte> payload);
/// Enqueues a timeout to the endpoint. /// Enqueues a timeout to the endpoint.
void enqueue(timeout_msg msg); void enqueue(timeout_msg msg);
......
...@@ -18,6 +18,10 @@ ...@@ -18,6 +18,10 @@
#pragma once #pragma once
#include "caf/abstract_actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp"
#include "caf/atom.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
namespace caf { namespace caf {
...@@ -48,10 +52,18 @@ public: ...@@ -48,10 +52,18 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
application_type& application() {
return transport_.application();
}
transport_type& transport() { transport_type& transport() {
return transport_; return transport_;
} }
endpoint_manager_impl& manager() {
return *this;
}
// -- timeout management ----------------------------------------------------- // -- timeout management -----------------------------------------------------
template <class... Ts> template <class... Ts>
......
...@@ -29,6 +29,8 @@ class multiplexer; ...@@ -29,6 +29,8 @@ class multiplexer;
class socket_manager; class socket_manager;
template <class Application, class IdType = unit_t> template <class Application, class IdType = unit_t>
class transport_worker; class transport_worker;
template <class Application, class IdType = unit_t>
class transport_worker_dispatcher;
struct network_socket; struct network_socket;
struct pipe_socket; struct pipe_socket;
......
...@@ -25,7 +25,6 @@ ...@@ -25,7 +25,6 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/abstract_socket.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
#include "caf/net/socket_id.hpp" #include "caf/net/socket_id.hpp"
...@@ -33,14 +32,10 @@ namespace caf { ...@@ -33,14 +32,10 @@ namespace caf {
namespace net { namespace net {
/// A bidirectional network communication endpoint. /// A bidirectional network communication endpoint.
struct network_socket : abstract_socket<network_socket> { struct network_socket : socket {
using super = abstract_socket<network_socket>; using super = socket;
using super::super; using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
}; };
/// Enables or disables `SIGPIPE` events from `x`. /// Enables or disables `SIGPIPE` events from `x`.
......
...@@ -23,7 +23,6 @@ ...@@ -23,7 +23,6 @@
#include <utility> #include <utility>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/abstract_socket.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
#include "caf/net/socket_id.hpp" #include "caf/net/socket_id.hpp"
...@@ -31,14 +30,10 @@ namespace caf { ...@@ -31,14 +30,10 @@ namespace caf {
namespace net { namespace net {
/// A unidirectional communication endpoint for inter-process communication. /// A unidirectional communication endpoint for inter-process communication.
struct pipe_socket : abstract_socket<pipe_socket> { struct pipe_socket : socket {
using super = abstract_socket<pipe_socket>; using super = socket;
using super::super; using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
}; };
/// Creates two connected sockets. The first socket is the read handle and the /// Creates two connected sockets. The first socket is the read handle and the
......
...@@ -23,8 +23,8 @@ ...@@ -23,8 +23,8 @@
#include <type_traits> #include <type_traits>
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/detail/comparable.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/abstract_socket.hpp"
#include "caf/net/socket_id.hpp" #include "caf/net/socket_id.hpp"
namespace caf { namespace caf {
...@@ -32,12 +32,33 @@ namespace net { ...@@ -32,12 +32,33 @@ namespace net {
/// An internal endpoint for sending or receiving data. Can be either a /// An internal endpoint for sending or receiving data. Can be either a
/// ::network_socket, ::pipe_socket, ::stream_socket, or ::datagram_socket. /// ::network_socket, ::pipe_socket, ::stream_socket, or ::datagram_socket.
struct socket : abstract_socket<socket> { struct socket : detail::comparable<socket> {
using super = abstract_socket<socket>; socket_id id;
using super::super; constexpr socket() noexcept : id(invalid_socket_id) {
// nop
}
constexpr explicit socket(socket_id id) noexcept : id(id) {
// nop
}
constexpr socket(const socket& other) noexcept = default;
socket& operator=(const socket& other) noexcept = default;
constexpr signed_socket_id compare(socket other) const noexcept {
return static_cast<signed_socket_id>(id)
- static_cast<signed_socket_id>(other.id);
}
}; };
/// @relates socket
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, socket& x) {
return f(x.id);
}
/// Denotes the invalid socket. /// Denotes the invalid socket.
constexpr auto invalid_socket = socket{invalid_socket_id}; constexpr auto invalid_socket = socket{invalid_socket_id};
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include <cstddef> #include <cstddef>
#include <limits> #include <limits>
#include <type_traits>
#include "caf/config.hpp" #include "caf/config.hpp"
...@@ -46,5 +47,9 @@ constexpr socket_id invalid_socket_id = -1; ...@@ -46,5 +47,9 @@ constexpr socket_id invalid_socket_id = -1;
#endif // CAF_WINDOWS #endif // CAF_WINDOWS
/// Signed counterpart of `socket_id`.
/// @relates socket
using signed_socket_id = std::make_signed<socket_id>::type;
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -26,18 +26,10 @@ namespace net { ...@@ -26,18 +26,10 @@ namespace net {
/// A connection-oriented network communication endpoint for bidirectional byte /// A connection-oriented network communication endpoint for bidirectional byte
/// streams. /// streams.
struct stream_socket : abstract_socket<stream_socket> { struct stream_socket : network_socket {
using super = abstract_socket<stream_socket>; using super = network_socket;
using super::super; using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
constexpr operator network_socket() const noexcept {
return network_socket{id};
}
}; };
/// Creates two connected sockets to mimic network communication (usually for /// Creates two connected sockets to mimic network communication (usually for
......
...@@ -41,6 +41,10 @@ public: ...@@ -41,6 +41,10 @@ public:
using application_type = Application; using application_type = Application;
using transport_type = stream_transport;
using worker_type = transport_worker<application_type>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket handle, application_type application) stream_transport(stream_socket handle, application_type application)
...@@ -51,7 +55,8 @@ public: ...@@ -51,7 +55,8 @@ public:
collected_(0), collected_(0),
max_(1024), max_(1024),
rd_flag_(net::receive_policy_flag::exactly), rd_flag_(net::receive_policy_flag::exactly),
written_(0) { written_(0),
manager_(nullptr) {
// nop // nop
} }
...@@ -61,18 +66,35 @@ public: ...@@ -61,18 +66,35 @@ public:
return handle_; return handle_;
} }
actor_system& system() {
return manager().system();
}
application_type& application() {
return worker_.application();
}
transport_type& transport() {
return *this;
}
endpoint_manager& manager() {
return *manager_;
}
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
if (auto err = worker_.init(parent)) manager_ = &parent;
if (auto err = worker_.init(*this))
return err; return err;
parent.mask_add(operation::read); parent.mask_add(operation::read);
return none; return none;
} }
template <class Parent> template <class Parent>
bool handle_read_event(Parent& parent) { bool handle_read_event(Parent&) {
auto buf = read_buf_.data() + collected_; auto buf = read_buf_.data() + collected_;
size_t len = read_threshold_ - collected_; size_t len = read_threshold_ - collected_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len)); CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len));
...@@ -82,8 +104,7 @@ public: ...@@ -82,8 +104,7 @@ public:
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes)); CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
collected_ += *num_bytes; collected_ += *num_bytes;
if (collected_ >= read_threshold_) { if (collected_ >= read_threshold_) {
auto decorator = make_write_packet_decorator(*this, parent); worker_.handle_data(*this, read_buf_);
worker_.handle_data(decorator, read_buf_);
prepare_next_read(); prepare_next_read();
} }
return true; return true;
...@@ -103,16 +124,15 @@ public: ...@@ -103,16 +124,15 @@ public:
// TODO: dont read all messages at once - get one by one. // TODO: dont read all messages at once - get one by one.
for (auto msg = parent.next_message(); msg != nullptr; for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) { msg = parent.next_message()) {
auto decorator = make_write_packet_decorator(*this, parent); worker_.write_message(*this, std::move(msg));
worker_.write_message(decorator, std::move(msg));
} }
// Write prepared data. // Write prepared data.
return write_some(); return write_some();
} }
template <class Parent> template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) { void resolve(Parent&, const std::string& path, actor listener) {
worker_.resolve(parent, path, listener); worker_.resolve(*this, path, listener);
} }
template <class... Ts> template <class... Ts>
...@@ -121,8 +141,8 @@ public: ...@@ -121,8 +141,8 @@ public:
} }
template <class Parent> template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) { void timeout(Parent&, atom_value value, uint64_t id) {
worker_.timeout(parent, value, id); worker_.timeout(*this, value, id);
} }
void handle_error(sec code) { void handle_error(sec code) {
...@@ -163,9 +183,10 @@ public: ...@@ -163,9 +183,10 @@ public:
prepare_next_read(); prepare_next_read();
} }
template <class Parent> void write_packet(span<const byte> header, span<const byte> payload,
void write_packet(Parent&, span<const byte> header, span<const byte> payload, typename worker_type::id_type) {
unit_t) { if (write_buf_.empty())
manager().mask_add(operation::write);
write_buf_.insert(write_buf_.end(), header.begin(), header.end()); write_buf_.insert(write_buf_.end(), header.begin(), header.end());
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end()); write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
} }
...@@ -198,7 +219,7 @@ private: ...@@ -198,7 +219,7 @@ private:
return true; return true;
} }
transport_worker<application_type> worker_; worker_type worker_;
stream_socket handle_; stream_socket handle_;
std::vector<byte> read_buf_; std::vector<byte> read_buf_;
...@@ -212,6 +233,8 @@ private: ...@@ -212,6 +233,8 @@ private:
receive_policy_flag rd_flag_; receive_policy_flag rd_flag_;
size_t written_; size_t written_;
endpoint_manager* manager_;
}; };
} // namespace net } // namespace net
......
...@@ -19,28 +19,18 @@ ...@@ -19,28 +19,18 @@
#pragma once #pragma once
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/abstract_socket.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/network_socket.hpp" #include "caf/net/network_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/uri.hpp" #include "caf/uri.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
/// Represents a TCP acceptor in listening mode. /// Represents a TCP acceptor in listening mode.
struct tcp_accept_socket : abstract_socket<tcp_accept_socket> { struct tcp_accept_socket : network_socket {
using super = abstract_socket<tcp_accept_socket>; using super = network_socket;
using super::super; using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
constexpr operator network_socket() const noexcept {
return network_socket{id};
}
}; };
/// Creates a new TCP socket to accept connections on a given port. /// Creates a new TCP socket to accept connections on a given port.
......
...@@ -19,8 +19,6 @@ ...@@ -19,8 +19,6 @@
#pragma once #pragma once
#include "caf/ip_endpoint.hpp" #include "caf/ip_endpoint.hpp"
#include "caf/net/abstract_socket.hpp"
#include "caf/net/network_socket.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/uri.hpp" #include "caf/uri.hpp"
...@@ -29,22 +27,10 @@ namespace caf { ...@@ -29,22 +27,10 @@ namespace caf {
namespace net { namespace net {
/// Represents a TCP connection. /// Represents a TCP connection.
struct tcp_stream_socket : abstract_socket<tcp_stream_socket> { struct tcp_stream_socket : stream_socket {
using super = abstract_socket<tcp_stream_socket>; using super = stream_socket;
using super::super; using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
constexpr operator network_socket() const noexcept {
return network_socket{id};
}
constexpr operator stream_socket() const noexcept {
return stream_socket{id};
}
}; };
/// Creates a `tcp_stream_socket` connected to given remote node. /// Creates a `tcp_stream_socket` connected to given remote node.
......
...@@ -46,16 +46,32 @@ public: ...@@ -46,16 +46,32 @@ public:
// nop // nop
} }
// -- properties -------------------------------------------------------------
application_type& application() noexcept {
return application_;
}
const application_type& application() const noexcept {
return application_;
}
const id_type& id() const noexcept {
return id_;
}
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
return application_.init(parent); auto decorator = make_write_packet_decorator(*this, parent);
return application_.init(decorator);
} }
template <class Parent> template <class Parent>
void handle_data(Parent& parent, span<const byte> data) { void handle_data(Parent& parent, span<const byte> data) {
application_.handle_data(parent, data); auto decorator = make_write_packet_decorator(*this, parent);
application_.handle_data(decorator, data);
} }
template <class Parent> template <class Parent>
...@@ -65,20 +81,16 @@ public: ...@@ -65,20 +81,16 @@ public:
application_.write_message(decorator, std::move(msg)); application_.write_message(decorator, std::move(msg));
} }
template <class Parent>
void write_packet(Parent& parent, span<const byte> header,
span<const byte> payload) {
parent.write_packet(header, payload, id_);
}
template <class Parent> template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) { void resolve(Parent& parent, const std::string& path, actor listener) {
application_.resolve(parent, path, listener); auto decorator = make_write_packet_decorator(*this, parent);
application_.resolve(decorator, path, listener);
} }
template <class Parent> template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) { void timeout(Parent& parent, atom_value value, uint64_t id) {
application_.timeout(parent, value, id); auto decorator = make_write_packet_decorator(*this, parent);
application_.timeout(decorator, value, id);
} }
void handle_error(sec error) { void handle_error(sec error) {
......
...@@ -32,7 +32,7 @@ ...@@ -32,7 +32,7 @@
namespace caf { namespace caf {
namespace net { namespace net {
/// implements a dispatcher that dispatches calls to the right worker /// implements a dispatcher that dispatches between transport and workers.
template <class ApplicationFactory, class IdType> template <class ApplicationFactory, class IdType>
class transport_worker_dispatcher { class transport_worker_dispatcher {
public: public:
...@@ -57,16 +57,9 @@ public: ...@@ -57,16 +57,9 @@ public:
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
/// Initializes all contained contained workers.
/// @param The parent of this Dispatcher
/// @return error if something went wrong sec::none when init() worked.
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent&) {
for (const auto& p : workers_by_id_) { CAF_ASSERT(workers_by_id_.empty());
auto worker = p.second;
if (auto err = worker->init(parent))
return err;
}
return none; return none;
} }
...@@ -74,8 +67,8 @@ public: ...@@ -74,8 +67,8 @@ public:
void handle_data(Parent& parent, span<byte> data, id_type id) { void handle_data(Parent& parent, span<byte> data, id_type id) {
auto it = workers_by_id_.find(id); auto it = workers_by_id_.find(id);
if (it == workers_by_id_.end()) { if (it == workers_by_id_.end()) {
// TODO: where to get node_id from in this scope. // TODO: where to get node_id from here?
add_new_worker(node_id{}, id); add_new_worker(parent, node_id{}, id);
it = workers_by_id_.find(id); it = workers_by_id_.find(id);
} }
auto worker = it->second; auto worker = it->second;
...@@ -91,12 +84,11 @@ public: ...@@ -91,12 +84,11 @@ public:
auto nid = sender->node(); auto nid = sender->node();
auto it = workers_by_node_.find(nid); auto it = workers_by_node_.find(nid);
if (it == workers_by_node_.end()) { if (it == workers_by_node_.end()) {
// TODO: where to get id_type from in this scope. // TODO: where to get id_type from here?
add_new_worker(nid, id_type{}); add_new_worker(parent, nid, id_type{});
it = workers_by_node_.find(nid); it = workers_by_node_.find(nid);
} }
auto worker = it->second; auto worker = it->second;
// parent should be a decorator with parent and parent->parent.
worker->write_message(parent, std::move(msg)); worker->write_message(parent, std::move(msg));
} }
...@@ -108,7 +100,9 @@ public: ...@@ -108,7 +100,9 @@ public:
template <class Parent> template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) { void resolve(Parent& parent, const std::string& path, actor listener) {
// TODO: How to find the right worker? use path somehow? // TODO path should be uri to lookup the corresponding worker
// if enpoint is known -> resolve actor through worker
// if not connect to endpoint?!
if (workers_by_id_.empty()) if (workers_by_id_.empty())
return; return;
auto worker = workers_by_id_.begin()->second; auto worker = workers_by_id_.begin()->second;
...@@ -134,11 +128,15 @@ public: ...@@ -134,11 +128,15 @@ public:
} }
} }
void add_new_worker(node_id node, id_type id) { template <class Parent>
error add_new_worker(Parent& parent, node_id node, id_type id) {
auto application = factory_.make(); auto application = factory_.make();
auto worker = std::make_shared<worker_type>(std::move(application), id); auto worker = std::make_shared<worker_type>(std::move(application), id);
if (auto err = worker->init(parent))
return err;
workers_by_id_.emplace(std::move(id), worker); workers_by_id_.emplace(std::move(id), worker);
workers_by_node_.emplace(std::move(node), std::move(worker)); workers_by_node_.emplace(std::move(node), std::move(worker));
return none;
} }
private: private:
...@@ -148,9 +146,8 @@ private: ...@@ -148,9 +146,8 @@ private:
std::unordered_map<node_id, worker_ptr> workers_by_node_; std::unordered_map<node_id, worker_ptr> workers_by_node_;
std::unordered_map<uint64_t, worker_ptr> workers_by_timeout_id_; std::unordered_map<uint64_t, worker_ptr> workers_by_timeout_id_;
/// Factory to make application instances for transport_workers
factory_type factory_; factory_type factory_;
}; // namespace net };
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -19,56 +19,48 @@ ...@@ -19,56 +19,48 @@
#pragma once #pragma once
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/network_socket.hpp" #include "caf/net/network_socket.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
/// A non connection-oriented network communication endpoint for bidirectional /// A datagram-oriented network communication endpoint for bidirectional
/// byte transmission. /// byte transmission.
struct udp_datagram_socket : abstract_socket<udp_datagram_socket> { struct udp_datagram_socket : network_socket {
using super = abstract_socket<udp_datagram_socket>; using super = network_socket;
using super::super; using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
constexpr operator network_socket() const noexcept {
return network_socket{id};
}
}; };
/// Creates a `tcp_stream_socket` connected to given remote node. /// Creates a `udp_datagram_socket` bound to given port.
/// @param node Host and port of the remote node. /// @param node ip_endpoint that contains the port to bind to. Pass port '0' to
/// bind to any unused port - The endpoint will be updated with the specific
/// port that was bound.
/// @returns The connected socket or an error. /// @returns The connected socket or an error.
/// @relates tcp_stream_socket /// @relates udp_datagram_socket
expected<udp_datagram_socket> make_udp_datagram_socket(ip_endpoint& node, expected<std::pair<udp_datagram_socket, uint16_t>>
bool reuse_addr = false); make_udp_datagram_socket(ip_endpoint ep, bool reuse_addr = false);
/// Enables or disables `SIO_UDP_CONNRESET` error on `x`. /// Enables or disables `SIO_UDP_CONNRESET` error on `x`.
/// @relates udp_datagram_socket /// @relates udp_datagram_socket
error allow_connreset(udp_datagram_socket x, bool new_value); error allow_connreset(udp_datagram_socket x, bool new_value);
/// Receives data from `x`. /// Receives the next datagram on socket `x`.
/// @param x udp_datagram_socket. /// @param x The UDP socket for receiving datagrams.
/// @param buf Points to destination buffer. /// @param buf Writable output buffer.
/// @returns The number of received bytes and the ip_endpoint on success, an /// @returns The number of received bytes and the sender as `ip_endpoint` on
/// error code otherwise. /// success, an error code otherwise.
/// @relates udp_datagram_socket /// @relates udp_datagram_socket
/// @post either the result is a `sec` or a pair of positive (non-zero) integer /// @post buf was modified and the resulting integer represents the length of
/// and ip_endpoint /// the received datagram, even if it did not fit into the given buffer.
variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x, variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
span<byte> buf); span<byte> buf);
/// Transmits data from `x` to its peer. /// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`.
/// @param x udp_datagram_socket. /// @param x The UDP socket for sending datagrams.
/// @param buf Points to the message to send. /// @param buf The buffer to send.
/// @returns The number of written bytes on success, otherwise an error code. /// @returns The number of written bytes on success, otherwise an error code.
/// @relates udp_datagram_socket /// @relates udp_datagram_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf, variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
ip_endpoint ep); ip_endpoint ep);
......
...@@ -52,12 +52,17 @@ public: ...@@ -52,12 +52,17 @@ public:
return parent_.transport(); return parent_.transport();
} }
endpoint_manager& manager() {
return parent_.manager();
}
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class... Ts> template <class... Ts>
void write_packet(span<const byte> header, span<const byte> payload, void write_packet(span<const byte> header, span<const byte> payload,
Ts&&... xs) { Ts&&... xs) {
object_.write_packet(parent_, header, payload, std::forward<Ts>(xs)...); parent_.write_packet(header, payload, std::forward<Ts>(xs)...,
object_.id());
} }
void cancel_timeout(atom_value type, uint64_t id) { void cancel_timeout(atom_value type, uint64_t id) {
......
...@@ -27,18 +27,18 @@ namespace net { ...@@ -27,18 +27,18 @@ namespace net {
actor_proxy_impl::actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst) actor_proxy_impl::actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst)
: super(cfg), sf_(dst->serialize_fun()), dst_(std::move(dst)) { : super(cfg), sf_(dst->serialize_fun()), dst_(std::move(dst)) {
// anon_send(broker_, monitor_atom::value, ctrl()); // nop
} }
actor_proxy_impl::~actor_proxy_impl() { actor_proxy_impl::~actor_proxy_impl() {
// anon_send(broker_, make_message(delete_atom::value, node(), id())); // nop
} }
void actor_proxy_impl::enqueue(mailbox_element_ptr what, execution_unit*) { void actor_proxy_impl::enqueue(mailbox_element_ptr what, execution_unit*) {
CAF_PUSH_AID(0); CAF_PUSH_AID(0);
CAF_ASSERT(what != nullptr); CAF_ASSERT(what != nullptr);
if (auto payload = sf_(home_system(), what->content())) if (auto payload = sf_(home_system(), what->content()))
dst_->enqueue(std::move(what), std::move(*payload)); dst_->enqueue(std::move(what), ctrl(), std::move(*payload));
else else
CAF_LOG_ERROR( CAF_LOG_ERROR(
"unable to serialize payload: " << home_system().render(payload.error())); "unable to serialize payload: " << home_system().render(payload.error()));
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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/net/basp/application.hpp"
#include <vector>
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/byte.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/detail/parse.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/no_stages.hpp"
#include "caf/none.hpp"
#include "caf/sec.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/type_erased_tuple.hpp"
namespace caf {
namespace net {
namespace basp {
application::application(proxy_registry_ptr proxies)
: proxies_(std::move(proxies)) {
// nop
}
expected<std::vector<byte>> application::serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<byte> result;
serializer_impl<std::vector<byte>> sink{sys, result};
if (auto err = message::save(sink, x))
return err;
return result;
}
strong_actor_ptr application::resolve_local_path(string_view path) {
// We currently support two path formats: `id/<actor_id>` and `name/<atom>`.
static constexpr string_view id_prefix = "id/";
if (starts_with(path, id_prefix)) {
path.remove_prefix(id_prefix.size());
actor_id aid;
if (auto err = detail::parse(path, aid))
return nullptr;
return system().registry().get(aid);
}
static constexpr string_view name_prefix = "name/";
if (starts_with(path, name_prefix)) {
path.remove_prefix(name_prefix.size());
atom_value name;
if (auto err = detail::parse(path, name))
return nullptr;
return system().registry().get(name);
}
return nullptr;
}
void application::resolve_remote_path(write_packet_callback& write_packet,
string_view path, actor listener) {
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
if (auto err = sink(path)) {
CAF_LOG_ERROR("unable to serialize path");
return;
}
auto req_id = next_request_id_++;
auto hdr = to_bytes(header{message_type::resolve_request,
static_cast<uint32_t>(buf_.size()), req_id});
if (auto err = write_packet(hdr, buf_)) {
CAF_LOG_ERROR("unable to write resolve_request header");
return;
}
response_promise rp{nullptr, actor_cast<strong_actor_ptr>(listener),
no_stages, make_message_id()};
pending_resolves_.emplace(req_id, std::move(rp));
}
error application::handle(size_t& next_read_size,
write_packet_callback& write_packet,
byte_span bytes) {
switch (state_) {
case connection_state::await_handshake_header: {
if (bytes.size() != header_size)
return ec::unexpected_number_of_bytes;
hdr_ = header::from_bytes(bytes);
if (hdr_.type != message_type::handshake)
return ec::missing_handshake;
if (hdr_.operation_data != version)
return ec::version_mismatch;
if (hdr_.payload_len == 0)
return ec::missing_payload;
state_ = connection_state::await_handshake_payload;
next_read_size = hdr_.payload_len;
return none;
}
case connection_state::await_handshake_payload: {
if (auto err = handle_handshake(write_packet, hdr_, bytes))
return err;
state_ = connection_state::await_header;
return none;
}
case connection_state::await_header: {
if (bytes.size() != header_size)
return ec::unexpected_number_of_bytes;
hdr_ = header::from_bytes(bytes);
if (hdr_.payload_len == 0)
return handle(write_packet, hdr_, byte_span{});
next_read_size = hdr_.payload_len;
state_ = connection_state::await_payload;
return none;
}
case connection_state::await_payload: {
if (bytes.size() != hdr_.payload_len)
return ec::unexpected_number_of_bytes;
state_ = connection_state::await_header;
return handle(write_packet, hdr_, bytes);
}
default:
return ec::illegal_state;
}
}
error application::handle(write_packet_callback& write_packet, header hdr,
byte_span payload) {
switch (hdr.type) {
case message_type::handshake:
return ec::unexpected_handshake;
case message_type::actor_message:
return handle_actor_message(write_packet, hdr, payload);
case message_type::resolve_request:
return handle_resolve_request(write_packet, hdr, payload);
case message_type::resolve_response:
return handle_resolve_response(write_packet, hdr, payload);
case message_type::heartbeat:
return none;
default:
return ec::unimplemented;
}
}
error application::handle_handshake(write_packet_callback&, header hdr,
byte_span payload) {
if (hdr.type != message_type::handshake)
return ec::missing_handshake;
if (hdr.operation_data != version)
return ec::version_mismatch;
node_id peer_id;
std::vector<std::string> app_ids;
binary_deserializer source{system(), payload};
if (auto err = source(peer_id, app_ids))
return err;
if (!peer_id || app_ids.empty())
return ec::invalid_handshake;
auto ids = get_or(system().config(), "middleman.app-identifiers",
defaults::middleman::app_identifiers);
auto predicate = [=](const std::string& x) {
return std::find(ids.begin(), ids.end(), x) != ids.end();
};
if (std::none_of(app_ids.begin(), app_ids.end(), predicate))
return ec::app_identifiers_mismatch;
peer_id_ = std::move(peer_id);
state_ = connection_state::await_header;
return none;
}
error application::handle_actor_message(write_packet_callback&, header hdr,
byte_span payload) {
// Deserialize payload.
actor_id src_id;
node_id src_node;
actor_id dst_id;
std::vector<strong_actor_ptr> fwd_stack;
message content;
binary_deserializer source{system(), payload};
if (auto err = source(src_node, src_id, dst_id, fwd_stack, content))
return err;
// Sanity checks.
if (dst_id == 0)
return ec::invalid_payload;
// Try to fetch the receiver.
auto dst_hdl = system().registry().get(dst_id);
if (dst_hdl == nullptr) {
CAF_LOG_DEBUG("no actor found for given ID, drop message");
return caf::none;
}
// Try to fetch the sender.
strong_actor_ptr src_hdl;
if (src_node != none && src_id != 0)
src_hdl = proxies_->get_or_put(src_node, src_id);
// Ship the message.
auto ptr = make_mailbox_element(std::move(src_hdl),
make_message_id(hdr.operation_data),
std::move(fwd_stack), std::move(content));
dst_hdl->get()->enqueue(std::move(ptr), nullptr);
return none;
}
error application::handle_resolve_request(write_packet_callback& write_packet,
header hdr, byte_span payload) {
CAF_ASSERT(hdr.type == message_type::resolve_request);
size_t path_size = 0;
binary_deserializer source{system(), payload};
if (auto err = source.begin_sequence(path_size))
return err;
// We expect the payload to consist only of the path.
if (path_size != source.remaining())
return ec::invalid_payload;
auto remainder = source.remainder();
string_view path{reinterpret_cast<const char*>(remainder.data()),
remainder.size()};
// Write result.
auto result = resolve_local_path(path);
buf_.clear();
actor_id aid = result ? result->id() : 0;
std::set<std::string> ifs;
// TODO: figure out how to obtain messaging interface.
serializer_impl<buffer_type> sink{system(), buf_};
if (auto err = sink(aid, ifs))
return err;
auto out_hdr = to_bytes(header{message_type::resolve_response,
static_cast<uint32_t>(buf_.size()),
hdr.operation_data});
return write_packet(out_hdr, buf_);
}
error application::handle_resolve_response(write_packet_callback&, header hdr,
byte_span payload) {
CAF_ASSERT(hdr.type == message_type::resolve_response);
auto i = pending_resolves_.find(hdr.operation_data);
if (i == pending_resolves_.end()) {
CAF_LOG_ERROR("received unknown ID in resolve_response message");
return none;
}
auto guard = detail::make_scope_guard([&] {
if (i->second.pending())
i->second.deliver(sec::remote_lookup_failed);
pending_resolves_.erase(i);
});
actor_id aid;
std::set<std::string> ifs;
binary_deserializer source{system(), payload};
if (auto err = source(aid, ifs))
return err;
if (aid == 0) {
i->second.deliver(strong_actor_ptr{nullptr}, std::move(ifs));
return none;
}
i->second.deliver(proxies_->get_or_put(peer_id_, aid), std::move(ifs));
return none;
}
error application::generate_handshake() {
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
return sink(system().node(),
get_or(system().config(), "middleman.app-identifiers",
defaults::middleman::app_identifiers));
}
} // namespace basp
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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/net/basp/connection_state.hpp"
namespace caf {
namespace net {
namespace basp {
namespace {
const char* connection_state_names[] = {
"await_handshake_header",
"await_handshake_payload",
"await_header",
"await_payload",
"shutdown",
};
} // namespace
std::string to_string(connection_state x) {
return connection_state_names[static_cast<uint8_t>(x)];
}
} // namespace basp
} // namespace net
} // namespace caf
...@@ -17,43 +17,46 @@ ...@@ -17,43 +17,46 @@
******************************************************************************/ ******************************************************************************/
#include "caf/detail/convert_ip_endpoint.hpp" #include "caf/detail/convert_ip_endpoint.hpp"
#include "caf/error.hpp"
#include "caf/ipv4_endpoint.hpp" #include "caf/ipv4_endpoint.hpp"
#include "caf/ipv6_endpoint.hpp" #include "caf/ipv6_endpoint.hpp"
#include <iostream> #include "caf/sec.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
sockaddr_storage to_sockaddr(const ip_endpoint& ep) { void convert(const ip_endpoint& src, sockaddr_storage& dst) {
sockaddr_storage sockaddr = {}; if (src.address().embeds_v4()) {
if (ep.address().embeds_v4()) { auto sockaddr4 = reinterpret_cast<sockaddr_in*>(&dst);
auto sockaddr4 = reinterpret_cast<sockaddr_in*>(&sockaddr);
sockaddr4->sin_family = AF_INET; sockaddr4->sin_family = AF_INET;
sockaddr4->sin_port = ntohs(ep.port()); sockaddr4->sin_port = ntohs(src.port());
sockaddr4->sin_addr.s_addr = ep.address().embedded_v4().bits(); sockaddr4->sin_addr.s_addr = src.address().embedded_v4().bits();
} else { } else {
auto sockaddr6 = reinterpret_cast<sockaddr_in6*>(&sockaddr); auto sockaddr6 = reinterpret_cast<sockaddr_in6*>(&dst);
sockaddr6->sin6_family = AF_INET6; sockaddr6->sin6_family = AF_INET6;
sockaddr6->sin6_port = ntohs(ep.port()); sockaddr6->sin6_port = ntohs(src.port());
memcpy(&sockaddr6->sin6_addr, ep.address().bytes().data(), memcpy(&sockaddr6->sin6_addr, src.address().bytes().data(),
ep.address().bytes().size()); src.address().bytes().size());
} }
return sockaddr;
} }
ip_endpoint to_ip_endpoint(const sockaddr_storage& addr) { error convert(const sockaddr_storage& src, ip_endpoint& dst) {
if (addr.ss_family == AF_INET) { if (src.ss_family == AF_INET) {
auto sockaddr4 = reinterpret_cast<const sockaddr_in*>(&addr); auto sockaddr4 = reinterpret_cast<const sockaddr_in&>(src);
ipv4_address ipv4_addr; ipv4_address ipv4_addr;
memcpy(ipv4_addr.data().data(), &sockaddr4->sin_addr, ipv4_addr.size()); memcpy(ipv4_addr.data().data(), &sockaddr4.sin_addr, ipv4_addr.size());
return {ipv4_addr, htons(sockaddr4->sin_port)}; dst = ip_endpoint{ipv4_addr, htons(sockaddr4.sin_port)};
} else { } else if (src.ss_family == AF_INET6) {
auto sockaddr6 = reinterpret_cast<const sockaddr_in6*>(&addr); auto sockaddr6 = reinterpret_cast<const sockaddr_in6&>(src);
ipv6_address ipv6_addr; ipv6_address ipv6_addr;
memcpy(ipv6_addr.bytes().data(), &sockaddr6->sin6_addr, memcpy(ipv6_addr.bytes().data(), &sockaddr6.sin6_addr,
ipv6_addr.bytes().size()); ipv6_addr.bytes().size());
return {ipv6_addr, htons(sockaddr6->sin6_port)}; dst = ip_endpoint{ipv6_addr, htons(sockaddr6.sin6_port)};
} else {
return sec::invalid_argument;
} }
return none;
} }
} // namespace detail } // namespace detail
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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/net/basp/ec.hpp"
#include "caf/atom.hpp"
#include "caf/error.hpp"
#include "caf/string_view.hpp"
namespace caf {
namespace net {
namespace basp {
namespace {
string_view ec_names[] = {
"none",
"invalid_magic_number",
"unexpected_number_of_bytes",
"unexpected_payload",
"missing_payload",
"illegal_state",
"invalid_handshake",
"missing_handshake",
"unexpected_handshake",
"version_mismatch",
"unimplemented",
"app_identifiers_mismatch",
"invalid_payload",
};
} // namespace
std::string to_string(ec x) {
auto result = ec_names[static_cast<uint8_t>(x)];
return std::string{result.begin(), result.end()};
}
error make_error(ec x) {
return {static_cast<uint8_t>(x), atom("basp")};
}
} // namespace basp
} // namespace net
} // namespace caf
...@@ -37,8 +37,11 @@ endpoint_manager::event::event(atom_value type, uint64_t id) ...@@ -37,8 +37,11 @@ endpoint_manager::event::event(atom_value type, uint64_t id)
} }
endpoint_manager::message::message(mailbox_element_ptr msg, endpoint_manager::message::message(mailbox_element_ptr msg,
strong_actor_ptr receiver,
std::vector<byte> payload) std::vector<byte> payload)
: msg(std::move(msg)), payload(std::move(payload)) { : msg(std::move(msg)),
receiver(std::move(receiver)),
payload(std::move(payload)) {
// nop // nop
} }
...@@ -87,8 +90,10 @@ void endpoint_manager::resolve(std::string path, actor listener) { ...@@ -87,8 +90,10 @@ void endpoint_manager::resolve(std::string path, actor listener) {
} }
void endpoint_manager::enqueue(mailbox_element_ptr msg, void endpoint_manager::enqueue(mailbox_element_ptr msg,
strong_actor_ptr receiver,
std::vector<byte> payload) { std::vector<byte> payload) {
auto ptr = new message(std::move(msg), std::move(payload)); auto ptr = new message(std::move(msg), std::move(receiver),
std::move(payload));
if (messages_.push_back(ptr) == intrusive::inbox_result::unblocked_reader) if (messages_.push_back(ptr) == intrusive::inbox_result::unblocked_reader)
mask_add(operation::write); mask_add(operation::write);
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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/net/basp/header.hpp"
#include <cstring>
#include "caf/byte.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/span.hpp"
namespace caf {
namespace net {
namespace basp {
int header::compare(header other) const noexcept {
auto x = to_bytes(*this);
auto y = to_bytes(other);
return memcmp(x.data(), y.data(), header_size);
}
header header::from_bytes(span<const byte> bytes) {
CAF_ASSERT(bytes.size() >= header_size);
header result;
auto ptr = bytes.data();
result.type = *reinterpret_cast<const message_type*>(ptr);
auto payload_len = *reinterpret_cast<const uint32_t*>(ptr + 1);
result.payload_len = detail::from_network_order(payload_len);
auto operation_data = *reinterpret_cast<const uint64_t*>(ptr + 5);
result.operation_data = detail::from_network_order(operation_data);
return result;
}
std::array<byte, header_size> to_bytes(header x) {
std::array<byte, header_size> result;
auto ptr = result.data();
*ptr = static_cast<byte>(x.type);
auto payload_len = detail::to_network_order(x.payload_len);
memcpy(ptr + 1, &payload_len, sizeof(payload_len));
auto operation_data = detail::to_network_order(x.operation_data);
memcpy(ptr + 5, &operation_data, sizeof(operation_data));
return result;
}
} // namespace basp
} // namespace net
} // namespace caf
...@@ -55,6 +55,9 @@ namespace ip { ...@@ -55,6 +55,9 @@ namespace ip {
namespace { namespace {
// Dummy port to resolve empty string with getaddrinfo.
constexpr auto dummy_port = "42";
void* fetch_in_addr(int family, sockaddr* addr) { void* fetch_in_addr(int family, sockaddr* addr) {
if (family == AF_INET) if (family == AF_INET)
return &reinterpret_cast<sockaddr_in*>(addr)->sin_addr; return &reinterpret_cast<sockaddr_in*>(addr)->sin_addr;
...@@ -83,7 +86,9 @@ std::vector<ip_address> resolve(string_view host) { ...@@ -83,7 +86,9 @@ std::vector<ip_address> resolve(string_view host) {
hint.ai_flags = AI_PASSIVE; hint.ai_flags = AI_PASSIVE;
addrinfo* tmp = nullptr; addrinfo* tmp = nullptr;
std::string host_str{host.begin(), host.end()}; std::string host_str{host.begin(), host.end()};
if (getaddrinfo(host_str.c_str(), nullptr, &hint, &tmp) != 0) if (getaddrinfo(host.empty() ? nullptr : host_str.c_str(),
host.empty() ? dummy_port : nullptr, &hint, &tmp)
!= 0)
return {}; return {};
std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo}; std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo};
char buffer[INET6_ADDRSTRLEN]; char buffer[INET6_ADDRSTRLEN];
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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/net/basp/message_type.hpp"
#include "caf/string_view.hpp"
namespace caf {
namespace net {
namespace basp {
namespace {
string_view message_type_names[] = {
"handshake", "actor_message", "resolve_request", "resolve_response",
"monitor_message", "down_message", "heartbeat",
};
} // namespace
std::string to_string(message_type x) {
auto result = message_type_names[static_cast<uint8_t>(x)];
return std::string{result.begin(), result.end()};
}
} // namespace basp
} // namespace net
} // namespace caf
...@@ -115,7 +115,7 @@ expected<std::pair<stream_socket, stream_socket>> make_stream_socket_pair() { ...@@ -115,7 +115,7 @@ expected<std::pair<stream_socket, stream_socket>> make_stream_socket_pair() {
accept(listener, nullptr, nullptr)); accept(listener, nullptr, nullptr));
close(socket{listener}); close(socket{listener});
guard.disable(); guard.disable();
return std::make_pair(read_fd, write_fd); return std::make_pair(stream_socket{read_fd}, stream_socket{write_fd});
} }
error keepalive(stream_socket x, bool new_value) { error keepalive(stream_socket x, bool new_value) {
......
...@@ -63,9 +63,10 @@ expected<tcp_accept_socket> new_tcp_acceptor_impl(uint16_t port, ...@@ -63,9 +63,10 @@ expected<tcp_accept_socket> new_tcp_acceptor_impl(uint16_t port,
socktype |= SOCK_CLOEXEC; socktype |= SOCK_CLOEXEC;
#endif #endif
CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(Family, socktype, 0)); CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(Family, socktype, 0));
child_process_inherit(fd, false); tcp_accept_socket sock{fd};
// sguard closes the socket in case of exception // sguard closes the socket in case of exception
auto sguard = make_socket_guard(tcp_accept_socket{fd}); auto sguard = make_socket_guard(tcp_accept_socket{fd});
child_process_inherit(sock, false);
if (reuse_addr) { if (reuse_addr) {
int on = 1; int on = 1;
CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0, CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0,
...@@ -79,7 +80,7 @@ expected<tcp_accept_socket> new_tcp_acceptor_impl(uint16_t port, ...@@ -79,7 +80,7 @@ expected<tcp_accept_socket> new_tcp_acceptor_impl(uint16_t port,
memset(&sa, 0, sizeof(sockaddr_type)); memset(&sa, 0, sizeof(sockaddr_type));
detail::family_of(sa) = Family; detail::family_of(sa) = Family;
if (any) if (any)
if (auto err = set_inaddr_any(fd, sa)) if (auto err = set_inaddr_any(sock, sa))
return err; return err;
CAF_NET_SYSCALL("inet_pton", tmp, !=, 1, CAF_NET_SYSCALL("inet_pton", tmp, !=, 1,
inet_pton(Family, addr, &detail::addr_of(sa))); inet_pton(Family, addr, &detail::addr_of(sa)));
...@@ -141,7 +142,7 @@ expected<tcp_stream_socket> accept(tcp_accept_socket x) { ...@@ -141,7 +142,7 @@ expected<tcp_stream_socket> accept(tcp_accept_socket x) {
} }
return caf::make_error(sec::socket_operation_failed, "tcp accept failed"); return caf::make_error(sec::socket_operation_failed, "tcp accept failed");
} }
return {sock}; return tcp_stream_socket{sock};
} }
} // namespace net } // namespace net
......
...@@ -60,14 +60,15 @@ expected<tcp_stream_socket> make_connected_tcp_stream_socket(ip_endpoint node) { ...@@ -60,14 +60,15 @@ expected<tcp_stream_socket> make_connected_tcp_stream_socket(ip_endpoint node) {
socktype |= SOCK_CLOEXEC; socktype |= SOCK_CLOEXEC;
#endif #endif
CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(proto, socktype, 0)); CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(proto, socktype, 0));
child_process_inherit(fd, false); tcp_stream_socket sock{fd};
auto sguard = make_socket_guard(tcp_stream_socket{fd}); child_process_inherit(sock, false);
auto sguard = make_socket_guard(sock);
if (proto == AF_INET6) { if (proto == AF_INET6) {
if (ip_connect<AF_INET6>(fd, to_string(node.address()), node.port())) { if (ip_connect<AF_INET6>(sock, to_string(node.address()), node.port())) {
CAF_LOG_INFO("successfully connected to (IPv6):" << to_string(node)); CAF_LOG_INFO("successfully connected to (IPv6):" << to_string(node));
return sguard.release(); return sguard.release();
} }
} else if (ip_connect<AF_INET>(fd, to_string(node.address().embedded_v4()), } else if (ip_connect<AF_INET>(sock, to_string(node.address().embedded_v4()),
node.port())) { node.port())) {
CAF_LOG_INFO("successfully connected to (IPv4):" << to_string(node)); CAF_LOG_INFO("successfully connected to (IPv4):" << to_string(node));
return sguard.release(); return sguard.release();
......
...@@ -55,28 +55,34 @@ error allow_connreset(udp_datagram_socket x, bool) { ...@@ -55,28 +55,34 @@ error allow_connreset(udp_datagram_socket x, bool) {
#endif // CAF_WINDOWS #endif // CAF_WINDOWS
expected<udp_datagram_socket> make_udp_datagram_socket(ip_endpoint& node, expected<std::pair<udp_datagram_socket, uint16_t>>
bool reuse_addr) { make_udp_datagram_socket(ip_endpoint ep, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(node)); CAF_LOG_TRACE(CAF_ARG(ep));
auto addr = to_sockaddr(node); sockaddr_storage addr;
detail::convert(ep, addr);
CAF_NET_SYSCALL("socket", fd, ==, invalid_socket_id, CAF_NET_SYSCALL("socket", fd, ==, invalid_socket_id,
::socket(addr.ss_family, SOCK_DGRAM, 0)); ::socket(addr.ss_family, SOCK_DGRAM, 0));
udp_datagram_socket sock{fd}; udp_datagram_socket sock{fd};
auto sguard = make_socket_guard(sock); auto sguard = make_socket_guard(sock);
socklen_t len = (addr.ss_family == AF_INET) ? sizeof(sockaddr_in) socklen_t len = (addr.ss_family == AF_INET) ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6); : sizeof(sockaddr_in6);
if (reuse_addr) {
int on = 1;
CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0,
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<setsockopt_ptr>(&on),
static_cast<socket_size_type>(sizeof(on))));
}
CAF_NET_SYSCALL("bind", err, !=, 0, CAF_NET_SYSCALL("bind", err, !=, 0,
::bind(sock.id, reinterpret_cast<sockaddr*>(&addr), len)); ::bind(sock.id, reinterpret_cast<sockaddr*>(&addr), len));
CAF_NET_SYSCALL("getsockname", erro, !=, 0, CAF_NET_SYSCALL("getsockname", erro, !=, 0,
getsockname(sock.id, reinterpret_cast<sockaddr*>(&addr), getsockname(sock.id, reinterpret_cast<sockaddr*>(&addr),
&len)); &len));
CAF_LOG_DEBUG(CAF_ARG(sock.id)); CAF_LOG_DEBUG(CAF_ARG(sock.id));
auto port = addr.ss_family == AF_INET auto port = addr.ss_family == AF_INET
? reinterpret_cast<sockaddr_in*>(&addr)->sin_port ? reinterpret_cast<sockaddr_in*>(&addr)->sin_port
: reinterpret_cast<sockaddr_in6*>(&addr)->sin6_port; : reinterpret_cast<sockaddr_in6*>(&addr)->sin6_port;
node.port(ntohs(port)); return std::make_pair(sguard.release(), port);
return sguard.release();
} }
variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x, variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
...@@ -88,13 +94,15 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x, ...@@ -88,13 +94,15 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
&len); &len);
auto ret = check_udp_datagram_socket_io_res(res); auto ret = check_udp_datagram_socket_io_res(res);
if (auto num_bytes = get_if<size_t>(&ret)) { if (auto num_bytes = get_if<size_t>(&ret)) {
if (*num_bytes == 0) CAF_LOG_INFO_IF(*num_bytes == 0, "Received empty datagram");
CAF_LOG_INFO("Received empty datagram"); CAF_LOG_WARNING_IF(*num_bytes > buf.size(),
else if (*num_bytes > buf.size()) "recvfrom cut of message, only received "
CAF_LOG_WARNING("recvfrom cut of message, only received "
<< CAF_ARG(buf.size()) << " of " << CAF_ARG(num_bytes) << CAF_ARG(buf.size()) << " of " << CAF_ARG(num_bytes)
<< " bytes"); << " bytes");
auto ep = detail::to_ip_endpoint(addr); ip_endpoint ep;
// TODO: how to properly pass error further?
if (auto err = detail::convert(addr, ep))
return sec::runtime_error;
return std::pair<size_t, ip_endpoint>(*num_bytes, ep); return std::pair<size_t, ip_endpoint>(*num_bytes, ep);
} else { } else {
return get<sec>(ret); return get<sec>(ret);
...@@ -103,7 +111,8 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x, ...@@ -103,7 +111,8 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf, variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
ip_endpoint ep) { ip_endpoint ep) {
auto addr = detail::to_sockaddr(ep); sockaddr_storage addr;
detail::convert(ep, addr);
auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), buf.size(), 0, reinterpret_cast<sockaddr*>(&addr),
ep.address().embeds_v4() ? sizeof(sockaddr_in) ep.address().embeds_v4() ? sizeof(sockaddr_in)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#define CAF_SUITE basp.application
#include "caf/net/basp/application.hpp"
#include "caf/test/dsl.hpp"
#include <vector>
#include "caf/byte.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/none.hpp"
#include "caf/uri.hpp"
using namespace caf;
using namespace caf::net;
#define REQUIRE_OK(statement) \
if (auto err = statement) \
CAF_FAIL("failed to serialize data: " << sys.render(err));
namespace {
struct fixture : test_coordinator_fixture<>, proxy_registry::backend {
using buffer_type = std::vector<byte>;
fixture() : app(std::make_shared<proxy_registry>(sys, *this)) {
REQUIRE_OK(app.init(*this));
uri mars_uri;
REQUIRE_OK(parse("tcp://mars", mars_uri));
mars = make_node_id(mars_uri);
}
template <class... Ts>
buffer_type to_buf(const Ts&... xs) {
buffer_type buf;
serializer_impl<buffer_type> sink{system(), buf};
REQUIRE_OK(sink(xs...));
return buf;
}
template <class... Ts>
void set_input(const Ts&... xs) {
input = to_buf(xs...);
}
void write_packet(span<const byte> hdr, span<const byte> payload) {
output.insert(output.end(), hdr.begin(), hdr.end());
output.insert(output.end(), payload.begin(), payload.end());
}
void handle_handshake() {
CAF_CHECK_EQUAL(app.state(),
basp::connection_state::await_handshake_header);
auto payload = to_buf(mars, defaults::middleman::app_identifiers);
set_input(basp::header{basp::message_type::handshake,
static_cast<uint32_t>(payload.size()),
basp::version});
REQUIRE_OK(app.handle_data(*this, input));
CAF_CHECK_EQUAL(app.state(),
basp::connection_state::await_handshake_payload);
REQUIRE_OK(app.handle_data(*this, payload));
}
void consume_handshake() {
if (output.size() < basp::header_size)
CAF_FAIL("BASP application did not write a handshake header");
auto hdr = basp::header::from_bytes(output);
if (hdr.type != basp::message_type::handshake || hdr.payload_len == 0
|| hdr.operation_data != basp::version)
CAF_FAIL("invalid handshake header");
node_id nid;
std::vector<std::string> app_ids;
binary_deserializer source{sys, output};
source.skip(basp::header_size);
if (auto err = source(nid, app_ids))
CAF_FAIL("unable to deserialize payload: " << sys.render(err));
if (source.remaining() > 0)
CAF_FAIL("trailing bytes after reading payload");
output.clear();
}
actor_system& system() {
return sys;
}
fixture& transport() {
return *this;
}
template <class... Ts>
void configure_read(Ts...) {
// nop
}
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override {
using impl_type = forwarding_actor_proxy;
using hdl_type = strong_actor_ptr;
actor_config cfg;
return make_actor<impl_type, hdl_type>(aid, nid, &sys, cfg, self);
}
void set_last_hop(node_id*) override {
// nop
}
buffer_type input;
buffer_type output;
node_id mars;
basp::application app;
};
} // namespace
#define MOCK(kind, op, ...) \
do { \
auto payload = to_buf(__VA_ARGS__); \
set_input(basp::header{kind, static_cast<uint32_t>(payload.size()), op}); \
if (auto err = app.handle_data(*this, input)) \
CAF_FAIL("application-under-test failed to process header: " \
<< sys.render(err)); \
if (auto err = app.handle_data(*this, payload)) \
CAF_FAIL("application-under-test failed to process payload: " \
<< sys.render(err)); \
} while (false)
#define RECEIVE(msg_type, op_data, ...) \
do { \
binary_deserializer source{sys, output}; \
basp::header hdr; \
if (auto err = source(hdr, __VA_ARGS__)) \
CAF_FAIL("failed to receive data: " << sys.render(err)); \
if (source.remaining() != 0) \
CAF_FAIL("unable to read entire message, " << source.remaining() \
<< " bytes left in buffer"); \
CAF_CHECK_EQUAL(hdr.type, msg_type); \
CAF_CHECK_EQUAL(hdr.operation_data, op_data); \
output.clear(); \
} while (false)
CAF_TEST_FIXTURE_SCOPE(application_tests, fixture)
CAF_TEST(missing handshake) {
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_header);
set_input(basp::header{basp::message_type::heartbeat, 0, 0});
CAF_CHECK_EQUAL(app.handle_data(*this, input), basp::ec::missing_handshake);
}
CAF_TEST(version mismatch) {
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_header);
set_input(basp::header{basp::message_type::handshake, 0, 0});
CAF_CHECK_EQUAL(app.handle_data(*this, input), basp::ec::version_mismatch);
}
CAF_TEST(missing payload in handshake) {
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_header);
set_input(basp::header{basp::message_type::handshake, 0, basp::version});
CAF_CHECK_EQUAL(app.handle_data(*this, input), basp::ec::missing_payload);
}
CAF_TEST(invalid handshake) {
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_header);
node_id no_nid;
std::vector<std::string> no_ids;
auto payload = to_buf(no_nid, no_ids);
set_input(basp::header{basp::message_type::handshake,
static_cast<uint32_t>(payload.size()), basp::version});
REQUIRE_OK(app.handle_data(*this, input));
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_payload);
CAF_CHECK_EQUAL(app.handle_data(*this, payload), basp::ec::invalid_handshake);
}
CAF_TEST(app identifier mismatch) {
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_header);
std::vector<std::string> wrong_ids{"YOLO!!!"};
auto payload = to_buf(mars, wrong_ids);
set_input(basp::header{basp::message_type::handshake,
static_cast<uint32_t>(payload.size()), basp::version});
REQUIRE_OK(app.handle_data(*this, input));
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_payload);
CAF_CHECK_EQUAL(app.handle_data(*this, payload),
basp::ec::app_identifiers_mismatch);
}
CAF_TEST(repeated handshake) {
handle_handshake();
consume_handshake();
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
node_id no_nid;
std::vector<std::string> no_ids;
auto payload = to_buf(no_nid, no_ids);
set_input(basp::header{basp::message_type::handshake,
static_cast<uint32_t>(payload.size()), basp::version});
CAF_CHECK_EQUAL(app.handle_data(*this, input), none);
CAF_CHECK_EQUAL(app.handle_data(*this, payload),
basp::ec::unexpected_handshake);
}
CAF_TEST(actor message) {
handle_handshake();
consume_handshake();
sys.registry().put(self->id(), self);
CAF_REQUIRE_EQUAL(self->mailbox().size(), 0u);
MOCK(basp::message_type::actor_message, make_message_id().integer_value(),
mars, actor_id{42}, self->id(), std::vector<strong_actor_ptr>{},
make_message("hello world!"));
allow((atom_value, strong_actor_ptr),
from(_).to(self).with(atom("monitor"), _));
expect((std::string), from(_).to(self).with("hello world!"));
}
CAF_TEST(resolve request without result) {
handle_handshake();
consume_handshake();
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
MOCK(basp::message_type::resolve_request, 42, std::string{"foo/bar"});
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
actor_id aid;
std::set<std::string> ifs;
RECEIVE(basp::message_type::resolve_response, 42u, aid, ifs);
CAF_CHECK_EQUAL(aid, 0u);
CAF_CHECK(ifs.empty());
}
CAF_TEST(resolve request on id with result) {
handle_handshake();
consume_handshake();
sys.registry().put(self->id(), self);
auto path = "id/" + std::to_string(self->id());
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
MOCK(basp::message_type::resolve_request, 42, path);
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
actor_id aid;
std::set<std::string> ifs;
RECEIVE(basp::message_type::resolve_response, 42u, aid, ifs);
CAF_CHECK_EQUAL(aid, self->id());
CAF_CHECK(ifs.empty());
}
CAF_TEST(resolve request on name with result) {
handle_handshake();
consume_handshake();
sys.registry().put(atom("foo"), self);
std::string path = "name/foo";
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
MOCK(basp::message_type::resolve_request, 42, path);
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
actor_id aid;
std::set<std::string> ifs;
RECEIVE(basp::message_type::resolve_response, 42u, aid, ifs);
CAF_CHECK_EQUAL(aid, self->id());
CAF_CHECK(ifs.empty());
}
CAF_TEST(resolve response with invalid actor handle) {
handle_handshake();
consume_handshake();
app.resolve(*this, "foo/bar", self);
std::string path;
RECEIVE(basp::message_type::resolve_request, 1u, path);
CAF_CHECK_EQUAL(path, "foo/bar");
actor_id aid = 0;
std::set<std::string> ifs;
MOCK(basp::message_type::resolve_response, 1u, aid, ifs);
self->receive([&](strong_actor_ptr& hdl, std::set<std::string>& hdl_ifs) {
CAF_CHECK_EQUAL(hdl, nullptr);
CAF_CHECK_EQUAL(ifs, hdl_ifs);
});
}
CAF_TEST(resolve response with valid actor handle) {
handle_handshake();
consume_handshake();
app.resolve(*this, "foo/bar", self);
std::string path;
RECEIVE(basp::message_type::resolve_request, 1u, path);
CAF_CHECK_EQUAL(path, "foo/bar");
actor_id aid = 42;
std::set<std::string> ifs;
MOCK(basp::message_type::resolve_response, 1u, aid, ifs);
self->receive([&](strong_actor_ptr& hdl, std::set<std::string>& hdl_ifs) {
CAF_REQUIRE(hdl != nullptr);
CAF_CHECK_EQUAL(ifs, hdl_ifs);
CAF_CHECK_EQUAL(hdl->id(), aid);
});
}
CAF_TEST(heartbeat message) {
handle_handshake();
consume_handshake();
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
auto bytes = to_bytes(basp::header{basp::message_type::heartbeat, 0, 0});
set_input(bytes);
REQUIRE_OK(app.handle_data(*this, input));
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -22,6 +22,8 @@ ...@@ -22,6 +22,8 @@
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <cstring> #include <cstring>
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
...@@ -31,40 +33,80 @@ ...@@ -31,40 +33,80 @@
using namespace caf; using namespace caf;
using namespace caf::detail; using namespace caf::detail;
namespace {
struct fixture : host_fixture {
fixture() : host_fixture() {
memset(&sockaddr6_src, 0, sizeof(sockaddr_in6));
memset(&sockaddr6_dst, 0, sizeof(sockaddr_in6));
sockaddr6_src.sin6_family = AF_INET6;
sockaddr6_src.sin6_port = htons(23);
sockaddr6_src.sin6_addr = in6addr_loopback;
memset(&sockaddr4_src, 0, sizeof(sockaddr_in));
memset(&sockaddr4_dst, 0, sizeof(sockaddr_in));
sockaddr4_src.sin_family = AF_INET;
sockaddr4_src.sin_port = htons(23);
sockaddr4_src.sin_addr.s_addr = INADDR_LOOPBACK;
}
sockaddr_in6 sockaddr6_src;
sockaddr_in6 sockaddr6_dst;
sockaddr_in sockaddr4_src;
sockaddr_in sockaddr4_dst;
ip_endpoint ep_src;
ip_endpoint ep_dst;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(convert_ip_endpoint_tests, fixture)
CAF_TEST(sockaddr_in6 roundtrip) { CAF_TEST(sockaddr_in6 roundtrip) {
sockaddr_in6 source_addr = {}; ip_endpoint ep;
source_addr.sin6_family = AF_INET6; CAF_MESSAGE("converting sockaddr_in6 to ip_endpoint");
source_addr.sin6_port = htons(23); CAF_CHECK_EQUAL(convert(reinterpret_cast<sockaddr_storage&>(sockaddr6_src),
source_addr.sin6_addr = in6addr_loopback; ep),
auto ep = to_ip_endpoint(reinterpret_cast<sockaddr_storage&>(source_addr)); none);
auto dest_addr = to_sockaddr(ep); CAF_MESSAGE("converting ip_endpoint to sockaddr_in6");
CAF_CHECK_EQUAL(memcmp(&source_addr, &dest_addr, sizeof(sockaddr_in6)), 0); convert(ep, reinterpret_cast<sockaddr_storage&>(sockaddr6_dst));
CAF_CHECK_EQUAL(memcmp(&sockaddr6_src, &sockaddr6_dst, sizeof(sockaddr_in6)),
0);
} }
CAF_TEST(ipv6_endpoint roundtrip) { CAF_TEST(ipv6_endpoint roundtrip) {
ipv6_endpoint source_ep; sockaddr_storage addr;
if (auto err = detail::parse("[::1]:55555", source_ep)) memset(&addr, 0, sizeof(sockaddr_storage));
if (auto err = detail::parse("[::1]:55555", ep_src))
CAF_FAIL("unable to parse input: " << err); CAF_FAIL("unable to parse input: " << err);
auto addr = to_sockaddr(source_ep); CAF_MESSAGE("converting ip_endpoint to sockaddr_in6");
auto dest_ep = to_ip_endpoint(addr); convert(ep_src, addr);
CAF_CHECK_EQUAL(source_ep, dest_ep); CAF_MESSAGE("converting sockaddr_in6 to ip_endpoint");
CAF_CHECK_EQUAL(convert(addr, ep_dst), none);
CAF_CHECK_EQUAL(ep_src, ep_dst);
} }
CAF_TEST(sockaddr_in4 roundtrip) { CAF_TEST(sockaddr_in4 roundtrip) {
sockaddr_in source_addr = {}; ip_endpoint ep;
source_addr.sin_family = AF_INET; CAF_MESSAGE("converting sockaddr_in to ip_endpoint");
source_addr.sin_port = htons(23); CAF_CHECK_EQUAL(convert(reinterpret_cast<sockaddr_storage&>(sockaddr4_src),
source_addr.sin_addr.s_addr = INADDR_LOOPBACK; ep),
auto ep = to_ip_endpoint(reinterpret_cast<sockaddr_storage&>(source_addr)); none);
auto dest_addr = to_sockaddr(ep); CAF_MESSAGE("converting ip_endpoint to sockaddr_in");
CAF_CHECK_EQUAL(memcmp(&source_addr, &dest_addr, sizeof(sockaddr_in)), 0); convert(ep, reinterpret_cast<sockaddr_storage&>(sockaddr4_dst));
CAF_CHECK_EQUAL(memcmp(&sockaddr4_src, &sockaddr4_dst, sizeof(sockaddr_in)),
0);
} }
CAF_TEST(ipv4_endpoint roundtrip) { CAF_TEST(ipv4_endpoint roundtrip) {
ipv6_endpoint source_ep; sockaddr_storage addr;
if (auto err = detail::parse("127.0.0.1:55555", source_ep)) memset(&addr, 0, sizeof(sockaddr_storage));
if (auto err = detail::parse("127.0.0.1:55555", ep_src))
CAF_FAIL("unable to parse input: " << err); CAF_FAIL("unable to parse input: " << err);
auto addr = to_sockaddr(source_ep); CAF_MESSAGE("converting ip_endpoint to sockaddr_in");
auto dest_ep = to_ip_endpoint(addr); convert(ep_src, addr);
CAF_CHECK_EQUAL(source_ep, dest_ep); CAF_MESSAGE("converting sockaddr_in to ip_endpoint");
CAF_CHECK_EQUAL(convert(addr, ep_dst), none);
CAF_CHECK_EQUAL(ep_src, ep_dst);
} }
CAF_TEST_FIXTURE_SCOPE_END();
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#define CAF_SUITE basp.header
#include "caf/net/basp/header.hpp"
#include "caf/test/dsl.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/serializer_impl.hpp"
using namespace caf;
using namespace caf::net;
CAF_TEST(serialization) {
basp::header x{basp::message_type::handshake, 42, 4};
std::vector<byte> buf;
{
serializer_impl<std::vector<byte>> sink{nullptr, buf};
CAF_CHECK_EQUAL(sink(x), none);
}
CAF_CHECK_EQUAL(buf.size(), basp::header_size);
auto buf2 = to_bytes(x);
CAF_REQUIRE_EQUAL(buf.size(), buf2.size());
CAF_CHECK(std::equal(buf.begin(), buf.end(), buf2.begin()));
basp::header y;
{
binary_deserializer source{nullptr, buf};
CAF_CHECK_EQUAL(source(y), none);
}
CAF_CHECK_EQUAL(x, y);
auto z = basp::header::from_bytes(buf);
CAF_CHECK_EQUAL(x, z);
CAF_CHECK_EQUAL(y, z);
}
CAF_TEST(to_string) {
basp::header x{basp::message_type::handshake, 42, 4};
CAF_CHECK_EQUAL(to_string(x), "basp::header(handshake, 42, 4)");
}
...@@ -32,7 +32,7 @@ using namespace caf::net; ...@@ -32,7 +32,7 @@ using namespace caf::net;
CAF_TEST_FIXTURE_SCOPE(ip_tests, host_fixture) CAF_TEST_FIXTURE_SCOPE(ip_tests, host_fixture)
CAF_TEST(resolve) { CAF_TEST(resolve localhost) {
ip_address v4_local{make_ipv4_address(127, 0, 0, 1)}; ip_address v4_local{make_ipv4_address(127, 0, 0, 1)};
ip_address v6_local{{0}, {0x1}}; ip_address v6_local{{0}, {0x1}};
auto addrs = ip::resolve("localhost"); auto addrs = ip::resolve("localhost");
...@@ -43,4 +43,15 @@ CAF_TEST(resolve) { ...@@ -43,4 +43,15 @@ CAF_TEST(resolve) {
CAF_CHECK(contains(v4_local) || contains(v6_local)); CAF_CHECK(contains(v4_local) || contains(v6_local));
} }
CAF_TEST(resolve any) {
ip_address v4_any{make_ipv4_address(0, 0, 0, 0)};
ip_address v6_any{{0}, {0}};
auto addrs = ip::resolve("");
CAF_CHECK(!addrs.empty());
auto contains = [&](ip_address x) {
return std::count(addrs.begin(), addrs.end(), x) > 0;
};
CAF_CHECK(contains(v4_any) || contains(v6_any));
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -22,7 +22,6 @@ ...@@ -22,7 +22,6 @@
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "caf/net/abstract_socket.hpp"
#include "caf/net/socket_id.hpp" #include "caf/net/socket_id.hpp"
using namespace caf; using namespace caf;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#define CAF_SUITE basp.stream_application
#include "caf/net/basp/application.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <vector>
#include "caf/byte.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/none.hpp"
#include "caf/uri.hpp"
using namespace caf;
using namespace caf::net;
#define REQUIRE_OK(statement) \
if (auto err = statement) \
CAF_FAIL(sys.render(err));
namespace {
using buffer_type = std::vector<byte>;
using transport_type = stream_transport<basp::application>;
size_t fetch_size(variant<size_t, sec> x) {
if (holds_alternative<sec>(x))
CAF_FAIL("read/write failed: " << to_string(get<sec>(x)));
return get<size_t>(x);
}
struct fixture : test_coordinator_fixture<>,
host_fixture,
proxy_registry::backend {
fixture() {
uri mars_uri;
REQUIRE_OK(parse("tcp://mars", mars_uri));
mars = make_node_id(mars_uri);
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
auto proxies = std::make_shared<proxy_registry>(sys, *this);
auto sockets = unbox(make_stream_socket_pair());
sock = sockets.first;
nonblocking(sockets.first, true);
nonblocking(sockets.second, true);
mgr = make_endpoint_manager(mpx, sys,
transport_type{sockets.second,
basp::application{proxies}});
REQUIRE_OK(mgr->init());
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
auto& dref = dynamic_cast<endpoint_manager_impl<transport_type>&>(*mgr);
app = &dref.application();
}
~fixture() {
close(sock);
}
bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false);
}
template <class... Ts>
buffer_type to_buf(const Ts&... xs) {
buffer_type buf;
serializer_impl<buffer_type> sink{system(), buf};
REQUIRE_OK(sink(xs...));
return buf;
}
template <class... Ts>
void mock(const Ts&... xs) {
auto buf = to_buf(xs...);
if (fetch_size(write(sock, make_span(buf))) != buf.size())
CAF_FAIL("unable to write " << buf.size() << " bytes");
run();
}
void handle_handshake() {
CAF_CHECK_EQUAL(app->state(),
basp::connection_state::await_handshake_header);
auto payload = to_buf(mars, defaults::middleman::app_identifiers);
mock(basp::header{basp::message_type::handshake,
static_cast<uint32_t>(payload.size()), basp::version});
CAF_CHECK_EQUAL(app->state(),
basp::connection_state::await_handshake_payload);
write(sock, make_span(payload));
run();
}
void consume_handshake() {
buffer_type buf(basp::header_size);
if (fetch_size(read(sock, make_span(buf))) != basp::header_size)
CAF_FAIL("unable to read " << basp::header_size << " bytes");
auto hdr = basp::header::from_bytes(buf);
if (hdr.type != basp::message_type::handshake || hdr.payload_len == 0
|| hdr.operation_data != basp::version)
CAF_FAIL("invalid handshake header");
buf.resize(hdr.payload_len);
if (fetch_size(read(sock, make_span(buf))) != hdr.payload_len)
CAF_FAIL("unable to read " << hdr.payload_len << " bytes");
node_id nid;
std::vector<std::string> app_ids;
binary_deserializer source{sys, buf};
if (auto err = source(nid, app_ids))
CAF_FAIL("unable to deserialize payload: " << sys.render(err));
if (source.remaining() > 0)
CAF_FAIL("trailing bytes after reading payload");
}
actor_system& system() {
return sys;
}
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override {
using impl_type = forwarding_actor_proxy;
using hdl_type = strong_actor_ptr;
actor_config cfg;
return make_actor<impl_type, hdl_type>(aid, nid, &sys, cfg, self);
}
void set_last_hop(node_id*) override {
// nop
}
node_id mars;
multiplexer_ptr mpx;
endpoint_manager_ptr mgr;
stream_socket sock;
basp::application* app;
};
} // namespace
#define MOCK(kind, op, ...) \
do { \
auto payload = to_buf(__VA_ARGS__); \
mock(basp::header{kind, static_cast<uint32_t>(payload.size()), op}); \
write(sock, make_span(payload)); \
run(); \
} while (false)
#define RECEIVE(msg_type, op_data, ...) \
do { \
buffer_type buf(basp::header_size); \
if (fetch_size(read(sock, make_span(buf))) != basp::header_size) \
CAF_FAIL("unable to read " << basp::header_size << " bytes"); \
auto hdr = basp::header::from_bytes(buf); \
CAF_CHECK_EQUAL(hdr.type, msg_type); \
CAF_CHECK_EQUAL(hdr.operation_data, op_data); \
buf.resize(hdr.payload_len); \
if (fetch_size(read(sock, make_span(buf))) != size_t{hdr.payload_len}) \
CAF_FAIL("unable to read " << hdr.payload_len << " bytes"); \
binary_deserializer source{sys, buf}; \
if (auto err = source(__VA_ARGS__)) \
CAF_FAIL("failed to receive data: " << sys.render(err)); \
} while (false)
CAF_TEST_FIXTURE_SCOPE(application_tests, fixture)
CAF_TEST(actor message) {
handle_handshake();
consume_handshake();
sys.registry().put(self->id(), self);
CAF_REQUIRE_EQUAL(self->mailbox().size(), 0u);
MOCK(basp::message_type::actor_message, make_message_id().integer_value(),
mars, actor_id{42}, self->id(), std::vector<strong_actor_ptr>{},
make_message("hello world!"));
allow((atom_value, strong_actor_ptr),
from(_).to(self).with(atom("monitor"), _));
expect((std::string), from(_).to(self).with("hello world!"));
}
CAF_TEST(resolve request without result) {
handle_handshake();
consume_handshake();
CAF_CHECK_EQUAL(app->state(), basp::connection_state::await_header);
MOCK(basp::message_type::resolve_request, 42, std::string{"foo/bar"});
CAF_CHECK_EQUAL(app->state(), basp::connection_state::await_header);
actor_id aid;
std::set<std::string> ifs;
RECEIVE(basp::message_type::resolve_response, 42u, aid, ifs);
CAF_CHECK_EQUAL(aid, 0u);
CAF_CHECK(ifs.empty());
}
CAF_TEST(resolve request on id with result) {
handle_handshake();
consume_handshake();
sys.registry().put(self->id(), self);
auto path = "id/" + std::to_string(self->id());
CAF_CHECK_EQUAL(app->state(), basp::connection_state::await_header);
MOCK(basp::message_type::resolve_request, 42, path);
CAF_CHECK_EQUAL(app->state(), basp::connection_state::await_header);
actor_id aid;
std::set<std::string> ifs;
RECEIVE(basp::message_type::resolve_response, 42u, aid, ifs);
CAF_CHECK_EQUAL(aid, self->id());
CAF_CHECK(ifs.empty());
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -84,15 +84,17 @@ public: ...@@ -84,15 +84,17 @@ public:
rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end()); rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end());
} }
template <class Manager> template <class Parent>
void resolve(Manager& manager, const std::string& path, actor listener) { void resolve(Parent& parent, const std::string& path, actor listener) {
actor_id aid = 42; actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899"; auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(42, hid)); auto nid = unbox(make_node_id(42, hid));
actor_config cfg; actor_config cfg;
endpoint_manager_ptr ptr{&parent.manager()};
auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid, auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid,
&manager.system(), &parent.system(),
cfg, &manager); cfg,
std::move(ptr));
anon_send(listener, resolve_atom::value, std::move(path), p); anon_send(listener, resolve_atom::value, std::move(path), p);
} }
......
...@@ -155,16 +155,16 @@ public: ...@@ -155,16 +155,16 @@ public:
} }
} }
template <class Manager> template <class Parent>
void resolve(Manager& manager, const std::string& path, actor listener) { void resolve(Parent& parent, const std::string& path, actor listener) {
actor_id aid = 42; actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899"; auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(aid, hid)); auto nid = unbox(make_node_id(aid, hid));
actor_config cfg; actor_config cfg;
auto p = make_actor<net::actor_proxy_impl, strong_actor_ptr>(aid, nid, auto sys = &parent.system();
&manager auto mgr = &parent.manager();
.system(), auto p = make_actor<net::actor_proxy_impl, strong_actor_ptr>(aid, nid, sys,
cfg, &manager); cfg, mgr);
anon_send(listener, resolve_atom::value, std::move(path), p); anon_send(listener, resolve_atom::value, std::move(path), p);
} }
......
...@@ -70,10 +70,10 @@ public: ...@@ -70,10 +70,10 @@ public:
return none; return none;
} }
template <class Transport> template <class Parent>
void write_message(Transport& transport, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager::message> msg) { std::unique_ptr<endpoint_manager::message> msg) {
transport.write_packet(span<byte>{}, msg->payload); parent.write_packet(span<const byte>{}, msg->payload);
} }
template <class Parent> template <class Parent>
...@@ -192,6 +192,7 @@ CAF_TEST(write_message) { ...@@ -192,6 +192,7 @@ CAF_TEST(write_message) {
hello_test.size()); hello_test.size());
std::vector<byte> payload(test_span.begin(), test_span.end()); std::vector<byte> payload(test_span.begin(), test_span.end());
auto message = detail::make_unique<endpoint_manager::message>(std::move(elem), auto message = detail::make_unique<endpoint_manager::message>(std::move(elem),
nullptr,
payload); payload);
worker.write_message(transport, std::move(message)); worker.write_message(transport, std::move(message));
auto& buf = transport_results->packet_buffer; auto& buf = transport_results->packet_buffer;
......
...@@ -187,13 +187,20 @@ struct fixture : host_fixture { ...@@ -187,13 +187,20 @@ struct fixture : host_fixture {
make_message_id(12345), std::move(stack), make_message_id(12345), std::move(stack),
make_message()); make_message());
return detail::make_unique<endpoint_manager::message>(std::move(elem), return detail::make_unique<endpoint_manager::message>(std::move(elem),
strong_actor,
payload); payload);
} }
bool contains(byte x) {
return std::count(buf->begin(), buf->end(), x) > 0;
}
void add_new_workers() { void add_new_workers() {
for (auto& data : test_data) { for (auto& data : test_data) {
dispatcher.add_new_worker(data.nid, data.ep); if (auto err = dispatcher.add_new_worker(dummy, data.nid, data.ep))
CAF_FAIL("add_new_worker returned an error: " << err);
} }
buf->clear();
} }
void test_write_message(testdata& testcase) { void test_write_message(testdata& testcase) {
...@@ -245,16 +252,9 @@ struct fixture : host_fixture { ...@@ -245,16 +252,9 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE(transport_worker_dispatcher_test, fixture) CAF_TEST_FIXTURE_SCOPE(transport_worker_dispatcher_test, fixture)
CAF_TEST(init) { CAF_TEST(init) {
auto contains = [&](byte x) { dispatcher_type dispatcher{dummy_application_factory{buf}};
return std::count(buf->begin(), buf->end(), x) > 0;
};
if (auto err = dispatcher.init(dummy)) if (auto err = dispatcher.init(dummy))
CAF_FAIL("init failed with error: " << err); CAF_FAIL("init failed with error: " << err);
CAF_CHECK_EQUAL(buf->size(), 4u);
CAF_CHECK(contains(byte(0)));
CAF_CHECK(contains(byte(1)));
CAF_CHECK(contains(byte(2)));
CAF_CHECK(contains(byte(3)));
} }
CAF_TEST(handle_data) { CAF_TEST(handle_data) {
...@@ -283,9 +283,6 @@ CAF_TEST(timeout) { ...@@ -283,9 +283,6 @@ CAF_TEST(timeout) {
} }
CAF_TEST(handle_error) { CAF_TEST(handle_error) {
auto contains = [&](byte x) {
return std::count(buf->begin(), buf->end(), x) > 0;
};
dispatcher.handle_error(sec::unavailable_or_would_block); dispatcher.handle_error(sec::unavailable_or_would_block);
CAF_CHECK_EQUAL(buf->size(), 4u); CAF_CHECK_EQUAL(buf->size(), 4u);
CAF_CHECK(contains(byte(0))); CAF_CHECK(contains(byte(0)));
......
...@@ -27,6 +27,7 @@ ...@@ -27,6 +27,7 @@
#include "caf/detail/net_syscall.hpp" #include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/ip_address.hpp" #include "caf/ip_address.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/ipv4_address.hpp" #include "caf/ipv4_address.hpp"
#include "caf/net/ip.hpp" #include "caf/net/ip.hpp"
#include "caf/net/socket_guard.hpp" #include "caf/net/socket_guard.hpp"
...@@ -43,6 +44,7 @@ struct fixture : host_fixture { ...@@ -43,6 +44,7 @@ struct fixture : host_fixture {
: host_fixture(), : host_fixture(),
v4_local{make_ipv4_address(127, 0, 0, 1)}, v4_local{make_ipv4_address(127, 0, 0, 1)},
v6_local{{0}, {0x1}}, v6_local{{0}, {0x1}},
// TODO: use local_addresses() when merged
addrs{net::ip::resolve("localhost")} { addrs{net::ip::resolve("localhost")} {
} }
...@@ -53,25 +55,29 @@ struct fixture : host_fixture { ...@@ -53,25 +55,29 @@ struct fixture : host_fixture {
void test_send_receive(ip_address addr) { void test_send_receive(ip_address addr) {
std::vector<byte> buf(1024); std::vector<byte> buf(1024);
ip_endpoint ep(addr, 0); ip_endpoint ep(addr, 0);
auto sender = unbox(make_udp_datagram_socket(ep)); auto send_pair = unbox(make_udp_datagram_socket(ep));
ep.port(0); auto send_socket = send_pair.first;
auto receiver = unbox(make_udp_datagram_socket(ep)); auto receive_pair = unbox(make_udp_datagram_socket(ep));
auto receive_socket = receive_pair.first;
ep.port(ntohs(receive_pair.second));
CAF_MESSAGE("sending data to: " << to_string(ep)); CAF_MESSAGE("sending data to: " << to_string(ep));
auto send_guard = make_socket_guard(sender); auto send_guard = make_socket_guard(send_socket);
auto receive_guard = make_socket_guard(receiver); auto receive_guard = make_socket_guard(receive_socket);
if (auto err = nonblocking(socket_cast<net::socket>(receiver), true)) if (auto err = nonblocking(socket_cast<net::socket>(receive_socket), true))
CAF_FAIL("nonblocking returned an error" << err); CAF_FAIL("nonblocking returned an error" << err);
auto test_read_res = read(receiver, make_span(buf)); auto test_read_res = read(receive_socket, make_span(buf));
if (auto err = get_if<sec>(&test_read_res)) if (auto err = get_if<sec>(&test_read_res))
CAF_CHECK_EQUAL(*err, sec::unavailable_or_would_block); CAF_CHECK_EQUAL(*err, sec::unavailable_or_would_block);
else else
CAF_FAIL("read should have failed"); CAF_FAIL("read should have failed");
auto write_ret = write(sender, as_bytes(make_span(hello_test)), ep);
auto write_ret = write(send_socket, as_bytes(make_span(hello_test)), ep);
if (auto num_bytes = get_if<size_t>(&write_ret)) if (auto num_bytes = get_if<size_t>(&write_ret))
CAF_CHECK_EQUAL(*num_bytes, hello_test.size()); CAF_CHECK_EQUAL(*num_bytes, hello_test.size());
else else
CAF_FAIL("write returned an error: " << sys.render(get<sec>(write_ret))); CAF_FAIL("write returned an error: " << sys.render(get<sec>(write_ret)));
auto read_ret = read(receiver, make_span(buf));
auto read_ret = read(receive_socket, make_span(buf));
if (auto read_res = get_if<std::pair<size_t, ip_endpoint>>(&read_ret)) { if (auto read_res = get_if<std::pair<size_t, ip_endpoint>>(&read_ret)) {
CAF_CHECK_EQUAL(read_res->first, hello_test.size()); CAF_CHECK_EQUAL(read_res->first, hello_test.size());
buf.resize(read_res->first); buf.resize(read_res->first);
...@@ -94,6 +100,8 @@ struct fixture : host_fixture { ...@@ -94,6 +100,8 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE(udp_datagram_socket_test, fixture) CAF_TEST_FIXTURE_SCOPE(udp_datagram_socket_test, fixture)
CAF_TEST(send and receive) { CAF_TEST(send and receive) {
// TODO: check which versions exist and test existing versions accordingly
// -> local_addresses()
if (contains(v4_local)) { if (contains(v4_local)) {
test_send_receive(v4_local); test_send_receive(v4_local);
} else if (contains(v6_local)) { } else if (contains(v6_local)) {
......
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