Commit 88196f3c authored by Jakob Otto's avatar Jakob Otto

Implement changes requested in PR

parent 9bb8fb8c
...@@ -41,17 +41,17 @@ public: ...@@ -41,17 +41,17 @@ public:
using config = std::pair<receive_policy_flag, size_t>; using config = std::pair<receive_policy_flag, size_t>;
static inline config at_least(size_t num_bytes) { static config at_least(size_t num_bytes) {
CAF_ASSERT(num_bytes > 0); CAF_ASSERT(num_bytes > 0);
return {receive_policy_flag::at_least, num_bytes}; return {receive_policy_flag::at_least, num_bytes};
} }
static inline config at_most(size_t num_bytes) { static config at_most(size_t num_bytes) {
CAF_ASSERT(num_bytes > 0); CAF_ASSERT(num_bytes > 0);
return {receive_policy_flag::at_most, num_bytes}; return {receive_policy_flag::at_most, num_bytes};
} }
static inline config exactly(size_t num_bytes) { static config exactly(size_t num_bytes) {
CAF_ASSERT(num_bytes > 0); CAF_ASSERT(num_bytes > 0);
return {receive_policy_flag::exactly, num_bytes}; return {receive_policy_flag::exactly, num_bytes};
} }
......
...@@ -78,7 +78,7 @@ public: ...@@ -78,7 +78,7 @@ public:
/// Called when the remote side becomes unreachable due to an error. /// Called when the remote side becomes unreachable due to an error.
/// @param reason The error code as reported by the operating system. /// @param reason The error code as reported by the operating system.
virtual void handle_error(caf::sec code) = 0; virtual void handle_error(sec code) = 0;
protected: protected:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
......
...@@ -28,24 +28,15 @@ ...@@ -28,24 +28,15 @@
#include <caf/span.hpp> #include <caf/span.hpp>
#include <caf/variant.hpp> #include <caf/variant.hpp>
#ifdef CAF_WINDOWS
# include <winsock2.h>
#else
# include <caf/actor.hpp>
# include <sys/socket.h>
# include <sys/types.h>
#endif
namespace caf { namespace caf {
namespace policy { namespace policy {
/// Implements a scribe policy that manages a stream-socket.
class scribe { class scribe {
public: public:
explicit scribe(net::stream_socket handle); explicit scribe(net::stream_socket handle);
net::stream_socket handle_; net::stream_socket handle() const noexcept {
net::stream_socket handle() {
return handle_; return handle_;
} }
...@@ -61,7 +52,8 @@ public: ...@@ -61,7 +52,8 @@ public:
void* buf = read_buf_.data() + collected_; void* 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));
variant<size_t, sec> ret = read(handle_, buf, len); auto ret = read(handle_, buf, len);
// Update state.
if (auto num_bytes = get_if<size_t>(&ret)) { if (auto num_bytes = get_if<size_t>(&ret)) {
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;
...@@ -71,24 +63,23 @@ public: ...@@ -71,24 +63,23 @@ public:
} }
return true; return true;
} else { } else {
// Make sure WSAGetLastError gets called immediately on Windows.
auto err = get<sec>(ret); auto err = get<sec>(ret);
CAF_LOG_DEBUG("receive failed" << CAF_ARG(err)); CAF_LOG_DEBUG("receive failed" << CAF_ARG(err));
handle_error(parent, err); parent.application().handle_error(err);
return false; return false;
} }
} }
template <class Parent> template <class Parent>
bool handle_write_event(Parent& parent) { bool handle_write_event(Parent& parent) {
// try to write leftover data! // Try to write leftover data.
write_some(parent); write_some(parent);
// get new data from parent // Get new data from parent.
for (auto msg = parent.next_message(); msg != nullptr; for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) { msg = parent.next_message()) {
parent.application().write_message(*this, std::move(msg)); parent.application().write_message(*this, std::move(msg));
} }
// write prepared data // Write prepared data.
return write_some(parent); return write_some(parent);
} }
...@@ -102,32 +93,30 @@ public: ...@@ -102,32 +93,30 @@ public:
auto ret = net::write(handle_, buf, len); auto ret = net::write(handle_, buf, len);
if (auto num_bytes = get_if<size_t>(&ret)) { if (auto num_bytes = get_if<size_t>(&ret)) {
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));
// update state // Update state.
written_ += *num_bytes; written_ += *num_bytes;
if (written_ >= write_buf_.size()) { if (written_ >= write_buf_.size()) {
written_ = 0; written_ = 0;
write_buf_.clear(); write_buf_.clear();
return false; return false;
} else {
return true;
} }
} else { } else {
CAF_LOG_ERROR("send failed"); auto err = get<sec>(ret);
handle_error(parent, get<sec>(ret)); CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
parent.application().handle_error(err);
return false; return false;
} }
return true;
} }
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) {
parent.application().resolve(parent, path, listener); parent.application().resolve(parent, path, listener);
// TODO should parent be passed as well?
} }
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) {
parent.application().timeout(*this, value, id); parent.application().timeout(*this, value, id);
// TODO should parent be passed as well?
} }
template <class Application> template <class Application>
...@@ -142,6 +131,8 @@ public: ...@@ -142,6 +131,8 @@ public:
void write_packet(span<char> buf); void write_packet(span<char> buf);
private: private:
net::stream_socket handle_;
std::vector<char> read_buf_; std::vector<char> read_buf_;
std::vector<char> write_buf_; std::vector<char> write_buf_;
......
...@@ -18,8 +18,7 @@ ...@@ -18,8 +18,7 @@
#define CAF_SUITE scribe_policy #define CAF_SUITE scribe_policy
#include "caf/net/endpoint_manager.hpp" #include "caf/policy/scribe.hpp"
#include <caf/policy/scribe.hpp>
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
...@@ -29,6 +28,7 @@ ...@@ -29,6 +28,7 @@
#include "caf/detail/scope_guard.hpp" #include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp" #include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
...@@ -38,9 +38,7 @@ using namespace caf::net; ...@@ -38,9 +38,7 @@ using namespace caf::net;
namespace { namespace {
string_view hello_manager{"hello manager!"}; constexpr string_view hello_manager = "hello manager!";
string_view hello_test{"hello test!"};
struct fixture : test_coordinator_fixture<>, host_fixture { struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() { fixture() {
...@@ -130,7 +128,6 @@ CAF_TEST(receive) { ...@@ -130,7 +128,6 @@ CAF_TEST(receive) {
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()), CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()),
sec::unavailable_or_would_block); sec::unavailable_or_would_block);
auto guard = detail::make_scope_guard([&] { close(sockets.second); }); auto guard = detail::make_scope_guard([&] { close(sockets.second); });
CAF_MESSAGE("configure scribe_policy"); CAF_MESSAGE("configure scribe_policy");
policy::scribe scribe{sockets.first}; policy::scribe scribe{sockets.first};
scribe.configure_read(net::receive_policy::exactly(hello_manager.size())); scribe.configure_read(net::receive_policy::exactly(hello_manager.size()));
...@@ -138,7 +135,6 @@ CAF_TEST(receive) { ...@@ -138,7 +135,6 @@ CAF_TEST(receive) {
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates(); mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_MESSAGE("sending data to scribe_policy"); CAF_MESSAGE("sending data to scribe_policy");
CAF_CHECK_EQUAL(write(sockets.second, hello_manager.data(), CAF_CHECK_EQUAL(write(sockets.second, hello_manager.data(),
hello_manager.size()), hello_manager.size()),
...@@ -169,10 +165,8 @@ CAF_TEST(resolve and proxy communication) { ...@@ -169,10 +165,8 @@ CAF_TEST(resolve and proxy communication) {
[&] { CAF_FAIL("manager did not respond with a proxy."); }); [&] { CAF_FAIL("manager did not respond with a proxy."); });
run(); run();
auto read_res = read(sockets.second, read_buf.data(), read_buf.size()); auto read_res = read(sockets.second, read_buf.data(), read_buf.size());
if (!holds_alternative<size_t>(read_res)) { if (!holds_alternative<size_t>(read_res))
CAF_ERROR("read() returned an error: " << sys.render(get<sec>(read_res))); CAF_FAIL("read() returned an error: " << sys.render(get<sec>(read_res)));
return;
}
read_buf.resize(get<size_t>(read_res)); read_buf.resize(get<size_t>(read_res));
CAF_MESSAGE("receive buffer contains " << read_buf.size() << " bytes"); CAF_MESSAGE("receive buffer contains " << read_buf.size() << " bytes");
message msg; message msg;
......
...@@ -192,7 +192,6 @@ CAF_TEST(receive) { ...@@ -192,7 +192,6 @@ CAF_TEST(receive) {
nonblocking(sockets.second, true); nonblocking(sockets.second, true);
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()), CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()),
sec::unavailable_or_would_block); sec::unavailable_or_would_block);
CAF_MESSAGE("adding both endpoint managers"); CAF_MESSAGE("adding both endpoint managers");
auto mgr1 = make_endpoint_manager(mpx, sys, policy::scribe{sockets.first}, auto mgr1 = make_endpoint_manager(mpx, sys, policy::scribe{sockets.first},
stream_string_application{sys, buf}); stream_string_application{sys, buf});
...@@ -204,7 +203,6 @@ CAF_TEST(receive) { ...@@ -204,7 +203,6 @@ CAF_TEST(receive) {
CAF_CHECK_EQUAL(mgr2->init(), none); CAF_CHECK_EQUAL(mgr2->init(), none);
mpx->handle_updates(); mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u);
CAF_MESSAGE("resolve actor-proxy"); CAF_MESSAGE("resolve actor-proxy");
mgr1->resolve("/id/42", self); mgr1->resolve("/id/42", self);
run(); run();
......
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