Commit ec98978c authored by Jakob Otto's avatar Jakob Otto

Implement changes requested in PR

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