Commit 9bb8fb8c authored by Jakob Otto's avatar Jakob Otto

Add string_application for testing purposes

parent bd0529b4
......@@ -19,7 +19,7 @@ set(LIBCAF_NET_SRCS
src/socket_manager.cpp
src/stream_socket.cpp
src/scribe.cpp
)
)
add_custom_target(libcaf_net)
......
......@@ -51,7 +51,7 @@ public:
template <class Parent>
error init(Parent& parent) {
prepare_next_read();
parent.application().init(parent);
parent.mask_add(net::operation::read);
return none;
}
......@@ -68,10 +68,8 @@ public:
if (collected_ >= read_threshold_) {
parent.application().handle_data(*this, read_buf_);
prepare_next_read();
return false;
} else {
return true;
}
return true;
} else {
// Make sure WSAGetLastError gets called immediately on Windows.
auto err = get<sec>(ret);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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 <caf/actor_config.hpp>
#include <caf/binary_serializer.hpp>
#include <caf/expected.hpp>
#include <caf/logger.hpp>
#include <caf/make_actor.hpp>
#include <caf/span.hpp>
#include <caf/test/dsl.hpp>
#include <netinet/in.h>
#include <vector>
namespace caf {
namespace policy {
struct string_application_header {
uint32_t payload;
};
/// @relates header
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f,
string_application_header& hdr) {
return f(meta::type_name("sa_header"), hdr.payload);
}
class string_application {
public:
using header_type = string_application_header;
template <class Parent>
error init(Parent&) {
return none;
}
template <class Parent>
void handle_packet(Parent&, header_type&, span<char>) {
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) {
header_type header{static_cast<uint32_t>(msg->payload.size())};
std::vector<char> data;
binary_serializer serializer{nullptr, data};
serializer(header);
serializer(msg->payload);
parent.write_packet(data);
}
static expected<std::vector<char>> serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<char> result;
binary_serializer sink{sys, result};
if (auto err = message::save(sink, x))
return err;
return result;
}
private:
std::vector<char> buf_;
};
class stream_string_application : public string_application {
public:
template <class Parent>
error init(Parent& parent) {
parent.transport().configure_read_size(sizeof(header_type));
return string_application::init(parent);
}
template <class Parent>
void handle_data(Parent& parent, span<char> data) {
if (await_payload_) {
handle_packet(parent, header_, data);
} else {
if (header_.payload == 0)
handle_packet(parent, header_, span<char>{});
else
parent.configure_read(net::receive_policy::exactly(header_.payload));
}
}
template <class Manager>
void resolve(Manager&, const std::string&, actor) {
actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(aid, hid));
actor_config cfg;
// TODO!
/*auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid,
&manager.system(),
cfg, &manager);
anon_send(listener, resolve_atom::value, std::move(path), p);*/
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
// nop
}
void handle_error(sec) {
// nop
}
private:
header_type header_;
bool await_payload_;
};
} // namespace policy
} // namespace caf
......@@ -37,6 +37,7 @@ scribe::scribe(caf::net::stream_socket handle)
}
void scribe::prepare_next_read() {
read_buf_.clear();
collected_ = 0;
// This cast does nothing, but prevents a weird compiler error on GCC <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
......
......@@ -20,7 +20,6 @@
#include "caf/net/endpoint_manager.hpp"
#include <caf/policy/scribe.hpp>
#include <caf/policy/string_application.hpp>
#include "caf/test/dsl.hpp"
......@@ -67,6 +66,11 @@ public:
~dummy_application() = default;
template <class Parent>
error init(Parent&) {
return none;
}
template <class Transport>
void write_message(Transport& transport,
std::unique_ptr<endpoint_manager::message> msg) {
......
......@@ -20,12 +20,12 @@
#include "caf/net/endpoint_manager.hpp"
#include <caf/policy/scribe.hpp>
#include <caf/policy/string_application.hpp>
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
......@@ -59,22 +59,154 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
multiplexer_ptr mpx;
};
struct string_application_header {
uint32_t payload;
};
/// @relates header
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f,
string_application_header& hdr) {
return f(meta::type_name("sa_header"), hdr.payload);
}
class string_application {
public:
using header_type = string_application_header;
string_application(actor_system& sys, std::shared_ptr<std::vector<char>> buf)
: sys_(sys), buf_(std::move(buf)) {
// nop
}
template <class Parent>
error init(Parent&) {
return none;
}
template <class Parent>
void handle_packet(Parent&, header_type&, span<char> payload) {
binary_deserializer source{sys_, payload.data(), payload.size()};
message msg;
if (auto err = msg.load(source))
CAF_FAIL("unable to deserialize message: " << err);
if (!msg.match_elements<std::string>())
CAF_FAIL("unexpected message: " << msg);
auto& str = msg.get_as<std::string>(0);
buf_->insert(buf_->end(), str.begin(), str.end());
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) {
std::vector<char> buf;
header_type header{static_cast<uint32_t>(msg->payload.size())};
buf.resize(sizeof(header_type));
memcpy(buf.data(), &header, buf.size());
buf.insert(buf.end(), msg->payload.begin(), msg->payload.end());
parent.write_packet(buf);
}
static expected<std::vector<char>> serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<char> result;
binary_serializer sink{sys, result};
if (auto err = message::save(sink, x))
return err;
return result;
}
private:
actor_system& sys_;
std::shared_ptr<std::vector<char>> buf_;
};
class stream_string_application : public string_application {
public:
stream_string_application(actor_system& sys,
std::shared_ptr<std::vector<char>> buf)
: string_application(sys, std::move(buf)), await_payload_(false) {
// nop
}
template <class Parent>
error init(Parent& parent) {
parent.transport().configure_read(
net::receive_policy::exactly(sizeof(header_type)));
return string_application::init(parent);
}
template <class Parent>
void handle_data(Parent& parent, span<char> data) {
if (await_payload_) {
handle_packet(parent, header_, data);
await_payload_ = false;
} else {
if (data.size() != sizeof(header_type))
CAF_FAIL("");
memcpy(&header_, data.data(), sizeof(header_type));
if (header_.payload == 0)
handle_packet(parent, header_, span<char>{});
else
parent.configure_read(net::receive_policy::exactly(header_.payload));
await_payload_ = true;
}
}
template <class Manager>
void resolve(Manager& manager, const std::string& path, actor listener) {
actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(aid, hid));
actor_config cfg;
auto p = make_actor<net::actor_proxy_impl, strong_actor_ptr>(aid, nid,
&manager
.system(),
cfg, &manager);
anon_send(listener, resolve_atom::value, std::move(path), p);
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
// nop
}
void handle_error(sec) {
// nop
}
private:
header_type header_;
bool await_payload_;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
/*
CAF_TEST(resolve and proxy communication) {
CAF_TEST(receive) {
std::vector<char> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<char>>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys, policy::scribe{sockets.first},
stream_string_application{});
CAF_CHECK_EQUAL(mgr->init(), none);
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});
CAF_CHECK_EQUAL(mgr1->init(), none);
mpx->handle_updates();
run();
mgr->resolve("/id/42", self);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
auto mgr2 = make_endpoint_manager(mpx, sys, policy::scribe{sockets.second},
stream_string_application{sys, buf});
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();
self->receive(
[&](resolve_atom, const std::string&, const strong_actor_ptr& p) {
......@@ -84,20 +216,7 @@ CAF_TEST(resolve and proxy communication) {
after(std::chrono::seconds(0)) >>
[&] { 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;
}
read_buf.resize(get<size_t>(read_res));
CAF_MESSAGE("receive buffer contains " << read_buf.size() << " bytes");
message msg;
binary_deserializer source{sys, read_buf};
CAF_CHECK_EQUAL(source(msg), none);
if (msg.match_elements<std::string>())
CAF_CHECK_EQUAL(msg.get_as<std::string>(0), "hello proxy!");
else
CAF_ERROR("expected a string, got: " << to_string(msg));
CAF_CHECK_EQUAL(string_view(buf->data(), buf->size()), "hello proxy!");
}
*/
CAF_TEST_FIXTURE_SCOPE_END()
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