Commit 7ea378a5 authored by Dominik Charousset's avatar Dominik Charousset

Add web_socket::connect plus example

parent f1b05f02
......@@ -139,5 +139,6 @@ else()
endfunction()
endif()
add_net_example(web_socket quote-server)
add_net_example(web_socket echo)
add_net_example(web_socket hello-client)
add_net_example(web_socket quote-server)
// Simple WebSocket server that sends everything it receives back to the sender.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/web_socket/connect.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
#include <utility>
static constexpr uint16_t default_port = 8080;
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<std::string>("host", "TCP endpoint to connect to")
.add<uint16_t>("port,p", "port for connecting to the host")
.add<std::string>("endpoint", "sets the Request-URI field")
.add<std::string>("protocols", "sets the Sec-WebSocket-Protocol field");
}
};
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
// Sanity checking.
auto host = caf::get_or(cfg, "host", "");
if (host.empty()) {
std::cerr << "*** missing mandatory argument: host\n";
return EXIT_FAILURE;
}
// Ask user for the hello message.
std::string hello;
std::cout << "Please enter a hello message for the server: " << std::flush;
std::getline(std::cin, hello);
// Open up the TCP connection.
auto port = caf::get_or(cfg, "port", default_port);
cn::tcp_stream_socket fd;
if (auto maybe_fd = cn::make_connected_tcp_stream_socket(host, port)) {
std::cout << "*** connected to " << host << ':' << port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to connect to " << host << ':' << port << ": "
<< to_string(maybe_fd.error()) << '\n';
return EXIT_FAILURE;
}
// Spin up the WebSocket.
ws::handshake hs;
hs.host(host);
hs.endpoint(caf::get_or(cfg, "endpoint", "/"));
if (auto str = caf::get_as<std::string>(cfg, "protocols");
str && !str->empty())
hs.protocols(std::move(*str));
ws::connect(sys, fd, std::move(hs), [&](const ws::connect_event_t& conn) {
sys.spawn([conn, hello](caf::event_based_actor* self) {
auto [pull, push] = conn.data();
// Print everything from the server to stdout.
self->make_observable()
.from_resource(pull)
.do_on_error([](const caf::error& what) {
std::cerr << "*** error while reading from the WebSocket: "
<< to_string(what) << '\n';
})
.for_each([](const ws::frame& msg) {
if (msg.is_text()) {
std::cout << "Server: " << msg.as_text() << '\n';
} else if (msg.is_binary()) {
std::cout << "Server: [binary message of size "
<< msg.as_binary().size() << "]\n";
}
});
// Send our hello message.
self->make_observable().just(ws::frame{hello}).subscribe(push);
});
});
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
......@@ -90,7 +90,7 @@ void accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out,
"invalid signature found for on_request");
using factory_t = detail::ws_acceptor_factory<Transport, trait_t>;
using impl_t = connection_acceptor<Socket, factory_t>;
using conn_t = flow_connector_impl<OnRequest, trait_t, Ts...>;
using conn_t = flow_connector_request_impl<OnRequest, trait_t, Ts...>;
if (auto buf = out.try_open()) {
auto& mpx = sys.network_manager().mpx();
auto conn = std::make_shared<conn_t>(std::move(on_request), std::move(buf));
......
......@@ -41,7 +41,7 @@ public:
template <class... Ts>
explicit client(handshake hs, Ts&&... xs)
: upper_layer_(std::forward<Ts>(xs)...) {
: framing_(std::forward<Ts>(xs)...) {
handshake_.reset(new handshake(std::move(hs)));
}
......@@ -67,35 +67,35 @@ public:
// -- properties -------------------------------------------------------------
auto& upper_layer() noexcept {
return upper_layer_.upper_layer();
return framing_.upper_layer();
}
const auto& upper_layer() const noexcept {
return upper_layer_.upper_layer();
return framing_.upper_layer();
}
// -- role: upper layer ------------------------------------------------------
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
return handshake_complete() ? upper_layer_.prepare_send(down) : true;
return handshake_complete() ? framing_.prepare_send(down) : true;
}
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr down) {
return handshake_complete() ? upper_layer_.done_sending(down) : true;
return handshake_complete() ? framing_.done_sending(down) : true;
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr down, const error& reason) {
if (handshake_complete())
upper_layer_.abort(down, reason);
framing_.abort(down, reason);
}
template <class LowerLayerPtr>
void continue_reading(LowerLayerPtr down) {
if (handshake_complete())
upper_layer_.continue_reading(down);
framing_.continue_reading(down);
}
template <class LowerLayerPtr>
......@@ -104,7 +104,7 @@ public:
<< CAF_ARG2("bytes", input.size()));
// Short circuit to the framing layer after the handshake completed.
if (handshake_complete())
return upper_layer_.consume(down, input, delta);
return framing_.consume(down, input, delta);
// Check whether received a HTTP header or else wait for more data or abort
// when exceeding the maximum size.
auto [hdr, remainder] = http::v1::split_header(input);
......@@ -126,7 +126,7 @@ public:
} else {
CAF_LOG_DEBUG(CAF_ARG2("socket", down->handle().id)
<< CAF_ARG2("remainder", remainder.size()));
if (auto sub_result = upper_layer_.consume(down, remainder, remainder);
if (auto sub_result = framing_.consume(down, remainder, remainder);
sub_result >= 0) {
return hdr.size() + sub_result;
} else {
......@@ -148,7 +148,7 @@ private:
auto http_ok = handshake_->is_valid_http_1_response(http);
handshake_.reset();
if (http_ok) {
if (auto err = upper_layer_.init(owner_, down, cfg_)) {
if (auto err = framing_.init(owner_, down, cfg_)) {
CAF_LOG_DEBUG("failed to initialize WebSocket framing layer");
down->abort_reason(std::move(err));
return false;
......@@ -171,7 +171,7 @@ private:
std::unique_ptr<handshake> handshake_;
/// Stores the upper layer.
framing<UpperLayer> upper_layer_;
framing<UpperLayer> framing_;
/// Stores a pointer to the owning manager for the delayed initialization.
socket_manager* owner_ = nullptr;
......@@ -179,17 +179,4 @@ private:
settings cfg_;
};
template <template <class> class Transport = stream_transport, class Socket,
class T, class Trait>
void run_client(multiplexer& mpx, Socket fd, handshake hs,
async::consumer_resource<T> in, async::producer_resource<T> out,
Trait trait) {
using app_t = message_flow_bridge<T, Trait, tag::mixed_message_oriented>;
using stack_t = Transport<client<app_t>>;
auto mgr = make_socket_manager<stack_t>(fd, &mpx, std::move(hs),
std::move(in), std::move(out),
std::move(trait));
mpx.init(mgr);
}
} // namespace caf::net::web_socket
// 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/net/web_socket/client.hpp"
#include "caf/net/web_socket/default_trait.hpp"
#include "caf/net/web_socket/flow_bridge.hpp"
#include "caf/net/web_socket/frame.hpp"
#include "caf/net/web_socket/handshake.hpp"
namespace caf::net::web_socket {
/// Describes the one-time connection event.
using connect_event_t
= cow_tuple<async::consumer_resource<frame>, // Socket to App.
async::producer_resource<frame>>; // App to Socket.
/// Tries to establish a WebSocket connection on @p fd.
/// @param sys The host system.
/// @param fd An connected socket.
/// @param init Function object for setting up the created flows.
template <template <class> class Transport = stream_transport, class Socket,
class 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.
}
} // namespace caf::net::web_socket
......@@ -134,7 +134,8 @@ public:
void abort(LowerLayerPtr, const error& reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (reason == sec::socket_disconnected || reason == sec::disposed)
if (reason == sec::connection_closed || reason == sec::socket_disconnected
|| reason == sec::disposed)
out_->close();
else
out_->abort(reason);
......
......@@ -13,6 +13,7 @@
namespace caf::net::web_socket {
/// Connects a @ref flow_bridge to input and output buffers.
template <class Trait>
class flow_connector {
public:
......@@ -33,8 +34,10 @@ public:
template <class Trait>
using flow_connector_ptr = std::shared_ptr<flow_connector<Trait>>;
/// Calls an `OnRequest` handler with a @ref request object and passes the
/// generated buffers to the @ref flow_bridge.
template <class OnRequest, class Trait, class... Ts>
class flow_connector_impl : public flow_connector<Trait> {
class flow_connector_request_impl : public flow_connector<Trait> {
public:
using super = flow_connector<Trait>;
......@@ -47,7 +50,7 @@ public:
using producer_type = async::blocking_producer<app_res_type>;
template <class T>
flow_connector_impl(OnRequest&& on_request, T&& out)
flow_connector_request_impl(OnRequest&& on_request, T&& out)
: on_request_(std::move(on_request)), out_(std::forward<T>(out)) {
// nop
}
......@@ -73,4 +76,34 @@ private:
producer_type out_;
};
/// Trivial flow connector that passes its constructor arguments to the
/// @ref flow_bridge.
template <class Trait>
class flow_connector_trivial_impl : public flow_connector<Trait> {
public:
using super = flow_connector<Trait>;
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
using pull_t = async::consumer_resource<input_type>;
using push_t = async::producer_resource<output_type>;
using result_type = typename super::result_type;
flow_connector_trivial_impl(pull_t pull, push_t push)
: pull_(std::move(pull)), push_(std::move(push)) { // nop
}
result_type on_request(const settings&) override {
return {{}, std::move(pull_), std::move(push_)};
}
private:
pull_t pull_;
push_t push_;
};
} // namespace caf::net::web_socket
......@@ -10,7 +10,7 @@ template <class Trait>
class flow_connector;
template <class OnRequest, class Trait, class... Ts>
class flow_connector_impl;
class flow_connector_request_impl;
template <class Trait>
class flow_bridge;
......
......@@ -19,7 +19,7 @@ template <class Trait, class... Ts>
class request {
public:
template <class, class, class...>
friend class flow_connector_impl;
friend class flow_connector_request_impl;
using input_type = typename Trait::input_type;
......
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