Commit f8c1c492 authored by Dominik Charousset's avatar Dominik Charousset

Integrate review feedback

parent d4175657
...@@ -45,6 +45,7 @@ namespace caf { ...@@ -45,6 +45,7 @@ namespace caf {
namespace net { namespace net {
namespace basp { namespace basp {
/// An implementation of BASP as an application layer protocol.
class application { class application {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -184,10 +185,10 @@ private: ...@@ -184,10 +185,10 @@ private:
/// Stores a pointer to the parent actor system. /// Stores a pointer to the parent actor system.
actor_system* system_ = nullptr; actor_system* system_ = nullptr;
/// Stores what we are expecting to receive next. /// Stores the expected type of the next incoming message.
connection_state state_ = connection_state::await_handshake_header; connection_state state_ = connection_state::await_handshake_header;
/// Caches the last header;we need to store it when waiting for the payload. /// Caches the last header while waiting for the matching payload.
header hdr_; header hdr_;
/// Re-usable buffer for storing payloads. /// Re-usable buffer for storing payloads.
...@@ -199,7 +200,7 @@ private: ...@@ -199,7 +200,7 @@ private:
/// Stores the ID of our peer. /// Stores the ID of our peer.
node_id peer_id_; node_id peer_id_;
/// Keeps track of which local actors our peer monitors. /// Tracks which local actors our peer monitors.
std::unordered_set<actor_addr> monitored_actors_; std::unordered_set<actor_addr> monitored_actors_;
/// Caches actor handles obtained via `resolve`. /// Caches actor handles obtained via `resolve`.
......
...@@ -28,11 +28,10 @@ namespace basp { ...@@ -28,11 +28,10 @@ namespace basp {
/// Stores the state of a connection in a `basp::application`. /// Stores the state of a connection in a `basp::application`.
enum class connection_state { enum class connection_state {
/// Indicates that we successfully checked the magic number and now wait for /// Initial state for any connection to wait for the peer's handshake.
/// the handshake header.
await_handshake_header, await_handshake_header,
/// Indicates that we received the header for the handshake and now wait for /// Indicates that the header for the peer's handshake arrived and BASP
/// the payload. /// requires the payload next.
await_handshake_payload, await_handshake_payload,
/// Indicates that a connection is established and this node is waiting for /// Indicates that a connection is established and this node is waiting for
/// the next BASP header. /// the next BASP header.
...@@ -40,7 +39,7 @@ enum class connection_state { ...@@ -40,7 +39,7 @@ enum class connection_state {
/// Indicates that this node has received a header with non-zero payload and /// Indicates that this node has received a header with non-zero payload and
/// is waiting for the data. /// is waiting for the data.
await_payload, await_payload,
/// Indicates that we are about to close this connection. /// Indicates that the connection is about to shut down.
shutdown, shutdown,
}; };
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#pragma once #pragma once
#include <cstddef>
#include <cstdint> #include <cstdint>
namespace caf { namespace caf {
...@@ -30,6 +31,9 @@ namespace basp { ...@@ -30,6 +31,9 @@ namespace basp {
/// @note BASP is not backwards compatible. /// @note BASP is not backwards compatible.
constexpr uint64_t version = 1; constexpr uint64_t version = 1;
/// Size of a BASP header in serialized form.
constexpr size_t header_size = 13;
/// @} /// @}
} // namespace basp } // namespace basp
......
...@@ -21,12 +21,12 @@ ...@@ -21,12 +21,12 @@
#include <array> #include <array>
#include <cstdint> #include <cstdint>
#include "caf/byte.hpp"
#include "caf/detail/comparable.hpp" #include "caf/detail/comparable.hpp"
#include "caf/fwd.hpp"
#include "caf/meta/hex_formatted.hpp" #include "caf/meta/hex_formatted.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/message_type.hpp" #include "caf/net/basp/message_type.hpp"
#include "caf/span.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
...@@ -34,10 +34,9 @@ namespace basp { ...@@ -34,10 +34,9 @@ namespace basp {
/// @addtogroup BASP /// @addtogroup BASP
/// The header of a Binary Actor System Protocol (BASP) message.
struct header : detail::comparable<header> { struct header : detail::comparable<header> {
message_type type; // -- constructors, destructors, and assignment operators --------------------
uint32_t payload_len;
uint64_t operation_data;
constexpr header() noexcept constexpr header() noexcept
: type(message_type::handshake), payload_len(0), operation_data(0) { : type(message_type::handshake), payload_len(0), operation_data(0) {
...@@ -54,15 +53,28 @@ struct header : detail::comparable<header> { ...@@ -54,15 +53,28 @@ struct header : detail::comparable<header> {
header& operator=(const header&) noexcept = default; header& operator=(const header&) noexcept = default;
int compare(header other) const noexcept; // -- factory functions ------------------------------------------------------
/// @pre `bytes.size() == header_size` /// @pre `bytes.size() == header_size`
static header from_bytes(span<const byte> bytes); static header from_bytes(span<const byte> bytes);
};
/// Size of a BASP header in serialized form // -- comparison -------------------------------------------------------------
constexpr size_t header_size = 13;
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 /// @relates header
std::array<byte, header_size> to_bytes(header x); std::array<byte, header_size> to_bytes(header x);
......
...@@ -33,41 +33,41 @@ enum class message_type : uint8_t { ...@@ -33,41 +33,41 @@ enum class message_type : uint8_t {
/// Sends supported BASP version and node information to the server. /// Sends supported BASP version and node information to the server.
/// ///
/// ![](client_handshake.png) /// ![](client_handshake.png)
handshake = 0x00, handshake = 0,
/// Transmits an actor-to-actor messages. /// Transmits an actor-to-actor messages.
/// ///
/// ![](direct_message.png) /// ![](direct_message.png)
actor_message = 0x01, actor_message = 1,
/// Tries to resolve a path on the receiving node. /// Tries to resolve a path on the receiving node.
/// ///
/// ![](resolve_request.png) /// ![](resolve_request.png)
resolve_request = 0x02, resolve_request = 2,
/// Transmits the result of a path lookup. /// Transmits the result of a path lookup.
/// ///
/// ![](resolve_response.png) /// ![](resolve_response.png)
resolve_response = 0x03, resolve_response = 3,
/// Informs the receiving node that the sending node has created a proxy /// 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 /// instance for one of its actors. Causes the receiving node to attach a
/// functor to the actor that triggers a down_message on termination. /// functor to the actor that triggers a down_message on termination.
/// ///
/// ![](monitor_message.png) /// ![](monitor_message.png)
monitor_message = 0x04, monitor_message = 4,
/// Informs the receiving node that it has a proxy for an actor that has been /// Informs the receiving node that it has a proxy for an actor that has been
/// terminated. /// terminated.
/// ///
/// ![](down_message.png) /// ![](down_message.png)
down_message = 0x05, down_message = 5,
/// Used to generate periodic traffic between two nodes in order to detect /// Used to generate periodic traffic between two nodes in order to detect
/// disconnects. /// disconnects.
/// ///
/// ![](heartbeat.png) /// ![](heartbeat.png)
heartbeat = 0x06, heartbeat = 6,
}; };
/// @relates message_type /// @relates message_type
......
...@@ -89,7 +89,7 @@ void application::resolve_remote_path(write_packet_callback& write_packet, ...@@ -89,7 +89,7 @@ void application::resolve_remote_path(write_packet_callback& write_packet,
auto hdr = to_bytes(header{message_type::resolve_request, auto hdr = to_bytes(header{message_type::resolve_request,
static_cast<uint32_t>(buf_.size()), req_id}); static_cast<uint32_t>(buf_.size()), req_id});
if (auto err = write_packet(hdr, buf_)) { if (auto err = write_packet(hdr, buf_)) {
CAF_LOG_ERROR("unable to serialize path"); CAF_LOG_ERROR("unable to write resolve_request header");
return; return;
} }
response_promise rp{nullptr, actor_cast<strong_actor_ptr>(listener), response_promise rp{nullptr, actor_cast<strong_actor_ptr>(listener),
...@@ -116,20 +116,8 @@ error application::handle(size_t& next_read_size, ...@@ -116,20 +116,8 @@ error application::handle(size_t& next_read_size,
return none; return none;
} }
case connection_state::await_handshake_payload: { case connection_state::await_handshake_payload: {
node_id nid; if (auto err = handle_handshake(write_packet, hdr_, bytes))
std::vector<std::string> app_ids;
binary_deserializer source{system(), bytes};
if (auto err = source(nid, app_ids))
return err; return err;
if (!nid || 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;
state_ = connection_state::await_header; state_ = connection_state::await_header;
return none; return none;
} }
...@@ -178,12 +166,20 @@ error application::handle_handshake(write_packet_callback&, header hdr, ...@@ -178,12 +166,20 @@ error application::handle_handshake(write_packet_callback&, header hdr,
return ec::missing_handshake; return ec::missing_handshake;
if (hdr.operation_data != version) if (hdr.operation_data != version)
return ec::version_mismatch; return ec::version_mismatch;
binary_deserializer source{nullptr, payload};
node_id peer_id; node_id peer_id;
std::vector<std::string> app_ids; std::vector<std::string> app_ids;
binary_deserializer source{system(), payload};
if (auto err = source(peer_id, app_ids)) if (auto err = source(peer_id, app_ids))
return err; return err;
// TODO: verify peer_id and app_ids 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); peer_id_ = std::move(peer_id);
state_ = connection_state::await_header; state_ = connection_state::await_header;
return none; return none;
...@@ -276,6 +272,7 @@ error application::handle_resolve_response(write_packet_callback&, header hdr, ...@@ -276,6 +272,7 @@ error application::handle_resolve_response(write_packet_callback&, header hdr,
} }
error application::generate_handshake() { error application::generate_handshake() {
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_}; serializer_impl<buffer_type> sink{system(), buf_};
return sink(system().node(), return sink(system().node(),
get_or(system().config(), "middleman.app-identifiers", get_or(system().config(), "middleman.app-identifiers",
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/string_view.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
...@@ -27,7 +28,7 @@ namespace basp { ...@@ -27,7 +28,7 @@ namespace basp {
namespace { namespace {
const char* ec_names[] = { string_view ec_names[] = {
"none", "none",
"invalid_magic_number", "invalid_magic_number",
"unexpected_number_of_bytes", "unexpected_number_of_bytes",
...@@ -36,8 +37,9 @@ const char* ec_names[] = { ...@@ -36,8 +37,9 @@ const char* ec_names[] = {
"illegal_state", "illegal_state",
"invalid_handshake", "invalid_handshake",
"missing_handshake", "missing_handshake",
"unexpected_handshake",
"version_mismatch", "version_mismatch",
"unimplemented", "unimplemented = 10",
"app_identifiers_mismatch", "app_identifiers_mismatch",
"invalid_payload", "invalid_payload",
}; };
...@@ -45,7 +47,8 @@ const char* ec_names[] = { ...@@ -45,7 +47,8 @@ const char* ec_names[] = {
} // namespace } // namespace
std::string to_string(ec x) { std::string to_string(ec x) {
return ec_names[static_cast<uint8_t>(x)]; auto result = ec_names[static_cast<uint8_t>(x)];
return std::string{result.begin(), result.end()};
} }
error make_error(ec x) { error make_error(ec x) {
......
...@@ -20,7 +20,9 @@ ...@@ -20,7 +20,9 @@
#include <cstring> #include <cstring>
#include "caf/byte.hpp"
#include "caf/detail/network_order.hpp" #include "caf/detail/network_order.hpp"
#include "caf/span.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
......
...@@ -34,8 +34,8 @@ string_view message_type_names[] = { ...@@ -34,8 +34,8 @@ string_view message_type_names[] = {
} // namespace } // namespace
std::string to_string(message_type x) { std::string to_string(message_type x) {
auto sv = message_type_names[static_cast<uint8_t>(x)]; auto result = message_type_names[static_cast<uint8_t>(x)];
return std::string{sv.begin(), sv.end()}; return std::string{result.begin(), result.end()};
} }
} // namespace basp } // namespace basp
......
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