Unverified Commit 28461e75 authored by Raphael's avatar Raphael Committed by GitHub

Merge pull request #1412

Fix initialization on Windows and enable Robot tests on Windows CI
parents 612e34fa aa7f2efb
......@@ -77,6 +77,13 @@ else
exit 1
fi
# Pick up Cirrus environment variables.
if [ -z "$CAF_NUM_CORES" ]; then
if [ ! -z "$CIRRUS_CPU" ]; then
CAF_NUM_CORES=$CIRRUS_CPU
fi
fi
runBuild() {
cat "$InitFile"
cd "$BuildDir"
......
......@@ -13,8 +13,14 @@ RUN Set-ExecutionPolicy Bypass -Scope Process -Force; \
RUN choco install -y --no-progress visualstudio2022buildtools
RUN choco install -y --no-progress visualstudio2022-workload-vctools
# Install Git and OpenSSL.
RUN choco install -y --no-progress msysgit openssl
# Install Git, OpenSSL and Python.
RUN choco install -y --no-progress msysgit openssl python
# Install Robot Framework and dependencies.
RUN python.exe -m pip install \
robotframework \
robotframework-requests \
robotframework-websocketclient
# Add git to the global PATH
SHELL [ "cmd", "/c" ]
......
......@@ -5,10 +5,10 @@ cmake.exe ^
-B build ^
-G "Visual Studio 17 2022" ^
-C ".ci\debug-flags.cmake" ^
-DEXTRA_FLAGS=/MP%CIRRUS_CPU% ^
-DCAF_ENABLE_ROBOT_TESTS=ON ^
-DBUILD_SHARED_LIBS=OFF ^
-DCMAKE_C_COMPILER=cl.exe ^
-DCMAKE_CXX_COMPILER=cl.exe ^
-DOPENSSL_ROOT_DIR="C:\Program Files\OpenSSL-Win64"
cmake.exe --build build --target install --config debug || exit \b 1
cmake.exe --build build --parallel %CIRRUS_CPU% --target install --config debug || exit \b 1
......@@ -17,6 +17,11 @@ is based on [Keep a Changelog](https://keepachangelog.com).
versions of OpenSSL). Please note that these lists are *not* recommended as
safe defaults, which is why we are no longer setting these values.
### Fixed
- Add missing initialization code for the new caf-net module when using the
`CAF_MAIN` macro. This fixes the `WSANOTINITIALISED` error on Windows (#1409).
## [0.19.1] - 2023-05-01
### Added
......
......@@ -29,7 +29,7 @@ void caf_main(caf::actor_system& sys, const config& cfg) {
// Only take the requested number of items from the infinite sequence.
.take(n)
// Print each integer.
.for_each([](int x) { std::cout << x << '\n'; });
.for_each([](int x) { std::cout << x << std::endl; });
});
}
// --(rst-main-end)--
......
......@@ -35,7 +35,7 @@ void caf_main(caf::actor_system& sys, const config& cfg) {
// Switch to `snk` for further processing.
.observe_on(snk)
// Print each integer.
.for_each([](int x) { std::cout << x << '\n'; });
.for_each([](int x) { std::cout << x << std::endl; });
// Allow the actors to run. After this point, we may no longer dereference
// the `src` and `snk` pointers! Calling these manually is optional. When
// removing these two lines, CAF automatically launches the actors at scope
......
......@@ -37,7 +37,7 @@ void sink(caf::event_based_actor* self, caf::async::consumer_resource<int> in) {
// Lift the input to an observable flow.
.from_resource(std::move(in))
// Print each integer.
.for_each([](int x) { std::cout << x << '\n'; });
.for_each([](int x) { std::cout << x << std::endl; });
}
// --(rst-sink-end)--
......
......@@ -77,7 +77,8 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
}
// Note: the actor system will only wait for actors on default. Since we don't
// start actors, we need to block on something else.
std::cout << "Server is up and running. Press <enter> to shut down.\n";
std::cout << "Server is up and running. Press <enter> to shut down."
<< std::endl;
getchar();
std::cout << "Terminating.\n";
return EXIT_SUCCESS;
......
......@@ -70,11 +70,13 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
pull
.observe_on(self) //
.do_on_error([](const caf::error& err) {
std::cout << "*** connection error: " << to_string(err) << '\n';
std::cout << "*** connection error: " << to_string(err)
<< std::endl;
})
.do_finally([self] {
std::cout << "*** lost connection to server -> quit\n"
<< "*** use CTRL+D or CTRL+C to terminate\n";
"*** use CTRL+D or CTRL+C to terminate"
<< std::endl;
self->quit();
})
.for_each([](const lp::frame& frame) {
......@@ -83,10 +85,10 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
auto str = std::string_view{
reinterpret_cast<const char*>(bytes.data()), bytes.size()};
if (std::all_of(str.begin(), str.end(), ::isprint)) {
std::cout << str << '\n';
std::cout << str << std::endl;
} else {
std::cout << "<non-ascii-data of size " << bytes.size()
<< ">\n";
std::cout << "<non-ascii-data of size " << bytes.size() << ">"
<< std::endl;
}
});
});
......
......@@ -49,7 +49,7 @@ void worker_impl(caf::event_based_actor* self,
messages.for_each([](const message_t& msg) {
const auto& [conn, frame] = msg;
std::cout << "*** got message of size " << frame.size() << " from "
<< to_string(conn) << '\n';
<< to_string(conn) << std::endl;
});
// Connect the flows for each incoming connection.
events
......@@ -58,7 +58,8 @@ void worker_impl(caf::event_based_actor* self,
[self, messages, pub = std::move(pub)](const auto& event) mutable {
// Each connection gets a unique ID.
auto conn = caf::uuid::random();
std::cout << "*** accepted new connection " << to_string(conn) << '\n';
std::cout << "*** accepted new connection " << to_string(conn)
<< std::endl;
auto& [pull, push] = event.data();
// Subscribe the `push` end to the central merge point.
messages
......@@ -72,14 +73,16 @@ void worker_impl(caf::event_based_actor* self,
})
.subscribe(push);
// Feed messages from the `pull` end into the central merge point.
auto inputs
auto inputs //
= pull.observe_on(self)
.do_on_error([](const caf::error& err) {
std::cout << "*** connection error: " << to_string(err) << '\n';
std::cout << "*** connection error: " << to_string(err)
<< std::endl;
})
.on_error_complete() // Cary on if a connection breaks.
.do_on_complete([conn] {
std::cout << "*** lost connection " << to_string(conn) << '\n';
std::cout << "*** lost connection " << to_string(conn)
<< std::endl;
})
.map([conn](const lp::frame& frame) {
return message_t{conn, frame};
......
......@@ -142,17 +142,17 @@ void caf_main(actor_system& sys) {
for (int row = 0; row < rows; ++row) {
for (int column = 0; column < columns; ++column)
cout << std::setw(4) << f(get_atom_v, row, column) << ' ';
cout << '\n';
cout << std::endl;
}
// Print out AVG for each row and column.
for (int row = 0; row < rows; ++row)
cout << "AVG(row " << row << ") = "
<< caf::to_string(f(get_atom_v, average_atom_v, row_atom_v, row))
<< '\n';
<< std::endl;
for (int column = 0; column < columns; ++column)
cout << "AVG(column " << column << ") = "
<< caf::to_string(f(get_atom_v, average_atom_v, column_atom_v, column))
<< '\n';
<< std::endl;
}
CAF_MAIN(id_block::fan_out_request)
......@@ -96,7 +96,7 @@ void client_repl(function_view<calculator> f) {
cout << " = "
<< caf::to_string(words[1] == "+" ? f(add_atom_v, *x, *y)
: f(sub_atom_v, *x, *y))
<< "\n";
<< std::endl;
}
}
......
......@@ -66,7 +66,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
// The header parameter contains fields from the WebSocket handshake
// such as the path and HTTP header fields..
auto path = acc.header().path();
std::cout << "*** new client request for path " << path << '\n';
std::cout << "*** new client request for path " << path << std::endl;
// Accept the WebSocket connection only if the path is "/".
if (path == "/") {
// Calling `accept` causes the server to acknowledge the client and
......@@ -95,20 +95,20 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
pull.observe_on(self)
.do_on_error([](const caf::error& what) { //
std::cout << "*** connection closed: " << to_string(what)
<< "\n";
<< std::endl;
})
.do_on_complete([] { //
std::cout << "*** connection closed\n";
std::cout << "*** connection closed" << std::endl;
})
.do_on_next([](const ws::frame& x) {
if (x.is_binary()) {
std::cout
<< "*** received a binary WebSocket frame of size "
<< x.size() << '\n';
<< x.size() << std::endl;
} else {
std::cout
<< "*** received a text WebSocket frame of size "
<< x.size() << '\n';
<< x.size() << std::endl;
}
})
.subscribe(push);
......
......@@ -52,7 +52,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
// Print errors to stderr.
.do_on_error([](const caf::error& what) {
std::cerr << "*** error while reading from the WebSocket: "
<< to_string(what) << '\n';
<< to_string(what) << std::endl;
})
// Restrict how many messages we receive if the user configured
// a limit.
......@@ -65,15 +65,15 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
})
// Print a bye-bye message if the server closes the connection.
.do_on_complete([] { //
std::cout << "Server has closed the connection\n";
std::cout << "Server has closed the connection" << std::endl;
})
// Print everything from the server to stdout.
.for_each([](const ws::frame& msg) {
if (msg.is_text()) {
std::cout << "Server: " << msg.as_text() << '\n';
std::cout << "Server: " << msg.as_text() << std::endl;
} else if (msg.is_binary()) {
std::cout << "Server: [binary message of size "
<< msg.as_binary().size() << "]\n";
<< msg.as_binary().size() << "]" << std::endl;
}
});
// Send our hello message and wait until the server closes the
......@@ -87,7 +87,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
});
if (!conn) {
std::cerr << "*** unable to connect to " << server->str() << ": "
<< to_string(conn.error()) << '\n';
<< to_string(conn.error()) << std::endl;
return EXIT_FAILURE;
}
// Note: the actor system will keep the application running for as long as the
......
......@@ -83,7 +83,7 @@ struct random_feed_state {
auto& x = update();
if (!writer.apply(x)) {
std::cerr << "*** failed to generate JSON: "
<< to_string(writer.get_error()) << '\n';
<< to_string(writer.get_error()) << std::endl;
return frame{};
}
return frame{writer.str()};
......@@ -95,18 +95,19 @@ struct random_feed_state {
.share();
// Subscribe once to start the feed immediately and to keep it running.
feed.for_each([n = 1](const frame&) mutable {
std::cout << "*** tick " << n++ << "\n";
std::cout << "*** tick " << n++ << std::endl;
});
// Add each incoming WebSocket listener to the feed.
auto n = std::make_shared<int>(0);
events
.observe_on(self) //
.for_each([this, n](const accept_event& ev) {
std::cout << "*** added listener (n = " << ++*n << ")\n";
std::cout << "*** added listener (n = " << ++*n << ")" << std::endl;
auto [pull, push] = ev.data();
pull.observe_on(self)
.do_finally([n] { //
std::cout << "*** removed listener (n = " << --*n << ")\n";
std::cout << "*** removed listener (n = " << --*n << ")"
<< std::endl;
})
.subscribe(std::ignore);
feed.subscribe(push);
......
......@@ -109,6 +109,39 @@ public:
actor_system(const actor_system&) = delete;
actor_system& operator=(const actor_system&) = delete;
/// Calls a cleanup function in its destructor for cleaning up global state.
class [[nodiscard]] global_state_guard {
public:
using void_fun_t = void (*)();
explicit global_state_guard(void_fun_t f) : fun_(f) {
// nop
}
global_state_guard(global_state_guard&& other) noexcept : fun_(other.fun_) {
other.fun_ = nullptr;
}
global_state_guard& operator=(global_state_guard&& other) noexcept {
std::swap(fun_, other.fun_);
return *this;
}
global_state_guard() = delete;
global_state_guard(const global_state_guard&) = delete;
global_state_guard& operator=(const global_state_guard&) = delete;
~global_state_guard() {
if (fun_ != nullptr)
fun_();
}
private:
void_fun_t fun_;
};
/// An (optional) component of the actor system.
class CAF_CORE_EXPORT module {
public:
......
......@@ -1060,6 +1060,23 @@ template <class T>
constexpr bool is_64bit_integer_v = std::is_same_v<T, int64_t>
|| std::is_same_v<T, uint64_t>;
/// Checks whether `T` has a static member function called `init_host_system`.
template <class T>
struct has_init_host_system {
template <class U>
static auto sfinae(U*) -> decltype(U::init_host_system(), std::true_type());
template <class U>
static auto sfinae(...) -> std::false_type;
using type = decltype(sfinae<T>(nullptr));
static constexpr bool value = type::value;
};
/// Convenience alias for `has_init_host_system<T>::value`.
template <class T>
constexpr bool has_init_host_system_v = has_init_host_system<T>::value;
} // namespace caf::detail
#undef CAF_HAS_MEMBER_TRAIT
......
......@@ -105,8 +105,28 @@ int exec_main(F fun, int argc, char** argv) {
} // namespace caf
namespace caf::detail {
template <class... Module>
auto do_init_host_system(type_list<Module...>, type_list<>) {
return std::make_tuple(Module::init_host_system()...);
}
template <class... Module, class T, class... Ts>
auto do_init_host_system(type_list<Module...>, type_list<T, Ts...>) {
if constexpr (detail::has_init_host_system_v<T>) {
return do_init_host_system(type_list<Module..., T>{}, type_list<Ts...>{});
} else {
return do_init_host_system(type_list<Module...>{}, type_list<Ts...>{});
}
}
} // namespace caf::detail
#define CAF_MAIN(...) \
int main(int argc, char** argv) { \
[[maybe_unused]] auto host_init_guard = caf::detail::do_init_host_system( \
caf::detail::type_list<>{}, caf::detail::type_list<__VA_ARGS__>{}); \
caf::exec_main_init_meta_objects<__VA_ARGS__>(); \
caf::core::init_global_meta_objects(); \
return caf::exec_main<__VA_ARGS__>(caf_main, argc, argv); \
......
......@@ -38,7 +38,7 @@ public:
/// Bundles the authority component of the URI, i.e., userinfo, host, and
/// port.
struct authority_type {
struct CAF_CORE_EXPORT authority_type {
std::string userinfo;
host_type host;
uint16_t port;
......
......@@ -569,9 +569,13 @@ void default_multiplexer::handle_socket_event(native_socket fd, int mask,
void default_multiplexer::init() {
#ifdef CAF_WINDOWS
WSADATA WinsockData;
if (WSAStartup(MAKEWORD(2, 2), &WinsockData) != 0) {
CAF_CRITICAL("WSAStartup failed");
// Note: when loading the caf-net module, users should call
// net::middleman::init_host_system() or net::this_host::startup().
if (!system().has_network_manager()) {
WSADATA WinsockData;
if (WSAStartup(MAKEWORD(2, 2), &WinsockData) != 0) {
CAF_CRITICAL("WSAStartup failed");
}
}
#endif
namespace sr = defaults::scheduler;
......@@ -631,7 +635,8 @@ default_multiplexer::~default_multiplexer() {
close_socket(pipe_reader_.fd());
pipe_reader_.init(invalid_native_socket);
#ifdef CAF_WINDOWS
WSACleanup();
if (!system().has_network_manager())
WSACleanup();
#endif
}
......
......@@ -23,16 +23,28 @@ namespace caf::net {
/// Provides a network backend for running protocol stacks.
class CAF_NET_EXPORT middleman : public actor_system::module {
public:
// -- constants --------------------------------------------------------------
/// Identifies the network manager module.
actor_system::module::id_t id_v = actor_system::module::network_manager;
// -- member types -----------------------------------------------------------
using module = actor_system::module;
using module_ptr = actor_system::module_ptr;
using void_fun_t = void (*)();
// -- static utility functions -----------------------------------------------
static void init_global_meta_objects();
/// Initializes global state for the network backend by calling
/// platform-dependent functions such as `WSAStartup` and `ssl::startup()`.
/// @returns a guard object shutting down the global state.
static actor_system::global_state_guard init_host_system();
// -- constructors, destructors, and assignment operators --------------------
explicit middleman(actor_system& sys);
......
......@@ -29,7 +29,7 @@ public:
virtual error start(lower_layer* down) = 0;
};
class upper_layer::server : public upper_layer {
class CAF_NET_EXPORT upper_layer::server : public upper_layer {
public:
virtual ~server();
......
......@@ -9,8 +9,10 @@
#include "caf/expected.hpp"
#include "caf/net/http/with.hpp"
#include "caf/net/prometheus.hpp"
#include "caf/net/ssl/startup.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/this_host.hpp"
#include "caf/raise_error.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
......@@ -57,6 +59,15 @@ void middleman::init_global_meta_objects() {
// nop
}
actor_system::global_state_guard middleman::init_host_system() {
this_host::startup();
ssl::startup();
return actor_system::global_state_guard{+[] {
ssl::cleanup();
this_host::cleanup();
}};
}
middleman::middleman(actor_system& sys)
: sys_(sys), mpx_(multiplexer::make(this)) {
// nop
......
......@@ -10,6 +10,26 @@ and assembling protocol stacks.
When using caf-net for applications, we generally recommend sticking to the
declarative API.
Initializing the Module
-----------------------
The networking module is an optional component of CAF. To use it, users can pass
``caf::net::middleman`` to ``CAF_MAIN``.
When not using the ``CAF_MAIN`` macro, users must initialize the library by:
- Calling ``caf::net::middleman::init_global_meta_objects()`` before
initializing the ``actor_system``. This step adds the necessary meta objects
to the global meta object registry.
- Calling ``caf::net::middleman::init_host_system()`` before initializing the
``actor_system`` (or calling any function of the module). This step runs
platform-specific initialization code (such as calling ``WSAStartup`` on
Windows) by calling ``caf::net::this_host::startup()`` and initializes the SSL
library by calling ``caf::net::ssl::startup()`` The function
``init_host_system`` returns a guard object that calls
``caf::net::this_host::cleanup()`` and ``caf::net::ssl::cleanup()`` in its
destructor.
Declarative High-level DSL :sup:`experimental`
----------------------------------------------
......
......@@ -49,16 +49,13 @@ HTTPS Test Delete Key Value Pair
*** Keywords ***
Start Servers
${res1}= Start Process ${BINARY_PATH} -p 55501
Set Suite Variable ${http_server_process} ${res1}
${res2}= Start Process ${BINARY_PATH} -p 55502 -k ${SSL_PATH}/key.pem -c ${SSL_PATH}/cert.pem
Set Suite Variable ${https_server_process} ${res2}
Start Process ${BINARY_PATH} -p 55501
Start Process ${BINARY_PATH} -p 55502 -k ${SSL_PATH}/key.pem -c ${SSL_PATH}/cert.pem
Wait Until Keyword Succeeds 5s 125ms Check If HTTP Server Is Reachable
Wait Until Keyword Succeeds 5s 125ms Check If HTTPS Server Is Reachable
Stop Servers
Terminate Process ${http_server_process}
Terminate Process ${https_server_process}
Run Keyword And Ignore Error Terminate All Processes
Check If HTTP Server Is Reachable
Log Try reaching ${HTTP_URL}/status.
......
......@@ -9,10 +9,10 @@ Suite Setup Start Servers
Suite Teardown Stop Servers
*** Variables ***
${HTTP_URL} http://localhost:55501
${HTTPS_URL} https://localhost:55502
${WS_URL} ws://localhost:55501
${WSS_URL} wss://localhost:55502
${HTTP_URL} http://localhost:55503
${HTTPS_URL} https://localhost:55504
${WS_URL} ws://localhost:55503
${WSS_URL} wss://localhost:55504
${FRAME_COUNT} ${10}
${BINARY_PATH} /path/to/the/server
${SSL_PATH} /path/to/the/pem/files
......@@ -37,16 +37,13 @@ Test WebSocket Over SSL Server
*** Keywords ***
Start Servers
${res1}= Start Process ${BINARY_PATH} -p 55501
Set Suite Variable ${ws_server_process} ${res1}
${res2}= Start Process ${BINARY_PATH} -p 55502 -k ${SSL_PATH}/key.pem -c ${SSL_PATH}/cert.pem
Set Suite Variable ${wss_server_process} ${res2}
Start Process ${BINARY_PATH} -p 55503
Start Process ${BINARY_PATH} -p 55504 -k ${SSL_PATH}/key.pem -c ${SSL_PATH}/cert.pem
Wait Until Keyword Succeeds 5s 125ms Check If HTTP Server Is Reachable
Wait Until Keyword Succeeds 5s 125ms Check If HTTPS Server Is Reachable
Stop Servers
Terminate Process ${ws_server_process}
Terminate Process ${wss_server_process}
Run Keyword And Ignore Error Terminate All Processes
Check If HTTP Server Is Reachable
# The server sends a "400 Bad Request" if we try HTTP on the WebSocket port.
......
......@@ -9,10 +9,10 @@ Suite Setup Start Servers
Suite Teardown Stop Servers
*** Variables ***
${HTTP_URL} http://localhost:55501
${HTTPS_URL} https://localhost:55502
${WS_URL} ws://localhost:55501/ws/quotes/seneca
${WSS_URL} wss://localhost:55502/ws/quotes/seneca
${HTTP_URL} http://localhost:55505
${HTTPS_URL} https://localhost:55506
${WS_URL} ws://localhost:55505/ws/quotes/seneca
${WSS_URL} wss://localhost:55506/ws/quotes/seneca
${FRAME_COUNT} ${10}
${BINARY_PATH} /path/to/the/server
${SSL_PATH} /path/to/the/pem/files
......@@ -45,16 +45,13 @@ Test WebSocket Over SSL Server Quotes
*** Keywords ***
Start Servers
${res1}= Start Process ${BINARY_PATH} -p 55501
Set Suite Variable ${ws_server_process} ${res1}
${res2}= Start Process ${BINARY_PATH} -p 55502 -k ${SSL_PATH}/key.pem -c ${SSL_PATH}/cert.pem
Set Suite Variable ${wss_server_process} ${res2}
Start Process ${BINARY_PATH} -p 55505
Start Process ${BINARY_PATH} -p 55506 -k ${SSL_PATH}/key.pem -c ${SSL_PATH}/cert.pem
Wait Until Keyword Succeeds 5s 125ms Check If HTTP Server Is Reachable
Wait Until Keyword Succeeds 5s 125ms Check If HTTPS Server Is Reachable
Stop Servers
Terminate Process ${ws_server_process}
Terminate Process ${wss_server_process}
Run Keyword And Ignore Error Terminate All Processes
Check If HTTP Server Is Reachable
Log Try reaching ${HTTP_URL}/status.
......
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