Commit 53269b12 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/neverlord/flow-additions'

parents b7aedb1d df975797
......@@ -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/tcp_stream_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)
......@@ -329,7 +329,7 @@ struct resource_ctrl : ref_counted {
~resource_ctrl() {
if (buf) {
if constexpr (IsProducer) {
auto err = make_error(sec::invalid_upstream,
auto err = make_error(sec::disposed,
"producer_resource destroyed without opening it");
buf->abort(err);
} else {
......@@ -397,11 +397,19 @@ public:
}
}
/// Convenience function for calling
/// `ctx->make_observable().from_resource(*this)`.
template <class Coordinator>
auto observe_on(Coordinator* ctx) const {
return ctx->make_observable().from_resource(*this);
}
/// Calls `try_open` and on success immediately calls `close` on the buffer.
void close() {
if (auto buf = try_open())
buf->close();
}
explicit operator bool() const noexcept {
return ctrl_ != nullptr;
}
......@@ -454,6 +462,12 @@ public:
}
}
/// Calls `try_open` and on success immediately calls `cancel` on the buffer.
void cancel() {
if (auto buf = try_open())
buf->cancel();
}
explicit operator bool() const noexcept {
return ctrl_ != nullptr;
}
......
......@@ -91,7 +91,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));
......
......@@ -43,6 +43,10 @@ public:
static std::unique_ptr<client> make(handshake_ptr hs, upper_layer_ptr up);
static std::unique_ptr<client> make(handshake&& hs, upper_layer_ptr up) {
return make(std::make_unique<handshake>(std::move(hs)), std::move(up));
}
// -- properties -------------------------------------------------------------
web_socket::upper_layer& upper_layer() noexcept {
......
// 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/stream_transport.hpp"
#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 <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.
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 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 bridge = flow_bridge<trait_t>::make(std::move(conn));
auto impl = client::make(std::move(hs), std::move(bridge));
auto transport = Transport::make(fd, std::move(impl));
auto ptr = socket_manager::make(&mpx, fd, std::move(transport));
mpx.init(ptr);
init(connect_event_t{app_pull, app_push});
}
} // namespace caf::net::web_socket
......@@ -146,7 +146,8 @@ public:
void abort(const error& reason) override {
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