Commit 823b8ad1 authored by Dominik Charousset's avatar Dominik Charousset

Fix formatting

parent e9eb8ba4
...@@ -38,7 +38,8 @@ public: ...@@ -38,7 +38,8 @@ public:
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class LowerLayerPtr> template <class LowerLayerPtr>
error init(socket_manager* owner, LowerLayerPtr parent, const settings& config) { error
init(socket_manager* owner, LowerLayerPtr parent, const settings& config) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
owner_ = owner; owner_ = owner;
cfg_ = config; cfg_ = config;
......
...@@ -39,9 +39,8 @@ public: ...@@ -39,9 +39,8 @@ public:
void on_consumer_demand(size_t new_demand) override { void on_consumer_demand(size_t new_demand) override {
auto prev = demand_.fetch_add(new_demand); auto prev = demand_.fetch_add(new_demand);
if (prev == 0) if (prev == 0)
mgr_->mpx().schedule_fn([adapter = strong_this()] { mgr_->mpx().schedule_fn(
adapter->continue_reading(); [adapter = strong_this()] { adapter->continue_reading(); });
});
} }
void ref_producer() const noexcept override { void ref_producer() const noexcept override {
......
...@@ -130,8 +130,9 @@ public: ...@@ -130,8 +130,9 @@ public:
error init(socket_manager* owner, ParentPtr parent, const settings& config) { error init(socket_manager* owner, ParentPtr parent, const settings& config) {
namespace mm = defaults::middleman; namespace mm = defaults::middleman;
auto default_max_reads = static_cast<uint32_t>(mm::max_consecutive_reads); auto default_max_reads = static_cast<uint32_t>(mm::max_consecutive_reads);
max_consecutive_reads_ = get_or( max_consecutive_reads_ = get_or(config,
config, "caf.middleman.max-consecutive-reads", default_max_reads); "caf.middleman.max-consecutive-reads",
default_max_reads);
auto sock = parent->handle(); auto sock = parent->handle();
if constexpr (std::is_base_of<tcp_stream_socket, decltype(sock)>::value) { if constexpr (std::is_base_of<tcp_stream_socket, decltype(sock)>::value) {
if (auto err = nodelay(sock, true)) { if (auto err = nodelay(sock, true)) {
......
...@@ -27,8 +27,9 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket { ...@@ -27,8 +27,9 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket {
/// Passing the address `0.0.0.0` will accept incoming connection from any host. /// Passing the address `0.0.0.0` will accept incoming connection from any host.
/// Passing port 0 lets the OS choose the port. /// Passing port 0 lets the OS choose the port.
/// @relates tcp_accept_socket /// @relates tcp_accept_socket
expected<tcp_accept_socket> CAF_NET_EXPORT expected<tcp_accept_socket>
make_tcp_accept_socket(ip_endpoint node, bool reuse_addr = false); CAF_NET_EXPORT make_tcp_accept_socket(ip_endpoint node,
bool reuse_addr = false);
/// Creates a new TCP socket to accept connections on a given port. /// Creates a new TCP socket to accept connections on a given port.
/// @param node The endpoint to listen on and the filter for incoming addresses. /// @param node The endpoint to listen on and the filter for incoming addresses.
......
...@@ -110,8 +110,8 @@ public: ...@@ -110,8 +110,8 @@ public:
[&](auto& layer) -> decltype(layer.resolve(*this, locator, listener)) { [&](auto& layer) -> decltype(layer.resolve(*this, locator, listener)) {
return layer.resolve(*this, locator, listener); return layer.resolve(*this, locator, listener);
}, },
[&](auto& layer) -> decltype( [&](auto& layer) -> decltype(layer.resolve(*this, locator.path(),
layer.resolve(*this, locator.path(), listener)) { listener)) {
return layer.resolve(*this, locator.path(), listener); return layer.resolve(*this, locator.path(), listener);
}); });
f(next_layer_); f(next_layer_);
......
...@@ -118,8 +118,8 @@ public: ...@@ -118,8 +118,8 @@ public:
} }
template <class Parent> template <class Parent>
expected<worker_ptr> add_new_worker(Parent& parent, node_id node, expected<worker_ptr>
id_type id) { add_new_worker(Parent& parent, node_id node, id_type id) {
CAF_LOG_TRACE(CAF_ARG(node) << CAF_ARG(id)); CAF_LOG_TRACE(CAF_ARG(node) << CAF_ARG(id));
auto application = factory_.make(); auto application = factory_.make();
auto worker = std::make_shared<worker_type>(std::move(application), id); auto worker = std::make_shared<worker_type>(std::move(application), id);
......
...@@ -42,7 +42,7 @@ public: ...@@ -42,7 +42,7 @@ public:
} }
template <class LowerLayerPtr> template <class LowerLayerPtr>
error init(net::socket_manager* mgr, LowerLayerPtr&&, const settings&cfg) { error init(net::socket_manager* mgr, LowerLayerPtr&&, const settings& cfg) {
if (auto err = reader_.init(cfg)) if (auto err = reader_.init(cfg))
return err; return err;
if (auto err = writer_.init(cfg)) if (auto err = writer_.init(cfg))
......
...@@ -36,8 +36,7 @@ public: ...@@ -36,8 +36,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
template <class... Ts> template <class... Ts>
explicit server(Ts&&... xs) explicit server(Ts&&... xs) : upper_layer_(std::forward<Ts>(xs)...) {
: upper_layer_(std::forward<Ts>(xs)...) {
// > A server MUST NOT mask any frames that it sends to the client. // > A server MUST NOT mask any frames that it sends to the client.
// See RFC 6455, Section 5.1. // See RFC 6455, Section 5.1.
upper_layer_.mask_outgoing_frames = false; upper_layer_.mask_outgoing_frames = false;
......
...@@ -177,9 +177,9 @@ void multiplexer::discard(const socket_manager_ptr& mgr) { ...@@ -177,9 +177,9 @@ void multiplexer::discard(const socket_manager_ptr& mgr) {
if (shutting_down_) { if (shutting_down_) {
// nop // nop
} else { } else {
mgr->handle_error(sec::discarded); mgr->handle_error(sec::discarded);
if (auto mgr_index = index_of(mgr); mgr_index != -1) if (auto mgr_index = index_of(mgr); mgr_index != -1)
del(mgr_index); del(mgr_index);
} }
} else { } else {
write_to_pipe(pollset_updater::code::discard_manager, mgr.get()); write_to_pipe(pollset_updater::code::discard_manager, mgr.get());
...@@ -387,7 +387,7 @@ short multiplexer::handle(const socket_manager_ptr& mgr, short events, ...@@ -387,7 +387,7 @@ short multiplexer::handle(const socket_manager_ptr& mgr, short events,
mgr->mask_del(operation::read_write); mgr->mask_del(operation::read_write);
events = 0; events = 0;
} else { } else {
switch (mgr->mask()){ switch (mgr->mask()) {
case operation::read: case operation::read:
events = input_mask; events = input_mask;
break; break;
......
...@@ -110,8 +110,9 @@ public: ...@@ -110,8 +110,9 @@ public:
auto hid = string_view("0011223344556677889900112233445566778899"); auto hid = string_view("0011223344556677889900112233445566778899");
auto nid = unbox(make_node_id(42, hid)); auto nid = unbox(make_node_id(42, hid));
actor_config cfg; actor_config cfg;
auto p = make_actor<actor_proxy_impl, strong_actor_ptr>( auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid,
aid, nid, &mgr.system(), cfg, &mgr); &mgr.system(), cfg,
&mgr);
std::string path{locator.path().begin(), locator.path().end()}; std::string path{locator.path().begin(), locator.path().end()};
anon_send(listener, resolve_atom_v, std::move(path), p); anon_send(listener, resolve_atom_v, std::move(path), p);
} }
......
...@@ -147,7 +147,7 @@ SCENARIO("publisher adapters suspend reads if the buffer becomes full") { ...@@ -147,7 +147,7 @@ SCENARIO("publisher adapters suspend reads if the buffer becomes full") {
.from_resource(rd) .from_resource(rd)
.for_each([&outputs](int32_t x) { outputs.emplace_back(x); }); .for_each([&outputs](int32_t x) { outputs.emplace_back(x); });
}); });
WHEN("a producer reads from a socket and publishes to the buffer"){ WHEN("a producer reads from a socket and publishes to the buffer") {
auto [fd1, fd2] = unbox(net::make_stream_socket_pair()); auto [fd1, fd2] = unbox(net::make_stream_socket_pair());
auto writer_thread = std::thread{[fd1{fd1}] { auto writer_thread = std::thread{[fd1{fd1}] {
writer out{fd1}; writer out{fd1};
...@@ -157,8 +157,9 @@ SCENARIO("publisher adapters suspend reads if the buffer becomes full") { ...@@ -157,8 +157,9 @@ SCENARIO("publisher adapters suspend reads if the buffer becomes full") {
if (auto err = nonblocking(fd2, true)) if (auto err = nonblocking(fd2, true))
FAIL("nonblocking(fd2) returned an error: " << err); FAIL("nonblocking(fd2) returned an error: " << err);
auto mgr = net::make_socket_manager<app_t, net::length_prefix_framing, auto mgr = net::make_socket_manager<app_t, net::length_prefix_framing,
net::stream_transport>( net::stream_transport>(fd2,
fd2, mm.mpx_ptr(), std::move(wr)); mm.mpx_ptr(),
std::move(wr));
if (auto err = mgr->init(content(cfg))) if (auto err = mgr->init(content(cfg)))
FAIL("mgr->init() failed: " << err); FAIL("mgr->init() failed: " << err);
THEN("the actor receives all items from the writer (socket)") { THEN("the actor receives all items from the writer (socket)") {
......
...@@ -59,16 +59,15 @@ struct app_t { ...@@ -59,16 +59,15 @@ struct app_t {
constexpr auto key = "the sample nonce"_sv; constexpr auto key = "the sample nonce"_sv;
constexpr auto http_request constexpr auto http_request = "GET /chat HTTP/1.1\r\n"
= "GET /chat HTTP/1.1\r\n" "Host: server.example.com\r\n"
"Host: server.example.com\r\n" "Upgrade: websocket\r\n"
"Upgrade: websocket\r\n" "Connection: Upgrade\r\n"
"Connection: Upgrade\r\n" "Sec-WebSocket-Version: 13\r\n"
"Sec-WebSocket-Version: 13\r\n" "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n"
"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" "Origin: http://example.com\r\n"
"Origin: http://example.com\r\n" "Sec-WebSocket-Protocol: chat, superchat\r\n"
"Sec-WebSocket-Protocol: chat, superchat\r\n" "\r\n"_sv;
"\r\n"_sv;
constexpr auto http_response constexpr auto http_response
= "HTTP/1.1 101 Switching Protocols\r\n" = "HTTP/1.1 101 Switching Protocols\r\n"
...@@ -102,7 +101,7 @@ BEGIN_FIXTURE_SCOPE(host_fixture) ...@@ -102,7 +101,7 @@ BEGIN_FIXTURE_SCOPE(host_fixture)
SCENARIO("the client performs the WebSocket handshake on startup") { SCENARIO("the client performs the WebSocket handshake on startup") {
GIVEN("valid WebSocket handshake data") { GIVEN("valid WebSocket handshake data") {
WHEN("starting a WebSocket client"){ WHEN("starting a WebSocket client") {
mock_client_type client{make_handshake()}; mock_client_type client{make_handshake()};
THEN("the client sends its HTTP request when initializing it") { THEN("the client sends its HTTP request when initializing it") {
CHECK_EQ(client.init(), error{}); CHECK_EQ(client.init(), error{});
......
...@@ -12,9 +12,7 @@ using namespace caf; ...@@ -12,9 +12,7 @@ using namespace caf;
namespace { namespace {
struct fixture { struct fixture {};
};
} // namespace } // namespace
......
...@@ -18,16 +18,15 @@ namespace { ...@@ -18,16 +18,15 @@ namespace {
constexpr auto key = "the sample nonce"_sv; constexpr auto key = "the sample nonce"_sv;
constexpr auto http_request constexpr auto http_request = "GET /chat HTTP/1.1\r\n"
= "GET /chat HTTP/1.1\r\n" "Host: server.example.com\r\n"
"Host: server.example.com\r\n" "Upgrade: websocket\r\n"
"Upgrade: websocket\r\n" "Connection: Upgrade\r\n"
"Connection: Upgrade\r\n" "Sec-WebSocket-Version: 13\r\n"
"Sec-WebSocket-Version: 13\r\n" "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n"
"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" "Origin: http://example.com\r\n"
"Origin: http://example.com\r\n" "Sec-WebSocket-Protocol: chat, superchat\r\n"
"Sec-WebSocket-Protocol: chat, superchat\r\n" "\r\n"_sv;
"\r\n"_sv;
constexpr auto http_response constexpr auto http_response
= "HTTP/1.1 101 Switching Protocols\r\n" = "HTTP/1.1 101 Switching Protocols\r\n"
......
...@@ -145,10 +145,10 @@ CAF_TEST(receive) { ...@@ -145,10 +145,10 @@ CAF_TEST(receive) {
recv_socket_guard.release(), &mpx, shared_recv_buf, shared_send_buf); recv_socket_guard.release(), &mpx, shared_recv_buf, shared_send_buf);
CAF_CHECK_EQUAL(mgr->init(config), none); CAF_CHECK_EQUAL(mgr->init(config), none);
CAF_CHECK_EQUAL(mpx.num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 2u);
CAF_CHECK_EQUAL( CAF_CHECK_EQUAL(static_cast<size_t>(
static_cast<size_t>( write(send_socket_guard.socket(),
write(send_socket_guard.socket(), as_bytes(make_span(hello_manager)))), as_bytes(make_span(hello_manager)))),
hello_manager.size()); hello_manager.size());
CAF_MESSAGE("wrote " << hello_manager.size() << " bytes."); CAF_MESSAGE("wrote " << hello_manager.size() << " bytes.");
run(); run();
CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(shared_recv_buf->data()), CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(shared_recv_buf->data()),
......
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