Commit 6fd5dab6 authored by Samir Halilcevic's avatar Samir Halilcevic

Use the new project structure for caf_net

parent bbb8db04
......@@ -28,78 +28,75 @@ caf_add_component(
HEADERS
${CAF_NET_HEADERS}
SOURCES
src/detail/convert_ip_endpoint.cpp
src/detail/pollset_updater.cpp
src/detail/rfc6455.cpp
src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp
src/net/datagram_socket.cpp
src/net/dsl/config_base.cpp
src/net/generic_lower_layer.cpp
src/net/generic_upper_layer.cpp
src/net/http/config.cpp
src/net/http/lower_layer.cpp
src/net/http/method.cpp
src/net/http/request.cpp
src/net/http/request_header.cpp
src/net/http/responder.cpp
src/net/http/response.cpp
src/net/http/route.cpp
src/net/http/router.cpp
src/net/http/server.cpp
src/net/http/server_factory.cpp
src/net/http/status.cpp
src/net/http/upper_layer.cpp
src/net/http/v1.cpp
src/net/ip.cpp
src/net/lp/default_trait.cpp
src/net/lp/frame.cpp
src/net/lp/framing.cpp
src/net/lp/lower_layer.cpp
src/net/lp/upper_layer.cpp
src/net/middleman.cpp
src/net/multiplexer.cpp
src/net/network_socket.cpp
src/net/octet_stream/lower_layer.cpp
src/net/octet_stream/policy.cpp
src/net/octet_stream/transport.cpp
src/net/octet_stream/upper_layer.cpp
src/net/pipe_socket.cpp
src/net/prometheus.cpp
src/net/socket.cpp
src/net/socket_event_layer.cpp
src/net/socket_manager.cpp
src/net/ssl/connection.cpp
src/net/ssl/context.cpp
src/net/ssl/dtls.cpp
src/net/ssl/errc.cpp
src/net/ssl/format.cpp
src/net/ssl/password.cpp
src/net/ssl/startup.cpp
src/net/ssl/tcp_acceptor.cpp
src/net/ssl/tls.cpp
src/net/ssl/transport.cpp
src/net/ssl/verify.cpp
src/net/stream_socket.cpp
src/net/tcp_accept_socket.cpp
src/net/tcp_stream_socket.cpp
src/net/this_host.cpp
src/net/udp_datagram_socket.cpp
src/net/web_socket/client.cpp
src/net/web_socket/default_trait.cpp
src/net/web_socket/frame.cpp
src/net/web_socket/framing.cpp
src/net/web_socket/handshake.cpp
src/net/web_socket/lower_layer.cpp
src/net/web_socket/server.cpp
src/net/web_socket/upper_layer.cpp
TEST_SOURCES
test/net-test.cpp
TEST_SUITES
detail.rfc6455
caf/detail/convert_ip_endpoint.cpp
caf/detail/pollset_updater.cpp
caf/detail/rfc6455.cpp
caf/detail/rfc6455.test.cpp
caf/net/abstract_actor_shell.cpp
caf/net/actor_shell.cpp
caf/net/datagram_socket.cpp
caf/net/dsl/config_base.cpp
caf/net/generic_lower_layer.cpp
caf/net/generic_upper_layer.cpp
caf/net/http/config.cpp
caf/net/http/lower_layer.cpp
caf/net/http/method.cpp
caf/net/http/request.cpp
caf/net/http/request_header.cpp
caf/net/http/responder.cpp
caf/net/http/response.cpp
caf/net/http/route.cpp
caf/net/http/router.cpp
caf/net/http/server.cpp
caf/net/http/server_factory.cpp
caf/net/http/status.cpp
caf/net/http/upper_layer.cpp
caf/net/http/v1.cpp
caf/net/ip.cpp
caf/net/lp/default_trait.cpp
caf/net/lp/frame.cpp
caf/net/lp/framing.cpp
caf/net/lp/lower_layer.cpp
caf/net/lp/upper_layer.cpp
caf/net/middleman.cpp
caf/net/multiplexer.cpp
caf/net/network_socket.cpp
caf/net/octet_stream/lower_layer.cpp
caf/net/octet_stream/policy.cpp
caf/net/octet_stream/transport.cpp
caf/net/octet_stream/upper_layer.cpp
caf/net/pipe_socket.cpp
caf/net/prometheus.cpp
caf/net/socket.cpp
caf/net/socket_event_layer.cpp
caf/net/socket_manager.cpp
caf/net/ssl/connection.cpp
caf/net/ssl/context.cpp
caf/net/ssl/dtls.cpp
caf/net/ssl/errc.cpp
caf/net/ssl/format.cpp
caf/net/ssl/password.cpp
caf/net/ssl/startup.cpp
caf/net/ssl/tcp_acceptor.cpp
caf/net/ssl/tls.cpp
caf/net/ssl/transport.cpp
caf/net/ssl/verify.cpp
caf/net/stream_socket.cpp
caf/net/tcp_accept_socket.cpp
caf/net/tcp_stream_socket.cpp
caf/net/this_host.cpp
caf/net/udp_datagram_socket.cpp
caf/net/web_socket/client.cpp
caf/net/web_socket/default_trait.cpp
caf/net/web_socket/frame.cpp
caf/net/web_socket/framing.cpp
caf/net/web_socket/handshake.cpp
caf/net/web_socket/lower_layer.cpp
caf/net/web_socket/server.cpp
caf/net/web_socket/upper_layer.cpp
LEGACY_TEST_SOURCES
${CMAKE_CURRENT_BINARY_DIR}/test/pem.cpp
test/net-legacy-test.cpp
test/net-test.cpp
LEGACY_TEST_SUITES
detail.convert_ip_endpoint
net.accept_socket
......
......@@ -4,6 +4,7 @@
#include "caf/detail/rfc6455.hpp"
#include "caf/test/caf_test_main.hpp"
#include "caf/test/test.hpp"
#include "caf/byte_buffer.hpp"
......@@ -18,8 +19,6 @@ using namespace caf;
using impl = detail::rfc6455;
SUITE("detail.rfc6455") {
auto bytes(std::initializer_list<uint8_t> xs) {
byte_buffer result;
for (auto x : xs)
......@@ -300,4 +299,4 @@ TEST("decode a header with valid mask key plus large data") {
check_eq(hdr.payload_len, data.size());
}
} // SUITE("detail.rfc6455")
CAF_TEST_MAIN()
#include "net-test.hpp"
#include "caf/error.hpp"
#include "caf/init_global_meta_objects.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/startup.hpp"
#include "caf/net/this_host.hpp"
#include "caf/raise_error.hpp"
#define CAF_TEST_NO_MAIN
#include "caf/test/unit_test_impl.hpp"
using namespace caf;
// -- mock_stream_transport ----------------------------------------------------
net::multiplexer& mock_stream_transport::mpx() noexcept {
return *mpx_;
}
bool mock_stream_transport::can_send_more() const noexcept {
return true;
}
bool mock_stream_transport::is_reading() const noexcept {
return max_read_size > 0;
}
void mock_stream_transport::write_later() {
// nop
}
void mock_stream_transport::shutdown() {
// nop
}
void mock_stream_transport::switch_protocol(upper_layer_ptr new_up) {
next.swap(new_up);
}
bool mock_stream_transport::switching_protocol() const noexcept {
return next != nullptr;
}
void mock_stream_transport::configure_read(net::receive_policy policy) {
min_read_size = policy.min_size;
max_read_size = policy.max_size;
}
void mock_stream_transport::begin_output() {
// nop
}
byte_buffer& mock_stream_transport::output_buffer() {
return output;
}
bool mock_stream_transport::end_output() {
return true;
}
ptrdiff_t mock_stream_transport::handle_input() {
ptrdiff_t result = 0;
auto switch_to_next_protocol = [this] {
assert(next);
// Switch to the new protocol and initialize it.
configure_read(net::receive_policy::stop());
up.reset(next.release());
if (auto err = up->start(this)) {
up.reset();
return false;
}
return true;
};
// Loop until we have drained the buffer as much as we can.
while (max_read_size > 0 && input.size() >= min_read_size) {
auto n = std::min(input.size(), size_t{max_read_size});
auto bytes = make_span(input.data(), n);
auto delta = bytes.subspan(delta_offset);
auto consumed = up->consume(bytes, delta);
if (consumed < 0) {
// Negative values indicate that the application encountered an
// unrecoverable error.
return result;
} else if (static_cast<size_t>(consumed) > n) {
// Must not happen. An application cannot handle more data then we pass
// to it.
up->abort(make_error(sec::logic_error, "consumed > buffer.size"));
return result;
} else if (consumed == 0) {
if (next) {
// When switching protocol, the new layer has never seen the data, so we
// might just re-invoke the same data again.
if (!switch_to_next_protocol())
return -1;
} else {
// See whether the next iteration would change what we pass to the
// application (max_read_size_ may have changed). Otherwise, we'll try
// again later.
delta_offset = static_cast<ptrdiff_t>(n);
if (n == std::min(input.size(), size_t{max_read_size})) {
return result;
}
// else: "Fall through".
}
} else {
if (next && !switch_to_next_protocol())
return -1;
// Shove the unread bytes to the beginning of the buffer and continue
// to the next loop iteration.
result += consumed;
auto del = static_cast<size_t>(consumed);
delta_offset = static_cast<ptrdiff_t>(n - del);
input.erase(input.begin(), input.begin() + del);
}
}
return result;
}
// -- mock_web_socket_app ------------------------------------------------------
mock_web_socket_app::mock_web_socket_app(bool request_messages_on_start)
: request_messages_on_start(request_messages_on_start) {
// nop
}
caf::error mock_web_socket_app::start(caf::net::web_socket::lower_layer* ll) {
down = ll;
if (request_messages_on_start)
down->request_messages();
return caf::none;
}
void mock_web_socket_app::prepare_send() {
// nop
}
bool mock_web_socket_app::done_sending() {
return true;
}
caf::error
mock_web_socket_app::accept(const caf::net::http::request_header& hdr) {
// Store the request information in cfg to evaluate them later.
auto& ws = cfg["web-socket"].as_dictionary();
put(ws, "method", to_rfc_string(hdr.method()));
put(ws, "path", std::string{hdr.path()});
put(ws, "query", hdr.query());
put(ws, "fragment", hdr.fragment());
put(ws, "http-version", hdr.version());
if (hdr.num_fields() > 0) {
auto& fields = ws["fields"].as_dictionary();
hdr.for_each_field([&fields](auto key, auto val) {
put(fields, std::string{key}, std::string{val});
});
}
return caf::none;
}
void mock_web_socket_app::abort(const caf::error& reason) {
abort_reason = reason;
}
ptrdiff_t mock_web_socket_app::consume_text(std::string_view text) {
text_input.insert(text_input.end(), text.begin(), text.end());
return static_cast<ptrdiff_t>(text.size());
}
ptrdiff_t mock_web_socket_app::consume_binary(caf::byte_span bytes) {
binary_input.insert(binary_input.end(), bytes.begin(), bytes.end());
return static_cast<ptrdiff_t>(bytes.size());
}
// -- barrier ------------------------------------------------------------------
void barrier::arrive_and_wait() {
std::unique_lock<std::mutex> guard{mx_};
auto new_count = ++count_;
if (new_count == num_threads_) {
cv_.notify_all();
} else if (new_count > num_threads_) {
count_ = 1;
cv_.wait(guard, [this] { return count_.load() == num_threads_; });
} else {
cv_.wait(guard, [this] { return count_.load() == num_threads_; });
}
}
// -- main --------------------------------------------------------------------
int main(int argc, char** argv) {
net::this_host::startup();
net::ssl::startup();
using namespace caf;
net::middleman::init_global_meta_objects();
core::init_global_meta_objects();
auto result = test::main(argc, argv);
net::ssl::cleanup();
net::this_host::cleanup();
return result;
}
// 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.
#include "net-test.hpp"
#include "caf/error.hpp"
#include "caf/init_global_meta_objects.hpp"
#include "caf/net/middleman.hpp"
#include "caf/test/caf_test_main.hpp"
#include "caf/net/ssl/startup.hpp"
#include "caf/net/this_host.hpp"
#include "caf/raise_error.hpp"
CAF_TEST_MAIN(caf::net::middleman)
#define CAF_TEST_NO_MAIN
#include "caf/test/unit_test_impl.hpp"
using namespace caf;
// -- mock_stream_transport ----------------------------------------------------
net::multiplexer& mock_stream_transport::mpx() noexcept {
return *mpx_;
}
bool mock_stream_transport::can_send_more() const noexcept {
return true;
}
bool mock_stream_transport::is_reading() const noexcept {
return max_read_size > 0;
}
void mock_stream_transport::write_later() {
// nop
}
void mock_stream_transport::shutdown() {
// nop
}
void mock_stream_transport::switch_protocol(upper_layer_ptr new_up) {
next.swap(new_up);
}
bool mock_stream_transport::switching_protocol() const noexcept {
return next != nullptr;
}
void mock_stream_transport::configure_read(net::receive_policy policy) {
min_read_size = policy.min_size;
max_read_size = policy.max_size;
}
void mock_stream_transport::begin_output() {
// nop
}
byte_buffer& mock_stream_transport::output_buffer() {
return output;
}
bool mock_stream_transport::end_output() {
return true;
}
ptrdiff_t mock_stream_transport::handle_input() {
ptrdiff_t result = 0;
auto switch_to_next_protocol = [this] {
assert(next);
// Switch to the new protocol and initialize it.
configure_read(net::receive_policy::stop());
up.reset(next.release());
if (auto err = up->start(this)) {
up.reset();
return false;
}
return true;
};
// Loop until we have drained the buffer as much as we can.
while (max_read_size > 0 && input.size() >= min_read_size) {
auto n = std::min(input.size(), size_t{max_read_size});
auto bytes = make_span(input.data(), n);
auto delta = bytes.subspan(delta_offset);
auto consumed = up->consume(bytes, delta);
if (consumed < 0) {
// Negative values indicate that the application encountered an
// unrecoverable error.
return result;
} else if (static_cast<size_t>(consumed) > n) {
// Must not happen. An application cannot handle more data then we pass
// to it.
up->abort(make_error(sec::logic_error, "consumed > buffer.size"));
return result;
} else if (consumed == 0) {
if (next) {
// When switching protocol, the new layer has never seen the data, so we
// might just re-invoke the same data again.
if (!switch_to_next_protocol())
return -1;
} else {
// See whether the next iteration would change what we pass to the
// application (max_read_size_ may have changed). Otherwise, we'll try
// again later.
delta_offset = static_cast<ptrdiff_t>(n);
if (n == std::min(input.size(), size_t{max_read_size})) {
return result;
}
// else: "Fall through".
}
} else {
if (next && !switch_to_next_protocol())
return -1;
// Shove the unread bytes to the beginning of the buffer and continue
// to the next loop iteration.
result += consumed;
auto del = static_cast<size_t>(consumed);
delta_offset = static_cast<ptrdiff_t>(n - del);
input.erase(input.begin(), input.begin() + del);
}
}
return result;
}
// -- mock_web_socket_app ------------------------------------------------------
mock_web_socket_app::mock_web_socket_app(bool request_messages_on_start)
: request_messages_on_start(request_messages_on_start) {
// nop
}
caf::error mock_web_socket_app::start(caf::net::web_socket::lower_layer* ll) {
down = ll;
if (request_messages_on_start)
down->request_messages();
return caf::none;
}
void mock_web_socket_app::prepare_send() {
// nop
}
bool mock_web_socket_app::done_sending() {
return true;
}
caf::error
mock_web_socket_app::accept(const caf::net::http::request_header& hdr) {
// Store the request information in cfg to evaluate them later.
auto& ws = cfg["web-socket"].as_dictionary();
put(ws, "method", to_rfc_string(hdr.method()));
put(ws, "path", std::string{hdr.path()});
put(ws, "query", hdr.query());
put(ws, "fragment", hdr.fragment());
put(ws, "http-version", hdr.version());
if (hdr.num_fields() > 0) {
auto& fields = ws["fields"].as_dictionary();
hdr.for_each_field([&fields](auto key, auto val) {
put(fields, std::string{key}, std::string{val});
});
}
return caf::none;
}
void mock_web_socket_app::abort(const caf::error& reason) {
abort_reason = reason;
}
ptrdiff_t mock_web_socket_app::consume_text(std::string_view text) {
text_input.insert(text_input.end(), text.begin(), text.end());
return static_cast<ptrdiff_t>(text.size());
}
ptrdiff_t mock_web_socket_app::consume_binary(caf::byte_span bytes) {
binary_input.insert(binary_input.end(), bytes.begin(), bytes.end());
return static_cast<ptrdiff_t>(bytes.size());
}
// -- barrier ------------------------------------------------------------------
void barrier::arrive_and_wait() {
std::unique_lock<std::mutex> guard{mx_};
auto new_count = ++count_;
if (new_count == num_threads_) {
cv_.notify_all();
} else if (new_count > num_threads_) {
count_ = 1;
cv_.wait(guard, [this] { return count_.load() == num_threads_; });
} else {
cv_.wait(guard, [this] { return count_.load() == num_threads_; });
}
}
// -- main --------------------------------------------------------------------
int main(int argc, char** argv) {
net::this_host::startup();
net::ssl::startup();
using namespace caf;
net::middleman::init_global_meta_objects();
core::init_global_meta_objects();
auto result = test::main(argc, argv);
net::ssl::cleanup();
net::this_host::cleanup();
return result;
}
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