Commit df9e8117 authored by Dominik Charousset's avatar Dominik Charousset

Add new flow adapter for socket managers

parent bf38718f
...@@ -61,6 +61,7 @@ caf_incubator_add_component( ...@@ -61,6 +61,7 @@ caf_incubator_add_component(
multiplexer multiplexer
net.actor_shell net.actor_shell
net.length_prefix_framing net.length_prefix_framing
net.subscriber_adapter
net.typed_actor_shell net.typed_actor_shell
net.web_socket.client net.web_socket.client
net.web_socket.handshake net.web_socket.handshake
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/flow/poll_subscriber.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
namespace caf::net {
/// Base class for buffered consumption of published items.
template <class T>
class subscriber_adapter : public flow::poll_subscriber<T> {
public:
using super = flow::poll_subscriber<T>;
explicit subscriber_adapter(socket_manager* owner) : mgr_(owner) {
// nop
}
private:
void wakeup(std::unique_lock<std::mutex>&) {
mgr_->mpx().register_writing(mgr_);
}
intrusive_ptr<socket_manager> mgr_;
};
template <class T>
using subscriber_adapter_ptr = intrusive_ptr<subscriber_adapter<T>>;
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE net.subscriber_adapter
#include "caf/net/subscriber_adapter.hpp"
#include "net-test.hpp"
#include "caf/flow/async/publisher.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/scheduled_actor/flow.hpp"
using namespace caf;
namespace {
class reader {
public:
reader(net::stream_socket fd, size_t n) : sg_(fd) {
if (auto err = nonblocking(fd, true))
FAIL("unable to set nonblocking flag: " << err);
buf_.resize(n);
}
auto fd() {
return sg_.socket();
}
void read_some() {
if (rd_pos_ < buf_.size()) {
auto res = read(fd(), make_span(buf_).subspan(rd_pos_));
if (res > 0) {
rd_pos_ += static_cast<size_t>(res);
MESSAGE(rd_pos_ << " bytes received");
} else if (res < 0 && !net::last_socket_error_is_temporary()) {
FAIL("failed to read: " << net::last_socket_error_as_string());
}
}
}
bool done() const noexcept {
return rd_pos_ == buf_.size();
}
auto& buf() const {
return buf_;
}
private:
size_t rd_pos_ = 0;
std::vector<byte> buf_;
net::socket_guard<net::stream_socket> sg_;
};
class app_t {
public:
using input_tag = tag::stream_oriented;
explicit app_t(flow::async::publisher_ptr<int32_t> input) : input_(input) {
// nop
}
template <class LowerLayerPtr>
error init(net::socket_manager* owner, LowerLayerPtr, const settings&) {
adapter_ = make_counted<net::subscriber_adapter<int32_t>>(owner);
input_->async_subscribe(adapter_);
input_ = nullptr;
return none;
}
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
while (!done_ && down->can_send_more()) {
auto [val, done, err] = adapter_->poll();
if (val) {
written_values_.emplace_back(*val);
auto offset = written_bytes_.size();
binary_serializer sink{nullptr, written_bytes_};
if (!sink.apply(*val))
FAIL("sink.apply failed: " << sink.get_error());
auto bytes = make_span(written_bytes_).subspan(offset);
down->begin_output();
auto& buf = down->output_buffer();
buf.insert(buf.end(), bytes.begin(), bytes.end());
down->end_output();
} else if (done) {
done_ = true;
if (err)
FAIL("flow error: " << *err);
} else {
break;
}
}
MESSAGE(written_bytes_.size() << " bytes written");
return true;
}
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) {
return !adapter_->has_data();
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) {
FAIL("app::abort called: " << reason);
}
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr, byte_span, byte_span) {
FAIL("app::consume called: unexpected data");
}
auto& written_values() {
return written_values_;
}
auto& written_bytes() {
return written_bytes_;
}
private:
bool done_ = false;
std::vector<int32_t> written_values_;
std::vector<byte> written_bytes_;
net::subscriber_adapter_ptr<int32_t> adapter_;
flow::async::publisher_ptr<int32_t> input_;
};
struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() : mm(sys) {
mm.mpx().set_thread_id();
if (auto err = mm.mpx().init())
CAF_FAIL("mpx.init() failed: " << err);
}
bool handle_io_event() override {
return mm.mpx().poll_once(false);
}
net::middleman mm;
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("subscriber adapters wake up idle socket managers") {
GIVEN("a publisher<T>") {
static constexpr size_t num_items = 4211;
auto src = flow::async::publisher_from(sys, [](auto* self) {
return self->make_publisher()->repeat(42)->take(num_items);
});
WHEN("sending items of the stream over a socket") {
auto [fd1, fd2] = unbox(net::make_stream_socket_pair());
if (auto err = nonblocking(fd1, true))
FAIL("nonblocking(fd1) returned an error: " << err);
if (auto err = nonblocking(fd2, true))
FAIL("nonblocking(fd2) returned an error: " << err);
auto mgr = net::make_socket_manager<app_t, net::stream_transport>(
fd1, mm.mpx_ptr(), src);
auto& app = mgr->top_layer();
if (auto err = mgr->init(content(cfg)))
FAIL("mgr->init() failed: " << err);
THEN("the reader receives all items before the connection closes") {
reader rd{fd2, num_items * sizeof(int32_t)};
while (!rd.done()) {
run();
rd.read_some();
}
CHECK_EQ(app.written_values(), std::vector<int32_t>(num_items, 42));
CHECK_EQ(app.written_bytes().size(), num_items * sizeof(int32_t));
CHECK_EQ(rd.buf().size(), num_items * sizeof(int32_t));
CHECK_EQ(app.written_bytes(), rd.buf());
}
}
}
}
END_FIXTURE_SCOPE()
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