Commit 2b7b17b8 authored by Dominik Charousset's avatar Dominik Charousset

Fix initialization ob WebSocket flow bridges

parent 53269b12
...@@ -102,7 +102,8 @@ int caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -102,7 +102,8 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
// Open up a TCP port for incoming connections. // Open up a TCP port for incoming connections.
auto port = caf::get_or(cfg, "port", default_port); auto port = caf::get_or(cfg, "port", default_port);
cn::tcp_accept_socket fd; cn::tcp_accept_socket fd;
if (auto maybe_fd = cn::make_tcp_accept_socket({caf::ipv4_address{}, port})) { if (auto maybe_fd = cn::make_tcp_accept_socket({caf::ipv4_address{}, port},
true)) {
std::cout << "*** started listening for incoming connections on port " std::cout << "*** started listening for incoming connections on port "
<< port << '\n'; << port << '\n';
fd = std::move(*maybe_fd); fd = std::move(*maybe_fd);
......
...@@ -52,6 +52,7 @@ public: ...@@ -52,6 +52,7 @@ public:
void handle_read_event() override { void handle_read_event() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
CAF_ASSERT(owner_ != nullptr);
if (auto x = accept(fd_)) { if (auto x = accept(fd_)) {
socket_manager_ptr child = factory_.make(owner_->mpx_ptr(), *x); socket_manager_ptr child = factory_.make(owner_->mpx_ptr(), *x);
if (!child) { if (!child) {
......
...@@ -56,7 +56,9 @@ public: ...@@ -56,7 +56,9 @@ public:
static ptr_type make(buf_ptr buf, socket_manager_ptr mgr, action do_wakeup) { static ptr_type make(buf_ptr buf, socket_manager_ptr mgr, action do_wakeup) {
if (buf) { if (buf) {
auto adapter = ptr_type{new consumer_adapter(buf, mgr, CAF_ASSERT(mgr);
CAF_ASSERT(do_wakeup);
auto adapter = ptr_type{new consumer_adapter(buf, std::move(mgr),
std::move(do_wakeup)), // std::move(do_wakeup)), //
false}; false};
buf->set_consumer(adapter); buf->set_consumer(adapter);
......
...@@ -27,7 +27,7 @@ namespace caf::net::web_socket { ...@@ -27,7 +27,7 @@ namespace caf::net::web_socket {
/// 6455. Initially, the layer performs the WebSocket handshake. Once completed, /// 6455. Initially, the layer performs the WebSocket handshake. Once completed,
/// this layer decodes RFC 6455 frames and forwards binary and text messages to /// this layer decodes RFC 6455 frames and forwards binary and text messages to
/// `UpperLayer`. /// `UpperLayer`.
class client : public stream_oriented::upper_layer { class CAF_NET_EXPORT client : public stream_oriented::upper_layer {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
......
...@@ -24,22 +24,6 @@ using connect_event_t ...@@ -24,22 +24,6 @@ using connect_event_t
/// @param init Function object for setting up the created flows. /// @param init Function object for setting up the created flows.
template <class Transport = stream_transport, class Socket, class Init> template <class Transport = stream_transport, class Socket, class Init>
void connect(actor_system& sys, Socket fd, handshake hs, Init init) { void connect(actor_system& sys, Socket fd, handshake hs, Init init) {
/*
using trait_t = default_trait;
static_assert(std::is_invocable_v<Init, connect_event_t&&>,
"invalid signature found for init");
auto [ws_pull, app_push] = async::make_spsc_buffer_resource<frame>();
auto [app_pull, ws_push] = async::make_spsc_buffer_resource<frame>();
using app_t = flow_bridge<trait_t>;
using stack_t = Transport<client<app_t>>;
using conn_t = flow_connector_trivial_impl<trait_t>;
auto& mpx = sys.network_manager().mpx();
auto conn = std::make_shared<conn_t>(std::move(ws_pull), std::move(ws_push));
auto mgr = make_socket_manager<stack_t>(fd, &mpx, std::move(hs),
std::move(conn));
mpx.init(mgr);
init(connect_event_t{app_pull, app_push});
*/
// TODO: connect() should return a disposable to stop the WebSocket. // TODO: connect() should return a disposable to stop the WebSocket.
using trait_t = default_trait; using trait_t = default_trait;
static_assert(std::is_invocable_v<Init, connect_event_t&&>, static_assert(std::is_invocable_v<Init, connect_event_t&&>,
......
...@@ -87,6 +87,7 @@ public: ...@@ -87,6 +87,7 @@ public:
error init(net::socket_manager* mgr, web_socket::lower_layer* down, error init(net::socket_manager* mgr, web_socket::lower_layer* down,
const settings& cfg) override { const settings& cfg) override {
CAF_ASSERT(mgr != nullptr);
down_ = down; down_ = down;
auto [err, pull, push] = conn_->on_request(cfg); auto [err, pull, push] = conn_->on_request(cfg);
if (!err) { if (!err) {
......
...@@ -25,7 +25,7 @@ ...@@ -25,7 +25,7 @@
namespace caf::net::web_socket { namespace caf::net::web_socket {
/// Implements the WebSocket framing protocol as defined in RFC-6455. /// Implements the WebSocket framing protocol as defined in RFC-6455.
class framing : public web_socket::lower_layer { class CAF_NET_EXPORT framing : public web_socket::lower_layer {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
......
...@@ -30,6 +30,7 @@ socket_manager::~socket_manager() { ...@@ -30,6 +30,7 @@ socket_manager::~socket_manager() {
socket_manager_ptr socket_manager::make(multiplexer* mpx, socket handle, socket_manager_ptr socket_manager::make(multiplexer* mpx, socket handle,
event_handler_ptr handler) { event_handler_ptr handler) {
CAF_ASSERT(mpx != nullptr);
return make_counted<socket_manager>(mpx, handle, std::move(handler)); return make_counted<socket_manager>(mpx, handle, std::move(handler));
} }
......
...@@ -40,6 +40,7 @@ std::unique_ptr<client> client::make(handshake_ptr hs, upper_layer_ptr up) { ...@@ -40,6 +40,7 @@ std::unique_ptr<client> client::make(handshake_ptr hs, upper_layer_ptr up) {
error client::init(socket_manager* owner, stream_oriented::lower_layer* down, error client::init(socket_manager* owner, stream_oriented::lower_layer* down,
const settings& cfg) { const settings& cfg) {
CAF_ASSERT(owner != nullptr);
CAF_ASSERT(hs_ != nullptr); CAF_ASSERT(hs_ != nullptr);
framing_.init(owner, down); framing_.init(owner, down);
if (!hs_->has_mandatory_fields()) if (!hs_->has_mandatory_fields())
...@@ -47,6 +48,7 @@ error client::init(socket_manager* owner, stream_oriented::lower_layer* down, ...@@ -47,6 +48,7 @@ error client::init(socket_manager* owner, stream_oriented::lower_layer* down,
"handshake data lacks mandatory fields"); "handshake data lacks mandatory fields");
if (!hs_->has_valid_key()) if (!hs_->has_valid_key())
hs_->randomize_key(); hs_->randomize_key();
owner_ = owner;
cfg_ = cfg; cfg_ = cfg;
down->begin_output(); down->begin_output();
hs_->write_http_1_request(down->output_buffer()); hs_->write_http_1_request(down->output_buffer());
......
...@@ -24,7 +24,9 @@ std::unique_ptr<server> server::make(upper_layer_ptr up) { ...@@ -24,7 +24,9 @@ std::unique_ptr<server> server::make(upper_layer_ptr up) {
error server::init(socket_manager* owner, stream_oriented::lower_layer* down, error server::init(socket_manager* owner, stream_oriented::lower_layer* down,
const settings& cfg) { const settings& cfg) {
CAF_ASSERT(owner != nullptr);
framing_.init(owner, down); framing_.init(owner, down);
owner_ = owner;
cfg_ = cfg; cfg_ = cfg;
lower_layer().configure_read(receive_policy::up_to(handshake::max_http_size)); lower_layer().configure_read(receive_policy::up_to(handshake::max_http_size));
return none; return none;
......
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