Commit 43f526d1 authored by Dominik Charousset's avatar Dominik Charousset

Revise intra-layer communication API

parent e1473d7f
......@@ -54,16 +54,16 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
/// from event queues, etc.
/// @returns `true` if the lower layers may proceed, `false` otherwise
/// (aborts execution).
template <class LowerLayer>
bool prepare_send(LowerLayer& down);
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down);
/// Called whenever the underlying transport finished writing all buffered
/// data for output to query whether an upper layer still has pending events
/// or may produce output data on the next call to `prepare_send`.
/// @returns `true` if the underlying socket may get removed from the I/O
/// event loop, `false` otherwise.
template <class LowerLayer>
bool done_sending(LowerLayer& down);
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr down);
}
interface base [role: lower layer] {
......@@ -73,8 +73,8 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
interface stream_oriented [role: upper layer] {
/// Called by the lower layer for cleaning up any state in case of an error.
template <class LowerLayer>
void abort(LowerLayer& down, const error& reason);
template <class LowerLayerPtr>
void abort(LowerLayerPtr down, const error& reason);
/// Consumes bytes from the lower layer.
/// @param down Reference to the lower layer that received the data.
......@@ -84,8 +84,8 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
/// input or negative to signal an error.
/// @note When returning a negative value, clients should also call
/// `down.abort_reason(...)` with an appropriate error code.
template <class LowerLayer>
ptrdiff_t consume(LowerLayer& down, byte_span buffer, byte_span delta);
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span buffer, byte_span delta);
}
interface stream_oriented [role: lower layer] {
......@@ -109,8 +109,11 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
/// Propagates an abort reason to the lower layers. After processing the
/// current read or write event, the lowest layer will call `abort` on its
// upper layer.
/// upper layer.
void abort_reason(error reason);
/// Returns the last recent abort reason or `none` if no error occurred.
const error& abort_reason();
}
interface datagram_oriented [role: upper layer] {
......@@ -123,8 +126,8 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
/// @note When returning a negative value for the number of consumed bytes,
/// clients must also call `down.set_read_error(...)` with an
/// appropriate error code.
template <class LowerLayer>
ptrdiff_t consume(LowerLayer& down, byte_span buffer);
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span buffer);
}
interface datagram_oriented [role: lower layer] {
......@@ -153,8 +156,8 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
/// @note When returning a negative value for the number of consumed bytes,
/// clients must also call `down.set_read_error(...)` with an
/// appropriate error code.
template <class LowerLayer>
ptrdiff_t consume(LowerLayer& down, byte_span buffer);
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span buffer);
}
interface message_oriented [role: lower layer] {
......@@ -172,7 +175,7 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
/// Seals and prepares a message for transfer.
/// @note When returning `false`, clients must also call
/// `down.set_read_error(...)` with an appropriate error code.
template <class LowerLayer>
template <class LowerLayerPtr>
bool end_message();
}
......@@ -141,22 +141,22 @@ public:
// -- initialization ---------------------------------------------------------
error init(const settings& config) override {
return protocol_.init(static_cast<socket_manager*>(this), *this, config);
return protocol_.init(static_cast<socket_manager*>(this), this, config);
}
// -- event callbacks --------------------------------------------------------
bool handle_read_event() override {
return protocol_.handle_read_event(*this);
return protocol_.handle_read_event(this);
}
bool handle_write_event() override {
return protocol_.handle_write_event(*this);
return protocol_.handle_write_event(this);
}
void handle_error(sec code) override {
abort_reason_ = code;
return protocol_.abort(*this, abort_reason_);
return protocol_.abort(this, abort_reason_);
}
auto& protocol() noexcept {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
namespace caf::net {
/// Wraps a pointer to a stream-oriented layer with a pointer to its lower
/// layer. Both pointers are then used to implement the interface required for a
/// stream-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class stream_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
void begin_output() {
lptr_->begin_output(llptr_);
}
byte_buffer& output_buffer() {
return lptr_->output_buffer(llptr_);
}
void end_output() {
lptr_->end_output(llptr_);
}
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
void configure_read(receive_policy policy) {
lptr_->configure_read(llptr_, policy);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
stream_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
stream_oriented_layer_ptr(const stream_oriented_layer_ptr&) = default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_stream_oriented_layer_ptr(Layer* this_layer, LowerLayerPtr down) {
return stream_oriented_layer_ptr<Layer, LowerLayerPtr>{this_layer, down};
}
} // namespace caf::net
......@@ -26,6 +26,7 @@
#include "caf/logger.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_oriented_layer_ptr.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/sec.hpp"
#include "caf/settings.hpp"
......@@ -55,53 +56,48 @@ public:
// nop
}
// -- interface for the upper layer ------------------------------------------
// -- interface for stream_oriented_layer_ptr --------------------------------
template <class Parent>
class access {
public:
access(Parent* parent, stream_transport* transport)
: parent_(parent), transport_(transport) {
// nop
template <class ParentPtr>
bool can_send_more(ParentPtr) const noexcept {
return write_buf_.size() < max_write_buf_size_;
}
void begin_output() {
if (transport_->write_buf_.empty())
parent_->register_writing();
template <class ParentPtr>
void begin_output(ParentPtr parent) {
if (write_buf_.empty())
parent->register_writing();
}
byte_buffer& output_buffer() {
return transport_->write_buf_;
template <class ParentPtr>
byte_buffer& output_buffer(ParentPtr) {
return write_buf_;
}
constexpr void end_output() {
template <class ParentPtr>
constexpr void end_output(ParentPtr) {
// nop
}
bool can_send_more() const noexcept {
return output_buffer().size() < transport_->max_write_buf_size_;
template <class ParentPtr>
void abort_reason(ParentPtr parent, error reason) {
return parent->abort_reason(std::move(reason));
}
void abort_reason(error reason) {
return parent_->abort_reason(std::move(reason));
template <class ParentPtr>
const error& abort_reason(ParentPtr parent) {
return parent->abort_reason();
}
void configure_read(receive_policy policy) {
if (policy.max_size > 0 && transport_->max_read_size_ == 0)
parent_->register_reading();
transport_->min_read_size_ = policy.min_size;
transport_->max_read_size_ = policy.max_size;
transport_->read_buf_.resize(policy.max_size);
template <class ParentPtr>
void configure_read(ParentPtr parent, receive_policy policy) {
if (policy.max_size > 0 && max_read_size_ == 0)
parent->register_reading();
min_read_size_ = policy.min_size;
max_read_size_ = policy.max_size;
read_buf_.resize(policy.max_size);
}
private:
Parent* parent_;
stream_transport* transport_;
};
template <class Parent>
friend class access;
// -- properties -------------------------------------------------------------
auto& read_buffer() noexcept {
......@@ -130,8 +126,8 @@ public:
// -- initialization ---------------------------------------------------------
template <class Parent>
error init(socket_manager* owner, Parent& parent, const settings& config) {
template <class ParentPtr>
error init(socket_manager* owner, ParentPtr parent, const settings& config) {
namespace mm = defaults::middleman;
auto default_max_reads = static_cast<uint32_t>(mm::max_consecutive_reads);
max_consecutive_reads_ = get_or(
......@@ -144,7 +140,7 @@ public:
return err;
}*/
if (auto socket_buf_size
= send_buffer_size(parent.template handle<network_socket>())) {
= send_buffer_size(parent->template handle<network_socket>())) {
max_write_buf_size_ = *socket_buf_size;
CAF_ASSERT(max_write_buf_size_ > 0);
write_buf_.reserve(max_write_buf_size_ * 2);
......@@ -152,31 +148,31 @@ public:
CAF_LOG_ERROR("send_buffer_size: " << socket_buf_size.error());
return std::move(socket_buf_size.error());
}
access<Parent> this_layer{&parent, this};
return upper_layer_.init(owner, this_layer, config);
auto this_layer_ptr = make_stream_oriented_layer_ptr(this, parent);
return upper_layer_.init(owner, this_layer_ptr, config);
}
// -- event callbacks --------------------------------------------------------
template <class Parent>
bool handle_read_event(Parent& parent) {
CAF_LOG_TRACE(CAF_ARG2("handle", parent.handle().id));
auto fail = [this, &parent](auto reason) {
template <class ParentPtr>
bool handle_read_event(ParentPtr parent) {
CAF_LOG_TRACE(CAF_ARG2("handle", parent->handle().id));
auto fail = [this, parent](auto reason) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(reason));
parent.abort_reason(std::move(reason));
access<Parent> this_layer{&parent, this};
upper_layer_.abort(this_layer, parent.abort_reason());
parent->abort_reason(std::move(reason));
auto this_layer_ptr = make_stream_oriented_layer_ptr(this, parent);
upper_layer_.abort(this_layer_ptr, parent->abort_reason());
return false;
};
access<Parent> this_layer{&parent, this};
auto this_layer_ptr = make_stream_oriented_layer_ptr(this, parent);
for (size_t i = 0; max_read_size_ > 0 && i < max_consecutive_reads_; ++i) {
CAF_ASSERT(max_read_size_ > read_size_);
auto buf = read_buf_.data() + read_size_;
size_t len = max_read_size_ - read_size_;
CAF_LOG_DEBUG(CAF_ARG2("missing", len));
auto num_bytes = read(parent.template handle<socket_type>(),
auto num_bytes = read(parent->template handle<socket_type>(),
make_span(buf, len));
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG2("handle", parent.handle().id)
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG2("handle", parent->handle().id)
<< CAF_ARG(num_bytes));
// Update state.
if (num_bytes > 0) {
......@@ -185,14 +181,14 @@ public:
continue;
auto delta = make_span(read_buf_.data() + delta_offset_,
read_size_ - delta_offset_);
auto consumed = upper_layer_.consume(this_layer, make_span(read_buf_),
delta);
auto consumed = upper_layer_.consume(this_layer_ptr,
make_span(read_buf_), delta);
if (consumed > 0) {
read_buf_.erase(read_buf_.begin(), read_buf_.begin() + consumed);
read_size_ -= consumed;
} else if (consumed < 0) {
upper_layer_.abort(this_layer,
parent.abort_reason_or(caf::sec::runtime_error));
upper_layer_.abort(this_layer_ptr,
parent->abort_reason_or(caf::sec::runtime_error));
return false;
}
delta_offset_ = read_size_;
......@@ -212,29 +208,29 @@ public:
return max_read_size_ > 0;
}
template <class Parent>
bool handle_write_event(Parent& parent) {
CAF_LOG_TRACE(CAF_ARG2("handle", parent.handle().id));
auto fail = [this, &parent](sec reason) {
template <class ParentPtr>
bool handle_write_event(ParentPtr parent) {
CAF_LOG_TRACE(CAF_ARG2("handle", parent->handle().id));
auto fail = [this, parent](sec reason) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(reason));
parent.abort_reason(reason);
access<Parent> this_layer{&parent, this};
upper_layer_.abort(this_layer, reason);
parent->abort_reason(reason);
auto this_layer_ptr = make_stream_oriented_layer_ptr(this, parent);
upper_layer_.abort(this_layer_ptr, reason);
return false;
};
// Allow the upper layer to add extra data to the write buffer.
access<Parent> this_layer{&parent, this};
if (!upper_layer_.prepare_send(this_layer)) {
upper_layer_.abort(this_layer,
parent.abort_reason_or(caf::sec::runtime_error));
auto this_layer_ptr = make_stream_oriented_layer_ptr(this, parent);
if (!upper_layer_.prepare_send(this_layer_ptr)) {
upper_layer_.abort(this_layer_ptr,
parent->abort_reason_or(caf::sec::runtime_error));
return false;
}
if (write_buf_.empty())
return !upper_layer_.done_sending(this_layer);
auto written = write(parent.template handle<socket_type>(), write_buf_);
return !upper_layer_.done_sending(this_layer_ptr);
auto written = write(parent->template handle<socket_type>(), write_buf_);
if (written > 0) {
write_buf_.erase(write_buf_.begin(), write_buf_.begin() + written);
return !write_buf_.empty() || !upper_layer_.done_sending(this_layer);
return !write_buf_.empty() || !upper_layer_.done_sending(this_layer_ptr);
} else if (written < 0) {
// Try again later on temporary errors such as EWOULDBLOCK and
// stop writing to the socket on hard errors.
......@@ -248,10 +244,10 @@ public:
}
}
template <class Parent>
void abort(Parent& parent, const error& reason) {
access<Parent> this_layer{&parent, this};
upper_layer_.abort(this_layer, reason);
template <class ParentPtr>
void abort(ParentPtr parent, const error& reason) {
auto this_layer_ptr = make_stream_oriented_layer_ptr(this, parent);
upper_layer_.abort(this_layer_ptr, reason);
}
private:
......
......@@ -86,34 +86,35 @@ public:
// -- initialization ---------------------------------------------------------
template <class LowerLayer>
error init(socket_manager* owner, LowerLayer& down, const settings& config) {
template <class LowerLayerPtr>
error
init(socket_manager* owner, LowerLayerPtr down, const settings& config) {
owner_ = owner;
cfg_ = config;
down.configure_read(net::receive_policy::up_to(max_header_size));
down->configure_read(net::receive_policy::up_to(max_header_size));
return none;
}
// -- role: upper layer ------------------------------------------------------
template <class LowerLayer>
bool prepare_send(LowerLayer& down) {
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
return handshake_complete_ ? upper_layer_.prepare_send(down) : true;
}
template <class LowerLayer>
bool done_sending(LowerLayer& down) {
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr down) {
return handshake_complete_ ? upper_layer_.done_sending(down) : true;
}
template <class LowerLayer>
void abort(LowerLayer& down, const error& reason) {
template <class LowerLayerPtr>
void abort(LowerLayerPtr down, const error& reason) {
if (handshake_complete_)
upper_layer_.abort(down, reason);
}
template <class LowerLayer>
ptrdiff_t consume(LowerLayer& down, byte_span buffer, byte_span delta) {
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span buffer, byte_span delta) {
if (handshake_complete_)
return upper_layer_.consume(down, buffer, delta);
// TODO: we could avoid repeated scans by using the delta parameter.
......@@ -124,7 +125,7 @@ public:
write(down, header_too_large);
auto err = make_error(pec::too_many_characters,
"exceeded maximum header size");
down.abort_reason(std::move(err));
down->abort_reason(std::move(err));
return -1;
}
return 0;
......@@ -151,17 +152,17 @@ public:
}
private:
template <class LowerLayer>
static void write(LowerLayer& down, string_view output) {
template <class LowerLayerPtr>
static void write(LowerLayerPtr down, string_view output) {
auto out = as_bytes(make_span(output));
down.begin_output();
auto& buf = down.output_buffer();
down->begin_output();
auto& buf = down->output_buffer();
buf.insert(buf.end(), out.begin(), out.end());
down.end_output();
down->end_output();
}
template <class LowerLayer>
bool handle_header(LowerLayer& down, string_view input) {
template <class LowerLayerPtr>
bool handle_header(LowerLayerPtr down, string_view input) {
// Parse the first line, i.e., "METHOD REQUEST-URI VERSION".
auto [first_line, remainder] = split(input, "\r\n");
auto [method, request_uri, version] = split2(first_line, " ");
......@@ -170,7 +171,7 @@ private:
auto err = make_error(pec::invalid_argument,
"invalid operation: expected GET, got "
+ to_string(method));
down.abort_reason(std::move(err));
down->abort_reason(std::move(err));
return false;
}
// Store the request information in the settings for the upper layer.
......@@ -196,17 +197,17 @@ private:
} else {
auto err = make_error(pec::missing_field,
"Mandatory field Sec-WebSocket-Key not found");
down.abort_reason(std::move(err));
down->abort_reason(std::move(err));
return false;
}
// Try initializing the upper layer.
if (auto err = upper_layer_.init(owner_, down, cfg_)) {
down.abort_reason(std::move(err));
down->abort_reason(std::move(err));
return false;
}
// Send server handshake.
down.begin_output();
auto& buf = down.output_buffer();
down->begin_output();
auto& buf = down->output_buffer();
auto append = [&buf](string_view output) {
auto out = as_bytes(make_span(output));
buf.insert(buf.end(), out.begin(), out.end());
......@@ -217,7 +218,7 @@ private:
"Sec-WebSocket-Accept: ");
append(sec_key);
append("\r\n\r\n");
down.end_output();
down->end_output();
// Done.
handshake_complete_ = true;
return true;
......
......@@ -18,18 +18,12 @@ public:
// -- interface for the upper layer ------------------------------------------
class access {
public:
explicit access(mock_stream_transport* transport) : transport_(transport) {
// nop
}
void begin_output() {
// nop
}
auto& output_buffer() {
return transport_->output;
return output;
}
constexpr void end_output() {
......@@ -40,26 +34,23 @@ public:
return true;
}
const caf::error& abort_reason() {
return abort_reason_;
}
void abort_reason(caf::error reason) {
transport_->abort_reason = std::move(reason);
abort_reason_ = std::move(reason);
}
void configure_read(caf::net::receive_policy policy) {
transport_->min_read_size = policy.min_size;
transport_->max_read_size = policy.max_size;
min_read_size = policy.min_size;
max_read_size = policy.max_size;
}
private:
mock_stream_transport* transport_;
};
friend class access;
// -- initialization ---------------------------------------------------------
caf::error init(const caf::settings& config) {
access this_layer{this};
return upper_layer.init(nullptr, this_layer, config);
return upper_layer.init(nullptr, this, config);
}
caf::error init() {
......@@ -89,7 +80,6 @@ public:
ptrdiff_t handle_input() {
ptrdiff_t result = 0;
access this_layer{this};
while (max_read_size > 0) {
CAF_ASSERT(max_read_size > static_cast<size_t>(read_size_));
size_t len = max_read_size - static_cast<size_t>(read_size_);
......@@ -106,16 +96,16 @@ public:
return result;
auto delta = make_span(read_buf_.data() + delta_offset,
read_size_ - delta_offset);
auto consumed = upper_layer.consume(this_layer, caf::make_span(read_buf_),
auto consumed = upper_layer.consume(this, caf::make_span(read_buf_),
delta);
if (consumed > 0) {
result += static_cast<ptrdiff_t>(consumed);
read_buf_.erase(read_buf_.begin(), read_buf_.begin() + consumed);
read_size_ -= consumed;
} else if (consumed < 0) {
if (!abort_reason)
abort_reason = caf::sec::runtime_error;
upper_layer.abort(this_layer, abort_reason);
if (!abort_reason_)
abort_reason_ = caf::sec::runtime_error;
upper_layer.abort(this, abort_reason_);
return -1;
}
}
......@@ -124,8 +114,6 @@ public:
// -- member variables -------------------------------------------------------
caf::error abort_reason;
UpperLayer upper_layer;
std::vector<caf::byte> output;
......@@ -140,4 +128,6 @@ private:
std::vector<caf::byte> read_buf_;
ptrdiff_t read_size_ = 0;
caf::error abort_reason_;
};
......@@ -47,39 +47,46 @@ struct app_t {
// nop
}
template <class LowerLayer>
error init(net::socket_manager* mgr, LowerLayer& down, const settings&) {
template <class LowerLayerPtr>
error init(net::socket_manager* mgr, LowerLayerPtr down, const settings&) {
self = mgr->make_actor_shell();
bhvr = behavior{
[this](std::string& line) { lines.emplace_back(std::move(line)); },
[this](std::string& line) {
CAF_MESSAGE("received an asynchronous message: " << line);
lines.emplace_back(std::move(line));
},
};
fallback = make_type_erased_callback([](message& msg) -> result<message> {
CAF_FAIL("unexpected message: " << msg);
return make_error(sec::unexpected_message);
});
down.configure_read(net::receive_policy::up_to(2048));
down->configure_read(net::receive_policy::up_to(2048));
return none;
}
template <class LowerLayer>
bool prepare_send(LowerLayer&) {
while (self->consume_message(bhvr, *fallback))
; // Repeat.
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
while (self->consume_message(bhvr, *fallback)) {
// We set abort_reason in our response handlers in case of an error.
if (down->abort_reason())
return false;
// else: repeat.
}
return true;
}
template <class LowerLayer>
bool done_sending(LowerLayer&) {
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) {
return self->try_block_mailbox();
}
template <class LowerLayer>
void abort(LowerLayer&, const error& reason) {
template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) {
CAF_FAIL("app::abort called: " << reason);
}
template <class LowerLayer>
ptrdiff_t consume(LowerLayer& down, byte_span buf, byte_span) {
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span buf, byte_span) {
// Seek newline character.
constexpr auto nl = byte{'\n'};
if (auto i = std::find(buf.begin(), buf.end(), nl); i != buf.end()) {
......@@ -97,11 +104,11 @@ struct app_t {
if (auto parsed_res = config_value::parse(line)) {
val = std::move(*parsed_res);
} else {
down.abort_reason(std::move(parsed_res.error()));
down->abort_reason(std::move(parsed_res.error()));
return -1;
}
if (!holds_alternative<settings>(val)) {
down.abort_reason(
down->abort_reason(
make_error(pec::type_mismatch,
"expected a dictionary, got a "s + val.type_name()));
return -1;
......@@ -110,20 +117,25 @@ struct app_t {
config_value_reader reader{&val};
caf::message msg;
if (!reader.apply_object(msg)) {
down.abort_reason(reader.get_error());
down->abort_reason(reader.get_error());
return -1;
}
// Dispatch message to worker.
CAF_MESSAGE("app received a message from its socket: " << msg);
self->request(worker, std::chrono::seconds{1}, std::move(msg))
.then(
[this](int32_t value) {
[this, down](int32_t value) mutable {
++received_responses;
CAF_CHECK_EQUAL(value, 246);
// Respond with the value as string.
auto str_response = std::to_string(value);
str_response += '\n';
down->begin_output();
auto& buf = down->output_buffer();
auto bytes = as_bytes(make_span(str_response));
buf.insert(buf.end(), bytes.begin(), bytes.end());
down->end_output();
},
[](error&) {
// TODO: implement me
});
[down](error& err) mutable { down->abort_reason(std::move(err)); });
// Try consuming more from the buffer.
consumed_bytes += static_cast<size_t>(num_bytes);
auto sub_buf = buf.subspan(num_bytes);
......@@ -163,6 +175,8 @@ struct fixture : host_fixture, test_coordinator_fixture<> {
auto sockets = unbox(net::make_stream_socket_pair());
self_socket_guard.reset(sockets.first);
testee_socket_guard.reset(sockets.second);
if (auto err = nonblocking(self_socket_guard.socket(), true))
CAF_FAIL("nonblocking returned an error: " << err);
if (auto err = nonblocking(testee_socket_guard.socket(), true))
CAF_FAIL("nonblocking returned an error: " << err);
}
......@@ -173,6 +187,10 @@ struct fixture : host_fixture, test_coordinator_fixture<> {
return;
for (size_t i = 0; i < 1000; ++i) {
mpx.poll_once(false);
byte tmp[1024];
auto bytes = read(self_socket_guard.socket(), make_span(tmp, 1024));
if (bytes > 0)
recv_buf.insert(recv_buf.end(), tmp, tmp + bytes);
if (!predicate())
return;
std::this_thread::sleep_for(std::chrono::milliseconds(1));
......@@ -190,6 +208,7 @@ struct fixture : host_fixture, test_coordinator_fixture<> {
net::multiplexer mpx;
net::socket_guard<net::stream_socket> self_socket_guard;
net::socket_guard<net::stream_socket> testee_socket_guard;
std::vector<byte> recv_buf;
};
constexpr std::string_view input = R"__(
......@@ -229,7 +248,11 @@ CAF_TEST(actor shells can send requests and receive responses) {
send(input);
run_while([&] { return app.consumed_bytes != input.size(); });
expect((int32_t), to(worker).with(123));
run_while([&] { return app.received_responses != 1; });
string_view expected_response = "246\n";
run_while([&] { return recv_buf.size() < expected_response.size(); });
string_view received_response{reinterpret_cast<char*>(recv_buf.data()),
recv_buf.size()};
CAF_CHECK_EQUAL(received_response, expected_response);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -41,29 +41,29 @@ struct app_t {
settings cfg;
template <class LowerLayer>
error init(net::socket_manager*, LowerLayer&, const settings& init_cfg) {
template <class LowerLayerPtr>
error init(net::socket_manager*, LowerLayerPtr, const settings& init_cfg) {
cfg = init_cfg;
return none;
}
template <class LowerLayer>
bool prepare_send(LowerLayer&) {
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr) {
return true;
}
template <class LowerLayer>
bool done_sending(LowerLayer&) {
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) {
return true;
}
template <class LowerLayer>
void abort(LowerLayer&, const error& reason) {
template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) {
CAF_FAIL("app::abort called: " << reason);
}
template <class LowerLayer>
ptrdiff_t consume(LowerLayer& down, byte_span buffer, byte_span) {
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span buffer, byte_span) {
constexpr auto nl = byte{'\n'};
auto e = buffer.end();
if (auto i = std::find(buffer.begin(), e, nl); i != e) {
......@@ -120,7 +120,7 @@ CAF_TEST(applications receive handshake data via config) {
{
auto consumed = transport.handle_input();
if (consumed < 0)
CAF_FAIL("error handling input: " << transport.abort_reason);
CAF_FAIL("error handling input: " << transport.abort_reason());
CAF_CHECK_EQUAL(consumed, static_cast<ptrdiff_t>(opening_handshake.size()));
}
CAF_CHECK_EQUAL(transport.input.size(), 0u);
......
......@@ -88,29 +88,29 @@ public:
~dummy_application() = default;
template <class Parent>
error init(socket_manager*, Parent& parent, const settings&) {
parent.configure_read(receive_policy::exactly(hello_manager.size()));
template <class ParentPtr>
error init(socket_manager*, ParentPtr parent, const settings&) {
parent->configure_read(receive_policy::exactly(hello_manager.size()));
return none;
}
template <class Parent>
bool prepare_send(Parent& parent) {
template <class ParentPtr>
bool prepare_send(ParentPtr parent) {
CAF_MESSAGE("prepare_send called");
auto& buf = parent.output_buffer();
auto& buf = parent->output_buffer();
auto data = as_bytes(make_span(hello_manager));
buf.insert(buf.end(), data.begin(), data.end());
return true;
}
template <class Parent>
bool done_sending(Parent&) {
template <class ParentPtr>
bool done_sending(ParentPtr) {
CAF_MESSAGE("done_sending called");
return true;
}
template <class Parent>
size_t consume(Parent&, span<const byte> data, span<const byte>) {
template <class ParentPtr>
size_t consume(ParentPtr, span<const byte> data, span<const byte>) {
recv_buf_->clear();
recv_buf_->insert(recv_buf_->begin(), data.begin(), data.end());
CAF_MESSAGE("Received " << recv_buf_->size()
......@@ -118,40 +118,25 @@ public:
return recv_buf_->size();
}
template <class Parent>
void resolve(Parent& parent, string_view path, const actor& listener) {
template <class ParentPtr>
void resolve(ParentPtr parent, string_view path, const actor& listener) {
actor_id aid = 42;
auto hid = string_view("0011223344556677889900112233445566778899");
auto nid = unbox(make_node_id(42, hid));
actor_config cfg;
endpoint_manager_ptr ptr{&parent.manager()};
endpoint_manager_ptr ptr{&parent->manager()};
auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(
aid, nid, &parent.system(), cfg, std::move(ptr));
aid, nid, &parent->system(), cfg, std::move(ptr));
anon_send(listener, resolve_atom_v, std::string{path.begin(), path.end()},
p);
}
template <class Parent>
void timeout(Parent&, const std::string&, uint64_t) {
// nop
}
template <class Parent>
void new_proxy(Parent&, actor_id) {
// nop
}
template <class Parent>
void local_actor_down(Parent&, actor_id, const error&) {
// nop
}
static void handle_error(sec code) {
CAF_FAIL("handle_error called with " << CAF_ARG(code));
}
template <class Parent>
static void abort(Parent&, const error& reason) {
template <class ParentPtr>
static void abort(ParentPtr, const error& reason) {
CAF_FAIL("abort called with " << CAF_ARG(reason));
}
......
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