Commit 87d4ebdd authored by Dominik Charousset's avatar Dominik Charousset

Port caf_net to latest CAF master API

parent 61efab6c
...@@ -2,30 +2,35 @@ ...@@ -2,30 +2,35 @@
file(GLOB_RECURSE CAF_NET_HEADERS "caf/*.hpp") file(GLOB_RECURSE CAF_NET_HEADERS "caf/*.hpp")
# -- auto generate to_string for enum types ------------------------------------ # -- add consistency checks for enum to_string implementations -----------------
enum_to_string("caf/net/basp/connection_state.hpp" "basp_conn_strings.cpp") add_enum_consistency_check("caf/net/basp/connection_state.hpp"
enum_to_string("caf/net/basp/ec.hpp" "basp_ec_strings.cpp") "src/basp/connection_state_strings.cpp")
enum_to_string("caf/net/basp/message_type.hpp" "basp_message_type_strings.cpp") add_enum_consistency_check("caf/net/basp/ec.hpp"
enum_to_string("caf/net/operation.hpp" "operation_strings.cpp") "src/basp/ec_strings.cpp")
add_enum_consistency_check("caf/net/basp/message_type.hpp"
"src/basp/message_type_strings.cpp")
add_enum_consistency_check("caf/net/operation.hpp"
"src/basp/operation_strings.cpp")
# -- list cpp files for caf::net ----------------------------------------------- # -- list cpp files for caf::net -----------------------------------------------
set(CAF_NET_SOURCES set(CAF_NET_SOURCES
"${CMAKE_BINARY_DIR}/basp_conn_strings.cpp"
"${CMAKE_BINARY_DIR}/basp_ec_strings.cpp"
"${CMAKE_BINARY_DIR}/basp_message_type_strings.cpp"
"${CMAKE_BINARY_DIR}/operation_strings.cpp"
src/actor_proxy_impl.cpp src/actor_proxy_impl.cpp
src/application.cpp src/application.cpp
src/basp/connection_state_strings.cpp
src/basp/ec_strings.cpp
src/basp/message_type_strings.cpp
src/basp/operation_strings.cpp
src/convert_ip_endpoint.cpp src/convert_ip_endpoint.cpp
src/datagram_socket.cpp src/datagram_socket.cpp
src/defaults.cpp src/defaults.cpp
src/ec.cpp src/defaults.cpp
src/endpoint_manager.cpp src/endpoint_manager.cpp
src/header.cpp src/header.cpp
src/host.cpp src/host.cpp
src/ip.cpp src/ip.cpp
src/message_queue.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/net/backend/test.cpp src/net/backend/test.cpp
src/net/endpoint_manager_queue.cpp src/net/endpoint_manager_queue.cpp
...@@ -36,13 +41,11 @@ set(CAF_NET_SOURCES ...@@ -36,13 +41,11 @@ set(CAF_NET_SOURCES
src/pipe_socket.cpp src/pipe_socket.cpp
src/pollset_updater.cpp src/pollset_updater.cpp
src/socket.cpp src/socket.cpp
src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
src/socket_manager.cpp src/socket_manager.cpp
src/stream_socket.cpp src/stream_socket.cpp
src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
src/udp_datagram_socket.cpp src/udp_datagram_socket.cpp
src/defaults.cpp
src/message_queue.cpp
src/worker.cpp src/worker.cpp
) )
......
...@@ -117,7 +117,7 @@ public: ...@@ -117,7 +117,7 @@ public:
void local_actor_down(packet_writer& writer, actor_id id, error reason); void local_actor_down(packet_writer& writer, actor_id id, error reason);
template <class Parent> template <class Parent>
void timeout(Parent&, atom_value, uint64_t) { void timeout(Parent&, const std::string&, uint64_t) {
// nop // nop
} }
......
...@@ -21,7 +21,8 @@ ...@@ -21,7 +21,8 @@
#include <cstdint> #include <cstdint>
#include <string> #include <string>
#include "caf/fwd.hpp" #include "caf/detail/net_export.hpp"
#include "caf/error_category.hpp"
namespace caf::net::basp { namespace caf::net::basp {
...@@ -44,9 +45,15 @@ enum class ec : uint8_t { ...@@ -44,9 +45,15 @@ enum class ec : uint8_t {
}; };
/// @relates ec /// @relates ec
std::string to_string(ec x); CAF_NET_EXPORT std::string to_string(ec x);
/// @relates ec
error make_error(ec x);
} // namespace caf::net::basp } // namespace caf::net::basp
namespace caf {
template <>
struct error_category<net::basp::ec> {
static constexpr uint8_t value = 4;
};
} // namespace caf
...@@ -95,7 +95,7 @@ public: ...@@ -95,7 +95,7 @@ public:
template <class Parent> template <class Parent>
void resolve(Parent&, const uri& locator, const actor& listener) { void resolve(Parent&, const uri& locator, const actor& listener) {
CAF_LOG_ERROR("doorman called to resolve" << CAF_ARG(locator)); CAF_LOG_ERROR("doorman called to resolve" << CAF_ARG(locator));
anon_send(listener, resolve_atom::value, "doormen cannot resolve paths"); anon_send(listener, resolve_atom_v, "doormen cannot resolve paths");
} }
void new_proxy(endpoint_manager&, const node_id& peer, actor_id id) { void new_proxy(endpoint_manager&, const node_id& peer, actor_id id) {
...@@ -114,10 +114,9 @@ public: ...@@ -114,10 +114,9 @@ public:
} }
template <class Parent> template <class Parent>
void timeout(Parent&, atom_value x, uint64_t id) { void timeout(Parent&, [[maybe_unused]] const std::string& tag,
CAF_LOG_ERROR("doorman received timeout" << CAF_ARG(x) << CAF_ARG(id)); [[maybe_unused]] uint64_t id) {
CAF_IGNORE_UNUSED(x); CAF_LOG_ERROR("doorman received timeout" << CAF_ARG(tag) << CAF_ARG(id));
CAF_IGNORE_UNUSED(id);
} }
void handle_error(sec err) { void handle_error(sec err) {
......
...@@ -21,7 +21,6 @@ ...@@ -21,7 +21,6 @@
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/atom.hpp"
#include "caf/detail/overload.hpp" #include "caf/detail/overload.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
...@@ -63,11 +62,11 @@ public: ...@@ -63,11 +62,11 @@ public:
// -- timeout management ----------------------------------------------------- // -- timeout management -----------------------------------------------------
template <class... Ts> template <class... Ts>
uint64_t set_timeout(actor_clock::time_point tp, atom_value type, uint64_t set_timeout(actor_clock::time_point tp, std::string type,
Ts&&... xs) { Ts&&... xs) {
auto act = actor_cast<abstract_actor*>(timeout_proxy_); auto act = actor_cast<abstract_actor*>(timeout_proxy_);
CAF_ASSERT(act != nullptr); CAF_ASSERT(act != nullptr);
sys_.clock().set_multi_timeout(tp, act, type, next_timeout_id_); sys_.clock().set_multi_timeout(tp, act, std::move(type), next_timeout_id_);
transport_.set_timeout(next_timeout_id_, std::forward<Ts>(xs)...); transport_.set_timeout(next_timeout_id_, std::forward<Ts>(xs)...);
return next_timeout_id_++; return next_timeout_id_++;
} }
......
...@@ -18,6 +18,8 @@ ...@@ -18,6 +18,8 @@
#pragma once #pragma once
#include <string>
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp" #include "caf/intrusive/drr_queue.hpp"
...@@ -73,7 +75,7 @@ public: ...@@ -73,7 +75,7 @@ public:
}; };
struct timeout { struct timeout {
atom_value type; std::string type;
uint64_t id; uint64_t id;
}; };
...@@ -83,7 +85,7 @@ public: ...@@ -83,7 +85,7 @@ public:
event(node_id observing_peer, actor_id local_actor_id, error reason); event(node_id observing_peer, actor_id local_actor_id, error reason);
event(atom_value type, uint64_t id); event(std::string type, uint64_t id);
~event() override; ~event() override;
......
...@@ -66,13 +66,13 @@ public: ...@@ -66,13 +66,13 @@ public:
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
void cancel_timeout(atom_value type, uint64_t id) { void cancel_timeout(std::string tag, uint64_t id) {
parent_.cancel_timeout(type, id); parent_.cancel_timeout(std::move(tag), id);
} }
template <class... Ts> template <class... Ts>
uint64_t set_timeout(timestamp tout, atom_value type, Ts&&... xs) { uint64_t set_timeout(timestamp tout, std::string tag, Ts&&... xs) {
return parent_.set_timeout(tout, type, std::forward<Ts>(xs)...); return parent_.set_timeout(tout, std::move(tag), std::forward<Ts>(xs)...);
} }
protected: protected:
......
...@@ -110,9 +110,9 @@ public: ...@@ -110,9 +110,9 @@ public:
bool handle_write_event(endpoint_manager& manager) override { bool handle_write_event(endpoint_manager& manager) override {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id) CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id)
<< CAF_ARG2("queue-size", write_queue_.size())); << CAF_ARG2("queue-size", write_queue_.size()));
auto drain_write_queue = []() -> error_code<sec> { auto drain_write_queue = [this]() -> error_code<sec> {
// Helper function to sort empty buffers back into the right caches. // Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() { auto recycle = [this]() {
auto& front = this->write_queue_.front(); auto& front = this->write_queue_.front();
auto& is_header = front.first; auto& is_header = front.first;
auto& buf = front.second; auto& buf = front.second;
......
...@@ -150,10 +150,10 @@ public: ...@@ -150,10 +150,10 @@ public:
/// Notifies the transport that the timeout identified by `value` plus `id` /// Notifies the transport that the timeout identified by `value` plus `id`
/// was triggered. /// was triggered.
/// @param value The `atom_value` of the timeout. /// @param tag The type tag of the timeout.
/// @param id The timeout id of the timeout. /// @param id The timeout id of the timeout.
void timeout(endpoint_manager&, atom_value value, uint64_t id) { void timeout(endpoint_manager&, std::string tag, uint64_t id) {
next_layer_.timeout(*this, value, id); next_layer_.timeout(*this, std::move(tag), id);
} }
/// Callback for setting a timeout. Will be called after setting a timeout to /// Callback for setting a timeout. Will be called after setting a timeout to
......
...@@ -98,9 +98,9 @@ public: ...@@ -98,9 +98,9 @@ public:
} }
template <class Parent> template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) { void timeout(Parent& parent, std::string tag, uint64_t id) {
auto writer = make_packet_writer_decorator(*this, parent); auto writer = make_packet_writer_decorator(*this, parent);
application_.timeout(writer, value, id); application_.timeout(writer, std::move(tag), id);
} }
void handle_error(sec error) { void handle_error(sec error) {
......
...@@ -117,9 +117,9 @@ public: ...@@ -117,9 +117,9 @@ public:
} }
template <class Parent> template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) { void timeout(Parent& parent, std::string tag, uint64_t id) {
if (auto worker = workers_by_timeout_id_.at(id)) { if (auto worker = workers_by_timeout_id_.at(id)) {
worker->timeout(parent, value, id); worker->timeout(parent, std::move(tag), id);
workers_by_timeout_id_.erase(id); workers_by_timeout_id_.erase(id);
} }
} }
......
...@@ -142,7 +142,7 @@ strong_actor_ptr application::resolve_local_path(string_view path) { ...@@ -142,7 +142,7 @@ strong_actor_ptr application::resolve_local_path(string_view path) {
static constexpr string_view name_prefix = "name/"; static constexpr string_view name_prefix = "name/";
if (starts_with(path, name_prefix)) { if (starts_with(path, name_prefix)) {
path.remove_prefix(name_prefix.size()); path.remove_prefix(name_prefix.size());
atom_value name; std::string name;
if (auto err = detail::parse(path, name)) if (auto err = detail::parse(path, name))
return nullptr; return nullptr;
return system().registry().get(name); return system().registry().get(name);
......
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/net/basp/connection_state.hpp"
#include <string>
namespace caf {
namespace net {
namespace basp {
std::string to_string(connection_state x) {
switch(x) {
default:
return "???";
case connection_state::await_handshake_header:
return "await_handshake_header";
case connection_state::await_handshake_payload:
return "await_handshake_payload";
case connection_state::await_header:
return "await_header";
case connection_state::await_payload:
return "await_payload";
case connection_state::shutdown:
return "shutdown";
};
}
} // namespace basp
} // namespace net
} // namespace caf
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/net/basp/ec.hpp"
#include <string>
namespace caf {
namespace net {
namespace basp {
std::string to_string(ec x) {
switch(x) {
default:
return "???";
case ec::invalid_magic_number:
return "invalid_magic_number";
case ec::unexpected_number_of_bytes:
return "unexpected_number_of_bytes";
case ec::unexpected_payload:
return "unexpected_payload";
case ec::missing_payload:
return "missing_payload";
case ec::illegal_state:
return "illegal_state";
case ec::invalid_handshake:
return "invalid_handshake";
case ec::missing_handshake:
return "missing_handshake";
case ec::unexpected_handshake:
return "unexpected_handshake";
case ec::version_mismatch:
return "version_mismatch";
case ec::unimplemented:
return "unimplemented";
case ec::app_identifiers_mismatch:
return "app_identifiers_mismatch";
case ec::invalid_payload:
return "invalid_payload";
case ec::invalid_scheme:
return "invalid_scheme";
case ec::invalid_locator:
return "invalid_locator";
};
}
} // namespace basp
} // namespace net
} // namespace caf
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/net/basp/message_type.hpp"
#include <string>
namespace caf {
namespace net {
namespace basp {
std::string to_string(message_type x) {
switch(x) {
default:
return "???";
case message_type::handshake:
return "handshake";
case message_type::actor_message:
return "actor_message";
case message_type::resolve_request:
return "resolve_request";
case message_type::resolve_response:
return "resolve_response";
case message_type::monitor_message:
return "monitor_message";
case message_type::down_message:
return "down_message";
case message_type::heartbeat:
return "heartbeat";
};
}
} // namespace basp
} // namespace net
} // namespace caf
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/net/operation.hpp"
#include <string>
namespace caf {
namespace net {
std::string to_string(operation x) {
switch(x) {
default:
return "???";
case operation::none:
return "none";
case operation::read:
return "read";
case operation::write:
return "write";
case operation::read_write:
return "read_write";
case operation::shutdown:
return "shutdown";
};
}
} // 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/ec.hpp"
#include "caf/atom.hpp"
#include "caf/error.hpp"
#include "caf/string_view.hpp"
namespace caf::net::basp {
error make_error(ec x) {
return {static_cast<uint8_t>(x), atom("basp")};
}
} // namespace caf::net::basp
...@@ -56,8 +56,7 @@ void endpoint_manager::resolve(uri locator, actor listener) { ...@@ -56,8 +56,7 @@ void endpoint_manager::resolve(uri locator, actor listener) {
using event_type = endpoint_manager_queue::event; using event_type = endpoint_manager_queue::event;
auto ptr = new event_type(std::move(locator), listener); auto ptr = new event_type(std::move(locator), listener);
if (!enqueue(ptr)) if (!enqueue(ptr))
anon_send(listener, resolve_atom::value, anon_send(listener, resolve_atom_v, make_error(sec::request_receiver_down));
make_error(sec::request_receiver_down));
} }
void endpoint_manager::enqueue(mailbox_element_ptr msg, void endpoint_manager::enqueue(mailbox_element_ptr msg,
......
...@@ -42,8 +42,8 @@ endpoint_manager_queue::event::event(node_id observing_peer, ...@@ -42,8 +42,8 @@ endpoint_manager_queue::event::event(node_id observing_peer,
// nop // nop
} }
endpoint_manager_queue::event::event(atom_value type, uint64_t id) endpoint_manager_queue::event::event(std::string tag, uint64_t id)
: element(element_type::event), value(timeout{type, id}) { : element(element_type::event), value(timeout{std::move(tag), id}) {
// nop // nop
} }
......
...@@ -99,7 +99,7 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x, ...@@ -99,7 +99,7 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
<< " bytes"); << " bytes");
ip_endpoint ep; ip_endpoint ep;
if (auto err = detail::convert(addr, ep)) { if (auto err = detail::convert(addr, ep)) {
CAF_ASSERT(err.category() == atom("system")); CAF_ASSERT(err.category() == error_category<sec>::value);
return static_cast<sec>(err.code()); return static_cast<sec>(err.code());
} }
return std::pair<size_t, ip_endpoint>(*num_bytes, ep); return std::pair<size_t, ip_endpoint>(*num_bytes, ep);
......
...@@ -246,8 +246,8 @@ CAF_TEST(actor message) { ...@@ -246,8 +246,8 @@ CAF_TEST(actor message) {
MOCK(basp::message_type::actor_message, make_message_id().integer_value(), MOCK(basp::message_type::actor_message, make_message_id().integer_value(),
mars, actor_id{42}, self->id(), std::vector<strong_actor_ptr>{}, mars, actor_id{42}, self->id(), std::vector<strong_actor_ptr>{},
make_message("hello world!")); make_message("hello world!"));
allow((atom_value, strong_actor_ptr), allow((monitor_atom, strong_actor_ptr),
from(_).to(self).with(atom("monitor"), _)); from(_).to(self).with(monitor_atom_v, _));
expect((std::string), from(_).to(self).with("hello world!")); expect((std::string), from(_).to(self).with("hello world!"));
} }
...@@ -282,7 +282,7 @@ CAF_TEST(resolve request on id with result) { ...@@ -282,7 +282,7 @@ CAF_TEST(resolve request on id with result) {
CAF_TEST(resolve request on name with result) { CAF_TEST(resolve request on name with result) {
handle_handshake(); handle_handshake();
consume_handshake(); consume_handshake();
sys.registry().put(atom("foo"), self); sys.registry().put("foo", self);
std::string path = "name/foo"; std::string path = "name/foo";
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header); CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
MOCK(basp::message_type::resolve_request, 42, path); MOCK(basp::message_type::resolve_request, 42, path);
......
...@@ -143,8 +143,8 @@ public: ...@@ -143,8 +143,8 @@ public:
&parent.system(), &parent.system(),
cfg, cfg,
std::move(ptr)); std::move(ptr));
anon_send(listener, resolve_atom::value, anon_send(listener, resolve_atom_v, std::string{path.begin(), path.end()},
std::string{path.begin(), path.end()}, p); p);
} }
template <class Parent> template <class Parent>
...@@ -158,7 +158,7 @@ public: ...@@ -158,7 +158,7 @@ public:
} }
template <class Parent> template <class Parent>
void timeout(Parent&, atom_value, uint64_t) { void timeout(Parent&, const std::string&, uint64_t) {
// nop // nop
} }
......
...@@ -87,13 +87,13 @@ public: ...@@ -87,13 +87,13 @@ public:
template <class Parent> template <class Parent>
void resolve(Parent&, string_view path, const actor& listener) { void resolve(Parent&, string_view path, const actor& listener) {
anon_send(listener, resolve_atom::value, anon_send(listener, resolve_atom_v,
"the resolved path is still " "the resolved path is still "
+ std::string(path.begin(), path.end())); + std::string(path.begin(), path.end()));
} }
template <class Parent> template <class Parent>
void timeout(Parent&, atom_value, uint64_t) { void timeout(Parent&, const std::string&, uint64_t) {
// nop // nop
} }
......
...@@ -131,11 +131,11 @@ public: ...@@ -131,11 +131,11 @@ public:
&mgr.system(), cfg, &mgr.system(), cfg,
&mgr); &mgr);
std::string path{locator.path().begin(), locator.path().end()}; std::string path{locator.path().begin(), locator.path().end()};
anon_send(listener, resolve_atom::value, std::move(path), p); anon_send(listener, resolve_atom_v, std::move(path), p);
} }
template <class Manager> template <class Manager>
void timeout(Manager&, atom_value, uint64_t) { void timeout(Manager&, const std::string&, uint64_t) {
// nop // nop
} }
......
...@@ -55,7 +55,7 @@ struct fixture : test_coordinator_fixture<> { ...@@ -55,7 +55,7 @@ struct fixture : test_coordinator_fixture<> {
void push(int msg_id) { void push(int msg_id) {
queue.push(nullptr, static_cast<uint64_t>(msg_id), testee, queue.push(nullptr, static_cast<uint64_t>(msg_id), testee,
make_mailbox_element(self->ctrl(), make_message_id(), {}, make_mailbox_element(self->ctrl(), make_message_id(), {},
ok_atom::value, msg_id)); ok_atom_v, msg_id));
} }
}; };
......
...@@ -30,9 +30,6 @@ using namespace caf::net; ...@@ -30,9 +30,6 @@ using namespace caf::net;
namespace { namespace {
using ping_atom = caf::atom_constant<caf::atom("ping")>;
using pong_atom = caf::atom_constant<caf::atom("pong")>;
struct earth_node { struct earth_node {
uri operator()() { uri operator()() {
return unbox(make_uri("test://earth")); return unbox(make_uri("test://earth"));
...@@ -118,15 +115,15 @@ private: ...@@ -118,15 +115,15 @@ private:
behavior ping_actor(event_based_actor* self, actor pong, size_t num_pings, behavior ping_actor(event_based_actor* self, actor pong, size_t num_pings,
std::shared_ptr<size_t> count) { std::shared_ptr<size_t> count) {
CAF_MESSAGE("num_pings: " << num_pings); CAF_MESSAGE("num_pings: " << num_pings);
self->send(pong, ping_atom::value, 1); self->send(pong, ping_atom_v, 1);
return { return {
[=](pong_atom, int value) -> std::tuple<atom_value, int> { [=](pong_atom, int value) {
CAF_MESSAGE("received `pong_atom`"); CAF_MESSAGE("received `pong_atom`");
if (++*count >= num_pings) { if (++*count >= num_pings) {
CAF_MESSAGE("received " << num_pings << " pings, call self->quit"); CAF_MESSAGE("received " << num_pings << " pings, call self->quit");
self->quit(); self->quit();
} }
return std::make_tuple(ping_atom::value, value + 1); return std::make_tuple(ping_atom_v, value + 1);
}, },
}; };
} }
...@@ -138,17 +135,16 @@ behavior pong_actor(event_based_actor* self) { ...@@ -138,17 +135,16 @@ behavior pong_actor(event_based_actor* self) {
self->quit(dm.reason); self->quit(dm.reason);
}); });
return { return {
[=](ping_atom, int value) -> std::tuple<atom_value, int> { [=](ping_atom, int value) {
CAF_MESSAGE("received `ping_atom` from " << self->current_sender()); CAF_MESSAGE("received `ping_atom` from " << self->current_sender());
if (self->current_sender() == self->ctrl()) if (self->current_sender() == self->ctrl())
abort(); abort();
self->monitor(self->current_sender()); self->monitor(self->current_sender());
// set next behavior // set next behavior
self->become([](ping_atom, int val) { self->become(
return std::make_tuple(pong_atom::value, val); [](ping_atom, int val) { return std::make_tuple(pong_atom_v, val); });
});
// reply to 'ping' // reply to 'ping'
return std::make_tuple(pong_atom::value, value); return std::make_tuple(pong_atom_v, value);
}, },
}; };
} }
...@@ -188,7 +184,7 @@ CAF_TEST_FIXTURE_SCOPE(ping_pong_tests, fixture) ...@@ -188,7 +184,7 @@ CAF_TEST_FIXTURE_SCOPE(ping_pong_tests, fixture)
CAF_TEST(full setup) { CAF_TEST(full setup) {
auto pong = earth.sys.spawn(pong_actor); auto pong = earth.sys.spawn(pong_actor);
run(); run();
earth.sys.registry().put(atom("pong"), pong); earth.sys.registry().put("pong", pong);
auto remote_pong = mars.resolve("test://earth/name/pong"); auto remote_pong = mars.resolve("test://earth/name/pong");
auto count = std::make_shared<size_t>(0); auto count = std::make_shared<size_t>(0);
auto ping = mars.sys.spawn(ping_actor, remote_pong, 10, count); auto ping = mars.sys.spawn(ping_actor, remote_pong, 10, count);
......
...@@ -111,7 +111,7 @@ CAF_TEST(deliver serialized message) { ...@@ -111,7 +111,7 @@ CAF_TEST(deliver serialized message) {
std::vector<strong_actor_ptr> stages; std::vector<strong_actor_ptr> stages;
binary_serializer sink{sys, payload}; binary_serializer sink{sys, payload};
if (auto err = sink(node_id{}, self->id(), testee.id(), stages, if (auto err = sink(node_id{}, self->id(), testee.id(), stages,
make_message(ok_atom::value))) make_message(ok_atom_v)))
CAF_FAIL("unable to serialize message: " << sys.render(err)); CAF_FAIL("unable to serialize message: " << sys.render(err));
net::basp::header hdr{net::basp::message_type::actor_message, net::basp::header hdr{net::basp::message_type::actor_message,
static_cast<uint32_t>(payload.size()), static_cast<uint32_t>(payload.size()),
......
...@@ -114,12 +114,12 @@ public: ...@@ -114,12 +114,12 @@ public:
&parent.system(), &parent.system(),
cfg, cfg,
std::move(ptr)); std::move(ptr));
anon_send(listener, resolve_atom::value, anon_send(listener, resolve_atom_v, std::string{path.begin(), path.end()},
std::string{path.begin(), path.end()}, p); p);
} }
template <class Parent> template <class Parent>
void timeout(Parent&, atom_value, uint64_t) { void timeout(Parent&, const std::string&, uint64_t) {
// nop // nop
} }
......
...@@ -175,12 +175,12 @@ public: ...@@ -175,12 +175,12 @@ public:
auto mgr = &parent.manager(); auto mgr = &parent.manager();
auto p = make_actor<net::actor_proxy_impl, strong_actor_ptr>(aid, nid, sys, auto p = make_actor<net::actor_proxy_impl, strong_actor_ptr>(aid, nid, sys,
cfg, mgr); cfg, mgr);
anon_send(listener, resolve_atom::value, anon_send(listener, resolve_atom_v, std::string{path.begin(), path.end()},
std::string{path.begin(), path.end()}, p); p);
} }
template <class Parent> template <class Parent>
void timeout(Parent&, atom_value, uint64_t) { void timeout(Parent&, const std::string&, uint64_t) {
// nop // nop
} }
......
...@@ -46,7 +46,7 @@ struct application_result { ...@@ -46,7 +46,7 @@ struct application_result {
std::vector<byte> data_buffer; std::vector<byte> data_buffer;
std::string resolve_path; std::string resolve_path;
actor resolve_listener; actor resolve_listener;
atom_value timeout_value; std::string timeout_value;
uint64_t timeout_id; uint64_t timeout_id;
sec err; sec err;
}; };
...@@ -93,8 +93,8 @@ public: ...@@ -93,8 +93,8 @@ public:
} }
template <class Parent> template <class Parent>
void timeout(Parent&, atom_value value, uint64_t id) { void timeout(Parent&, std::string value, uint64_t id) {
res_->timeout_value = value; res_->timeout_value = std::move(value);
res_->timeout_id = id; res_->timeout_id = id;
} }
...@@ -221,8 +221,8 @@ CAF_TEST(resolve) { ...@@ -221,8 +221,8 @@ CAF_TEST(resolve) {
} }
CAF_TEST(timeout) { CAF_TEST(timeout) {
worker.timeout(transport, atom("bar"), 42u); worker.timeout(transport, "bar", 42u);
CAF_CHECK_EQUAL(application_results->timeout_value, atom("bar")); CAF_CHECK_EQUAL(application_results->timeout_value, "bar");
CAF_CHECK_EQUAL(application_results->timeout_id, 42u); CAF_CHECK_EQUAL(application_results->timeout_id, 42u);
} }
......
...@@ -84,7 +84,7 @@ public: ...@@ -84,7 +84,7 @@ public:
} }
template <class Transport> template <class Transport>
void timeout(Transport&, atom_value, uint64_t) { void timeout(Transport&, const std::string&, uint64_t) {
rec_buf_->push_back(static_cast<byte>(id_)); rec_buf_->push_back(static_cast<byte>(id_));
} }
...@@ -263,7 +263,7 @@ struct fixture : host_fixture { ...@@ -263,7 +263,7 @@ struct fixture : host_fixture {
#define CHECK_TIMEOUT(testcase) \ #define CHECK_TIMEOUT(testcase) \
dispatcher.set_timeout(1u, testcase.ep); \ dispatcher.set_timeout(1u, testcase.ep); \
dispatcher.timeout(dummy, atom("dummy"), 1u); \ dispatcher.timeout(dummy, "dummy", 1u); \
CAF_CHECK_EQUAL(buf->size(), 1u); \ CAF_CHECK_EQUAL(buf->size(), 1u); \
CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \ CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \
buf->clear(); buf->clear();
......
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