Commit f411f91d authored by Dominik Charousset's avatar Dominik Charousset

Rename stream_oriented to octet_stream

parent bd962d17
......@@ -7,9 +7,6 @@
#include "caf/event_based_actor.hpp"
#include "caf/net/http/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
......
......@@ -6,9 +6,6 @@
#include "caf/event_based_actor.hpp"
#include "caf/net/lp/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/span.hpp"
#include "caf/uuid.hpp"
......
......@@ -6,9 +6,6 @@
#include "caf/event_based_actor.hpp"
#include "caf/net/lp/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/uuid.hpp"
......
......@@ -20,8 +20,6 @@
#include "caf/all.hpp"
#include "caf/net/lp/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_stream_socket.hpp"
CAF_PUSH_WARNINGS
#include "ui_chatwindow.h" // auto generated from chatwindow.ui
......
......@@ -5,9 +5,6 @@
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/web_socket/with.hpp"
#include "caf/scheduled_actor/flow.hpp"
......
......@@ -9,7 +9,6 @@
#include "caf/event_based_actor.hpp"
#include "caf/json_writer.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/web_socket/with.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/span.hpp"
......
......@@ -19,11 +19,11 @@ caf_add_component(
ENUM_TYPES
net.http.method
net.http.status
net.octet_stream.errc
net.ssl.dtls
net.ssl.errc
net.ssl.format
net.ssl.tls
net.stream_transport_error
net.web_socket.status
HEADERS
${CAF_NET_HEADERS}
......@@ -55,6 +55,10 @@ caf_add_component(
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/pollset_updater.cpp
src/net/prometheus/server.cpp
......@@ -72,9 +76,7 @@ caf_add_component(
src/net/ssl/tls.cpp
src/net/ssl/transport.cpp
src/net/ssl/verify.cpp
src/net/stream_oriented.cpp
src/net/stream_socket.cpp
src/net/stream_transport.cpp
src/net/tcp_accept_socket.cpp
src/net/tcp_stream_socket.cpp
src/net/this_host.cpp
......@@ -102,13 +104,13 @@ caf_add_component(
net.length_prefix_framing
net.multiplexer
net.network_socket
net.octet_stream.transport
net.pipe_socket
net.prometheus.server
net.socket
net.socket_guard
net.ssl.transport
net.stream_socket
net.stream_transport
net.tcp_socket
net.typed_actor_shell
net.udp_datagram_socket
......
......@@ -16,8 +16,6 @@ namespace caf::net {
// -- templates ----------------------------------------------------------------
class stream_transport;
template <class Factory>
class datagram_transport;
......@@ -81,9 +79,13 @@ make_actor_shell(actor_system&, async::execution_context_ptr);
} // namespace caf::net
namespace caf::net::stream_oriented {
namespace caf::net::octet_stream {
class lower_layer;
class policy;
class transport;
class upper_layer;
} // namespace caf::net::stream_oriented
enum class errc;
} // namespace caf::net::octet_stream
......@@ -18,9 +18,9 @@
#include "caf/net/http/upper_layer.hpp"
#include "caf/net/http/v1.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/octet_stream/upper_layer.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_oriented.hpp"
#include "caf/pec.hpp"
#include "caf/settings.hpp"
#include "caf/unordered_flat_map.hpp"
......@@ -30,7 +30,7 @@
namespace caf::net::http {
/// Implements the server part for the HTTP Protocol as defined in RFC 7231.
class CAF_NET_EXPORT server : public stream_oriented::upper_layer,
class CAF_NET_EXPORT server : public octet_stream::upper_layer,
public http::lower_layer {
public:
// -- member types -----------------------------------------------------------
......@@ -108,9 +108,9 @@ public:
bool send_end_of_chunks() override;
// -- stream_oriented::upper_layer implementation ----------------------------
// -- octet_stream::upper_layer implementation -------------------------------
error start(stream_oriented::lower_layer* down) override;
error start(octet_stream::lower_layer* down) override;
void abort(const error& reason) override;
......@@ -129,7 +129,7 @@ private:
bool handle_header(std::string_view http);
stream_oriented::lower_layer* down_;
octet_stream::lower_layer* down_;
upper_layer_ptr up_;
......
......@@ -17,8 +17,8 @@
#include "caf/net/http/server.hpp"
#include "caf/net/http/upper_layer.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/octet_stream/transport.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/none.hpp"
......
......@@ -18,7 +18,7 @@
#include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_oriented.hpp"
#include "caf/net/octet_stream/upper_layer.hpp"
#include <cstdint>
#include <cstring>
......@@ -32,7 +32,7 @@ namespace caf::net::lp {
/// but messages (including the 4 Bytes for the length prefix) are limited to a
/// maximum size of INT32_MAX. This limitation comes from the POSIX API (recv)
/// on 32-bit platforms.
class CAF_NET_EXPORT framing : public stream_oriented::upper_layer,
class CAF_NET_EXPORT framing : public octet_stream::upper_layer,
public binary::lower_layer {
public:
// -- member types -----------------------------------------------------------
......@@ -55,9 +55,9 @@ public:
static std::unique_ptr<framing> make(upper_layer_ptr up);
// -- implementation of stream_oriented::upper_layer -------------------------
// -- implementation of octet_stream::upper_layer ----------------------------
error start(stream_oriented::lower_layer* down) override;
error start(octet_stream::lower_layer* down) override;
void abort(const error& reason) override;
......@@ -94,7 +94,7 @@ public:
private:
// -- member variables -------------------------------------------------------
stream_oriented::lower_layer* down_;
octet_stream::lower_layer* down_;
upper_layer_ptr up_;
size_t message_offset_ = 0;
};
......
......@@ -15,8 +15,8 @@
#include "caf/net/http/server.hpp"
#include "caf/net/lp/framing.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/octet_stream/transport.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/none.hpp"
......
......@@ -13,9 +13,10 @@
#include <string>
#include <type_traits>
namespace caf::net {
namespace caf::net::octet_stream {
enum class stream_transport_error {
/// Holds error codes for the @ref octet_stream::transport.
enum class errc {
/// Indicates that the transport should try again later.
temporary,
/// Indicates that the transport must read data before trying again.
......@@ -26,20 +27,19 @@ enum class stream_transport_error {
permanent,
};
/// @relates stream_transport_error
CAF_NET_EXPORT std::string to_string(stream_transport_error);
/// @relates errc
CAF_NET_EXPORT std::string to_string(errc);
/// @relates stream_transport_error
CAF_NET_EXPORT bool from_string(std::string_view, stream_transport_error&);
/// @relates errc
CAF_NET_EXPORT bool from_string(std::string_view, errc&);
/// @relates stream_transport_error
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<stream_transport_error>,
stream_transport_error&);
/// @relates errc
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<errc>, errc&);
/// @relates stream_transport_error
/// @relates errc
template <class Inspector>
bool inspect(Inspector& f, stream_transport_error& x) {
bool inspect(Inspector& f, errc& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::net
} // namespace caf::net::octet_stream
......@@ -8,29 +8,8 @@
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/generic_lower_layer.hpp"
#include "caf/net/generic_upper_layer.hpp"
namespace caf::net::stream_oriented {
/// The upper layer requests bytes from the lower layer and consumes raw chunks
/// of data.
class CAF_NET_EXPORT upper_layer : public generic_upper_layer {
public:
virtual ~upper_layer();
/// Initializes the upper layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
virtual error start(lower_layer* down) = 0;
/// Consumes bytes from the lower layer.
/// @param buffer Available bytes to read.
/// @param delta Bytes that arrived since last calling this function.
/// @returns The number of consumed bytes. May be zero if waiting for more
/// input or negative to signal an error.
[[nodiscard]] virtual ptrdiff_t consume(byte_span buffer, byte_span delta)
= 0;
};
namespace caf::net::octet_stream {
/// Provides access to a resource that operates on a byte stream, e.g., a TCP
/// socket.
......@@ -61,4 +40,4 @@ public:
virtual bool end_output() = 0;
};
} // namespace caf::net::stream_oriented
} // namespace caf::net::octet_stream
// 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/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include <cstdint>
namespace caf::net::octet_stream {
/// Configures a @ref transport object. The default implementation simply
/// forwards to the free functions of @ref stream_socket.
class CAF_NET_EXPORT policy {
public:
virtual ~policy();
/// Reads data from the socket into the buffer.
virtual ptrdiff_t read(stream_socket x, byte_span buf);
/// Writes data from the buffer to the socket.
virtual ptrdiff_t write(stream_socket x, const_byte_span buf);
/// Returns the last socket error on this thread.
virtual errc last_error(stream_socket, ptrdiff_t);
/// Checks whether connecting a non-blocking socket was successful.
virtual ptrdiff_t connect(stream_socket x);
/// Convenience function that always returns 1. Exists to make writing code
/// against multiple policies easier by providing the same interface.
virtual ptrdiff_t accept(stream_socket);
/// Returns the number of bytes that are buffered internally and available
/// for immediate read.
virtual size_t buffered() const noexcept;
};
} // namespace caf::net::octet_stream
......@@ -11,15 +11,16 @@
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/octet_stream/lower_layer.hpp"
#include "caf/net/octet_stream/policy.hpp"
#include "caf/net/octet_stream/upper_layer.hpp"
#include "caf/net/socket_event_layer.hpp"
#include "caf/net/stream_oriented.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport_error.hpp"
namespace caf::net {
namespace caf::net::octet_stream {
class CAF_NET_EXPORT stream_transport : public socket_event_layer,
public stream_oriented::lower_layer {
class CAF_NET_EXPORT transport : public socket_event_layer,
public octet_stream::lower_layer {
public:
// -- member types -----------------------------------------------------------
......@@ -28,58 +29,7 @@ public:
using connection_handle = stream_socket;
/// An owning smart pointer type for storing an upper layer object.
using upper_layer_ptr = std::unique_ptr<stream_oriented::upper_layer>;
class policy {
public:
enum class ec {
/// Indicates that the transport should try again later.
temporary,
/// Indicates that the transport must read data before trying again.
want_read,
/// Indicates that the transport must write data before trying again.
want_write,
/// Indicates that the transport cannot resume this operation.
permanent,
};
virtual ~policy();
/// Reads data from the socket into the buffer.
virtual ptrdiff_t read(stream_socket x, byte_span buf) = 0;
/// Writes data from the buffer to the socket.
virtual ptrdiff_t write(stream_socket x, const_byte_span buf) = 0;
/// Returns the last socket error on this thread.
virtual stream_transport_error last_error(stream_socket, ptrdiff_t) = 0;
/// Checks whether connecting a non-blocking socket was successful.
virtual ptrdiff_t connect(stream_socket x) = 0;
/// Convenience function that always returns 1. Exists to make writing code
/// against multiple policies easier by providing the same interface.
virtual ptrdiff_t accept(stream_socket) = 0;
/// Returns the number of bytes that are buffered internally and available
/// for immediate read.
virtual size_t buffered() const noexcept = 0;
};
class default_policy : public policy {
public:
ptrdiff_t read(stream_socket x, byte_span buf) override;
ptrdiff_t write(stream_socket x, const_byte_span buf) override;
stream_transport_error last_error(stream_socket, ptrdiff_t) override;
ptrdiff_t connect(stream_socket x) override;
ptrdiff_t accept(stream_socket) override;
size_t buffered() const noexcept override;
};
using upper_layer_ptr = std::unique_ptr<octet_stream::upper_layer>;
/// Bundles various flags into a single block of memory.
struct flags_t {
......@@ -99,20 +49,19 @@ public:
// -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket fd, upper_layer_ptr up);
transport(stream_socket fd, upper_layer_ptr up);
stream_transport(stream_socket fd, upper_layer_ptr up, policy* custom);
transport(stream_socket fd, upper_layer_ptr up, policy* custom);
stream_transport(const stream_transport&) = delete;
transport(const transport&) = delete;
stream_transport& operator=(const stream_transport&) = delete;
transport& operator=(const transport&) = delete;
// -- factories --------------------------------------------------------------
static std::unique_ptr<stream_transport> make(stream_socket fd,
upper_layer_ptr up);
static std::unique_ptr<transport> make(stream_socket fd, upper_layer_ptr up);
// -- implementation of stream_oriented::lower_layer -------------------------
// -- implementation of octet_stream::lower_layer ----------------------------
bool can_send_more() const noexcept override;
......@@ -235,25 +184,9 @@ protected:
policy* policy_ = nullptr;
/// Fallback policy.
default_policy default_policy_;
policy default_policy_;
// TODO: add [[no_unique_address]] to default_policy_ when switching to C++20.
};
/// @relates stream_transport
CAF_NET_EXPORT std::string to_string(stream_transport::policy::ec);
/// @relates stream_transport
CAF_NET_EXPORT bool from_string(std::string_view,
stream_transport::policy::ec&);
/// @relates stream_transport
CAF_NET_EXPORT bool from_integer(int, stream_transport::policy::ec&);
/// @relates stream_transport
template <class Inspector>
bool inspect(Inspector& f, stream_transport::policy::ec& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::net
} // namespace caf::net::octet_stream
// 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/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/generic_upper_layer.hpp"
namespace caf::net::octet_stream {
/// The upper layer requests bytes from the lower layer and consumes raw chunks
/// of data.
class CAF_NET_EXPORT upper_layer : public generic_upper_layer {
public:
virtual ~upper_layer();
/// Initializes the upper layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
virtual error start(lower_layer* down) = 0;
/// Consumes bytes from the lower layer.
/// @param buffer Available bytes to read.
/// @param delta Bytes that arrived since last calling this function.
/// @returns The number of consumed bytes. May be zero if waiting for more
/// input or negative to signal an error.
[[nodiscard]] virtual ptrdiff_t consume(byte_span buffer, byte_span delta)
= 0;
};
} // namespace caf::net::octet_stream
......@@ -14,9 +14,9 @@
#include "caf/net/dsl/server_factory_base.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/octet_stream/transport.hpp"
#include "caf/net/prometheus/server.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/none.hpp"
......
......@@ -8,15 +8,15 @@
namespace caf::net {
/// Configures how many bytes a @ref stream_transport receives before calling
/// Configures how many bytes an octet stream transport receives before calling
/// `consume` on its upper layer.
struct receive_policy {
/// Configures how many bytes @ref stream_transport must read before it may
/// call `consume` on its upper layer.
/// Configures how many bytes the transport must read before it may call
/// `consume` on its upper layer.
uint32_t min_size;
/// Configures how many bytes @ref stream_transport may read at most before
/// it calls `consume` on its upper layer.
/// Configures how many bytes the transport may read at most before it calls
/// `consume` on its upper layer.
uint32_t max_size;
/// @pre `min_size > 0`
......
......@@ -21,8 +21,6 @@ public:
using transport_type = transport;
using accept_result_type = connection;
// -- constructors, destructors, and assignment operators --------------------
tcp_acceptor() = delete;
......
......@@ -5,24 +5,24 @@
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/net/octet_stream/transport.hpp"
#include "caf/net/ssl/connection.hpp"
#include "caf/net/stream_transport.hpp"
namespace caf::net::ssl {
/// Implements a stream_transport that manages a stream socket with encrypted
/// communication over OpenSSL.
class CAF_NET_EXPORT transport : public stream_transport {
/// Implements a octet stream transport that manages a stream socket with
/// encrypted communication over TLS.
class CAF_NET_EXPORT transport : public octet_stream::transport {
public:
// -- member types -----------------------------------------------------------
using super = stream_transport;
using super = octet_stream::transport;
using connection_handle = connection;
using worker_ptr = std::unique_ptr<socket_event_layer>;
class policy_impl : public stream_transport::policy {
class policy_impl : public octet_stream::policy {
public:
explicit policy_impl(connection conn);
......@@ -30,7 +30,7 @@ public:
ptrdiff_t write(stream_socket, const_byte_span) override;
stream_transport_error last_error(stream_socket, ptrdiff_t) override;
octet_stream::errc last_error(stream_socket, ptrdiff_t) override;
ptrdiff_t connect(stream_socket) override;
......
......@@ -25,7 +25,7 @@ struct CAF_NET_EXPORT stream_socket : network_socket {
using super = network_socket;
/// The default transport for exchanging raw bytes over a stream socket.
using transport_type = stream_transport;
using transport_type = octet_stream::transport;
using super::super;
......
......@@ -15,13 +15,13 @@ namespace caf::net {
/// Represents a TCP acceptor in listening mode.
struct CAF_NET_EXPORT tcp_accept_socket : network_socket {
/// The parent type.
using super = network_socket;
using super::super;
using transport_type = stream_transport;
/// The default transport for exchanging raw bytes over accepted sockets.
using transport_type = octet_stream::transport;
using accept_result_type = tcp_stream_socket;
using super::super;
};
/// Creates a new TCP socket to accept connections on a given port.
......
......@@ -12,8 +12,8 @@
#include "caf/logger.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/http/v1.hpp"
#include "caf/net/octet_stream/upper_layer.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_oriented.hpp"
#include "caf/net/web_socket/framing.hpp"
#include "caf/net/web_socket/handshake.hpp"
#include "caf/pec.hpp"
......@@ -27,7 +27,7 @@ namespace caf::net::web_socket {
/// 6455. Initially, the layer performs the WebSocket handshake. Once completed,
/// this layer decodes RFC 6455 frames and forwards binary and text messages to
/// `UpperLayer`.
class CAF_NET_EXPORT client : public stream_oriented::upper_layer {
class CAF_NET_EXPORT client : public octet_stream::upper_layer {
public:
// -- member types -----------------------------------------------------------
......@@ -71,11 +71,11 @@ public:
return static_cast<const client::upper_layer&>(framing_.up());
}
stream_oriented::lower_layer& down() noexcept {
octet_stream::lower_layer& down() noexcept {
return framing_.down();
}
const stream_oriented::lower_layer& down() const noexcept {
const octet_stream::lower_layer& down() const noexcept {
return framing_.down();
}
......@@ -83,9 +83,9 @@ public:
return hs_ == nullptr;
}
// -- implementation of stream_oriented::upper_layer -------------------------
// -- implementation of octet_stream::upper_layer ----------------------------
error start(stream_oriented::lower_layer* down) override;
error start(octet_stream::lower_layer* down) override;
void abort(const error& reason) override;
......
......@@ -8,8 +8,8 @@
#include "caf/byte_span.hpp"
#include "caf/detail/rfc6455.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/octet_stream/lower_layer.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_oriented.hpp"
#include "caf/net/web_socket/lower_layer.hpp"
#include "caf/net/web_socket/status.hpp"
#include "caf/net/web_socket/upper_layer.hpp"
......@@ -49,7 +49,7 @@ public:
// -- initialization ---------------------------------------------------------
void start(stream_oriented::lower_layer* down);
void start(octet_stream::lower_layer* down);
// -- properties -------------------------------------------------------------
......@@ -120,7 +120,7 @@ private:
// -- member variables -------------------------------------------------------
/// Points to the transport layer below.
stream_oriented::lower_layer* down_;
octet_stream::lower_layer* down_;
/// Buffer for assembling binary frames.
binary_buffer binary_buf_;
......
......@@ -15,9 +15,9 @@
#include "caf/net/http/status.hpp"
#include "caf/net/http/v1.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/octet_stream/upper_layer.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_oriented.hpp"
#include "caf/net/web_socket/framing.hpp"
#include "caf/net/web_socket/handshake.hpp"
#include "caf/net/web_socket/lower_layer.hpp"
......@@ -35,7 +35,7 @@ namespace caf::net::web_socket {
/// 6455. Initially, the layer performs the WebSocket handshake. Once completed,
/// this layer decodes RFC 6455 frames and forwards binary and text messages to
/// `UpperLayer`.
class CAF_NET_EXPORT server : public stream_oriented::upper_layer {
class CAF_NET_EXPORT server : public octet_stream::upper_layer {
public:
// -- member types -----------------------------------------------------------
......@@ -74,11 +74,11 @@ public:
return static_cast<const server::upper_layer&>(framing_.up());
}
stream_oriented::lower_layer& down() noexcept {
octet_stream::lower_layer& down() noexcept {
return framing_.down();
}
const stream_oriented::lower_layer& down() const noexcept {
const octet_stream::lower_layer& down() const noexcept {
return framing_.down();
}
......@@ -86,9 +86,9 @@ public:
return handshake_complete_;
}
// -- stream_oriented::upper_layer implementation ----------------------------
// -- octet_stream::upper_layer implementation -------------------------------
error start(stream_oriented::lower_layer* down) override;
error start(octet_stream::lower_layer* down) override;
void abort(const error& reason) override;
......
......@@ -16,8 +16,8 @@
#include "caf/net/http/header.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/octet_stream/transport.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/web_socket/server.hpp"
......
#include "caf/net/http/server.hpp"
#include "caf/net/octet_stream/lower_layer.hpp"
namespace caf::net::http {
// -- factories ----------------------------------------------------------------
......@@ -78,9 +80,9 @@ bool server::send_end_of_chunks() {
return down_->end_output();
}
// -- stream_oriented::upper_layer implementation ------------------------------
// -- octet_stream::upper_layer implementation ---------------------------------
error server::start(stream_oriented::lower_layer* down) {
error server::start(octet_stream::lower_layer* down) {
down_ = down;
return up_->start(this);
}
......
......@@ -3,6 +3,7 @@
#include "caf/detail/network_order.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp"
#include "caf/net/octet_stream/lower_layer.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/sec.hpp"
......@@ -14,9 +15,9 @@ std::unique_ptr<framing> framing::make(upper_layer_ptr up) {
return std::make_unique<framing>(std::move(up));
}
// -- implementation of stream_oriented::upper_layer ---------------------------
// -- implementation of octet_stream::upper_layer ------------------------------
error framing::start(stream_oriented::lower_layer* down) {
error framing::start(octet_stream::lower_layer* down) {
down_ = down;
return up_->start(this);
}
......
......@@ -2,16 +2,12 @@
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/net/stream_oriented.hpp"
#include "caf/net/octet_stream/lower_layer.hpp"
namespace caf::net::stream_oriented {
upper_layer::~upper_layer() {
// nop
}
namespace caf::net::octet_stream {
lower_layer::~lower_layer() {
// nop
}
} // namespace caf::net::stream_oriented
} // namespace caf::net::octet_stream
#include "caf/net/octet_stream/policy.hpp"
#include "caf/net/octet_stream/errc.hpp"
#include "caf/net/stream_socket.hpp"
namespace caf::net::octet_stream {
policy::~policy() {
// nop
}
ptrdiff_t policy::read(stream_socket x, byte_span buf) {
return net::read(x, buf);
}
ptrdiff_t policy::write(stream_socket x, const_byte_span buf) {
return net::write(x, buf);
}
errc policy::last_error(stream_socket, ptrdiff_t) {
return last_socket_error_is_temporary() ? errc::temporary : errc::permanent;
}
ptrdiff_t policy::connect(stream_socket x) {
// A connection is established if the OS reports a socket as ready for read
// or write and if there is no error on the socket.
return probe(x) ? 1 : -1;
}
ptrdiff_t policy::accept(stream_socket) {
return 1;
}
size_t policy::buffered() const noexcept {
return 0;
}
} // namespace caf::net::octet_stream
......@@ -2,77 +2,41 @@
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/net/stream_transport.hpp"
#include "caf/net/octet_stream/transport.hpp"
#include "caf/defaults.hpp"
#include "caf/net/octet_stream/errc.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/socket_manager.hpp"
namespace caf::net {
// -- member types -------------------------------------------------------------
stream_transport::policy::~policy() {
// nop
}
ptrdiff_t stream_transport::default_policy::read(stream_socket x,
byte_span buf) {
return net::read(x, buf);
}
ptrdiff_t stream_transport::default_policy::write(stream_socket x,
const_byte_span buf) {
return net::write(x, buf);
}
stream_transport_error
stream_transport::default_policy::last_error(stream_socket, ptrdiff_t) {
return last_socket_error_is_temporary() ? stream_transport_error::temporary
: stream_transport_error::permanent;
}
ptrdiff_t stream_transport::default_policy::connect(stream_socket x) {
// A connection is established if the OS reports a socket as ready for read
// or write and if there is no error on the socket.
return net::probe(x) ? 1 : -1;
}
ptrdiff_t stream_transport::default_policy::accept(stream_socket) {
return 1;
}
size_t stream_transport::default_policy::buffered() const noexcept {
return 0;
}
namespace caf::net::octet_stream {
// -- constructors, destructors, and assignment operators ----------------------
stream_transport::stream_transport(stream_socket fd, upper_layer_ptr up)
transport::transport(stream_socket fd, upper_layer_ptr up)
: fd_(fd), up_(std::move(up)), policy_(&default_policy_) {
memset(&flags_, 0, sizeof(flags_t));
}
stream_transport::stream_transport(stream_socket fd, upper_layer_ptr up,
policy* policy)
transport::transport(stream_socket fd, upper_layer_ptr up, policy* policy)
: fd_(fd), up_(std::move(up)), policy_(policy) {
memset(&flags_, 0, sizeof(flags_t));
}
// -- factories ----------------------------------------------------------------
std::unique_ptr<stream_transport> stream_transport::make(stream_socket fd,
upper_layer_ptr up) {
return std::make_unique<stream_transport>(fd, std::move(up));
std::unique_ptr<transport> transport::make(stream_socket fd,
upper_layer_ptr up) {
return std::make_unique<transport>(fd, std::move(up));
}
// -- implementation of stream_oriented::lower_layer ---------------------------
// -- implementation of octet_stream::lower_layer ------------------------------
bool stream_transport::can_send_more() const noexcept {
bool transport::can_send_more() const noexcept {
return write_buf_.size() < max_write_buf_size_;
}
void stream_transport::configure_read(receive_policy rd) {
void transport::configure_read(receive_policy rd) {
auto restarting = rd.max_size > 0 && max_read_size_ == 0;
min_read_size_ = rd.min_size;
max_read_size_ = rd.max_size;
......@@ -95,28 +59,28 @@ void stream_transport::configure_read(receive_policy rd) {
}
}
void stream_transport::begin_output() {
void transport::begin_output() {
if (write_buf_.empty())
parent_->register_writing();
}
byte_buffer& stream_transport::output_buffer() {
byte_buffer& transport::output_buffer() {
return write_buf_;
}
bool stream_transport::end_output() {
bool transport::end_output() {
return true;
}
bool stream_transport::is_reading() const noexcept {
bool transport::is_reading() const noexcept {
return max_read_size_ > 0;
}
void stream_transport::write_later() {
void transport::write_later() {
parent_->register_writing();
}
void stream_transport::shutdown() {
void transport::shutdown() {
if (write_buf_.empty()) {
parent_->shutdown();
} else {
......@@ -128,7 +92,7 @@ void stream_transport::shutdown() {
// -- implementation of transport ----------------------------------------------
error stream_transport::start(socket_manager* owner) {
error transport::start(socket_manager* owner) {
parent_ = owner;
// if (auto err = nodelay(fd_, true)) {
// CAF_LOG_ERROR("nodelay failed: " << err);
......@@ -145,11 +109,11 @@ error stream_transport::start(socket_manager* owner) {
return up_->start(this);
}
socket stream_transport::handle() const {
socket transport::handle() const {
return fd_;
}
void stream_transport::handle_read_event() {
void transport::handle_read_event() {
CAF_LOG_TRACE(CAF_ARG2("socket", fd_.id));
// Resume a write operation if the transport waited for the socket to be
// readable from the last call to handle_write_event.
......@@ -179,11 +143,11 @@ void stream_transport::handle_read_event() {
// Stop if we failed to get more data.
if (rd < 0) {
switch (policy_->last_error(fd_, rd)) {
case stream_transport_error::temporary:
case stream_transport_error::want_read:
case errc::temporary:
case errc::want_read:
// Try again later.
return;
case stream_transport_error::want_write:
case errc::want_write:
// Wait for writable socket and then call handle_read_event again.
flags_.wanted_write_from_read_event = true;
parent_->register_writing();
......@@ -214,7 +178,7 @@ void stream_transport::handle_read_event() {
handle_buffered_data();
}
void stream_transport::handle_buffered_data() {
void transport::handle_buffered_data() {
CAF_LOG_TRACE(CAF_ARG(buffered_));
// Loop until we have drained the buffer as much as we can.
CAF_ASSERT(min_read_size_ <= max_read_size_);
......@@ -262,7 +226,7 @@ void stream_transport::handle_buffered_data() {
}
}
void stream_transport::fail(const error& reason) {
void transport::fail(const error& reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
up_->abort(reason);
up_.reset();
......@@ -270,7 +234,7 @@ void stream_transport::fail(const error& reason) {
parent_->shutdown();
}
void stream_transport::handle_write_event() {
void transport::handle_write_event() {
CAF_LOG_TRACE(CAF_ARG2("socket", fd_.id));
// Resume a read operation if the transport waited for the socket to be
// writable from the last call to handle_read_event.
......@@ -310,10 +274,10 @@ void stream_transport::handle_write_event() {
// Try again later on temporary errors such as EWOULDBLOCK and
// stop writing to the socket on hard errors.
switch (policy_->last_error(fd_, write_res)) {
case stream_transport_error::temporary:
case stream_transport_error::want_write:
case errc::temporary:
case errc::want_write:
return;
case stream_transport_error::want_read:
case errc::want_read:
flags_.wanted_read_from_write_event = true;
parent_->register_reading();
parent_->deregister_writing();
......@@ -327,57 +291,13 @@ void stream_transport::handle_write_event() {
}
}
void stream_transport::abort(const error& reason) {
void transport::abort(const error& reason) {
up_->abort(reason);
flags_.shutting_down = true;
}
bool stream_transport::finalized() const noexcept {
bool transport::finalized() const noexcept {
return write_buf_.empty();
}
// -- free functions -----------------------------------------------------------
std::string to_string(stream_transport::policy::ec code) {
switch (code) {
case stream_transport::policy::ec::temporary:
return "temporary";
case stream_transport::policy::ec::want_read:
return "want_read";
case stream_transport::policy::ec::want_write:
return "want_write";
case stream_transport::policy::ec::permanent:
return "permanent";
default:
return "invalid";
}
}
bool from_string(std::string_view str, stream_transport::policy::ec& code) {
if (str == "temporary") {
code = stream_transport::policy::ec::temporary;
return true;
} else if (str == "want_read") {
code = stream_transport::policy::ec::want_read;
return true;
} else if (str == "want_write") {
code = stream_transport::policy::ec::want_write;
return true;
} else if (str == "permanent") {
code = stream_transport::policy::ec::permanent;
return true;
} else {
return false;
}
}
bool from_integer(int value, stream_transport::policy::ec& code) {
if (value >= 0 && value < 4) {
code = static_cast<stream_transport::policy::ec>(value);
return true;
} else {
return false;
}
}
} // namespace caf::net
} // namespace caf::net::octet_stream
// 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 "caf/net/octet_stream/upper_layer.hpp"
namespace caf::net::octet_stream {
upper_layer::~upper_layer() {
// nop
}
} // namespace caf::net::octet_stream
......@@ -4,9 +4,9 @@
#include "caf/net/ssl/transport.hpp"
#include "caf/net/octet_stream/errc.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/ssl/connection.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/settings.hpp"
CAF_PUSH_WARNINGS
......@@ -55,10 +55,10 @@ public:
} else {
auto err = policy_.last_error(policy_.conn.fd(), res);
switch (err) {
case stream_transport_error::want_read:
case stream_transport_error::temporary:
case octet_stream::errc::want_read:
case octet_stream::errc::temporary:
break;
case stream_transport_error::want_write:
case octet_stream::errc::want_write:
owner_->deregister_reading();
owner_->register_writing();
break;
......@@ -82,10 +82,10 @@ public:
owner_->deregister();
} else {
switch (policy_.last_error(policy_.conn.fd(), res)) {
case stream_transport_error::want_write:
case stream_transport_error::temporary:
case octet_stream::errc::want_write:
case octet_stream::errc::temporary:
break;
case stream_transport_error::want_read:
case octet_stream::errc::want_read:
owner_->deregister_writing();
owner_->register_reading();
break;
......@@ -141,25 +141,25 @@ ptrdiff_t transport::policy_impl::write(stream_socket, const_byte_span buf) {
return conn.write(buf);
}
stream_transport_error transport::policy_impl::last_error(stream_socket fd,
ptrdiff_t ret) {
octet_stream::errc transport::policy_impl::last_error(stream_socket fd,
ptrdiff_t ret) {
switch (conn.last_error(ret)) {
case errc::none:
case errc::want_accept:
case errc::want_connect:
// For all of these, OpenSSL docs say to do the operation again later.
return stream_transport_error::temporary;
return octet_stream::errc::temporary;
case errc::syscall_failed:
// Need to consult errno, which we just leave to the default policy.
return stream_transport::default_policy{}.last_error(fd, ret);
return octet_stream::policy{}.last_error(fd, ret);
case errc::want_read:
return stream_transport_error::want_read;
return octet_stream::errc::want_read;
case errc::want_write:
return stream_transport_error::want_write;
return octet_stream::errc::want_write;
default:
// Errors like SSL_ERROR_WANT_X509_LOOKUP are technically temporary, but
// we do not configure any callbacks. So seeing this is a red flag.
return stream_transport_error::permanent;
return octet_stream::errc::permanent;
}
}
......
......@@ -13,7 +13,6 @@
#include "caf/net/fwd.hpp"
#include "caf/net/http/v1.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_oriented.hpp"
#include "caf/net/web_socket/framing.hpp"
#include "caf/net/web_socket/handshake.hpp"
#include "caf/pec.hpp"
......@@ -42,9 +41,9 @@ std::unique_ptr<client> client::make(handshake_ptr hs, upper_layer_ptr up_ptr) {
return std::make_unique<client>(std::move(hs), std::move(up_ptr));
}
// -- implementation of stream_oriented::upper_layer ---------------------------
// -- implementation of octet_stream::upper_layer ------------------------------
error client::start(stream_oriented::lower_layer* down_ptr) {
error client::start(octet_stream::lower_layer* down_ptr) {
CAF_ASSERT(hs_ != nullptr);
framing_.start(down_ptr);
if (!hs_->has_mandatory_fields())
......
......@@ -10,7 +10,7 @@ namespace caf::net::web_socket {
// -- initialization ---------------------------------------------------------
void framing::start(stream_oriented::lower_layer* down) {
void framing::start(octet_stream::lower_layer* down) {
std::random_device rd;
rng_.seed(rd());
down_ = down;
......
......@@ -26,9 +26,9 @@ std::unique_ptr<server> server::make(upper_layer_ptr up) {
return std::make_unique<server>(std::move(up));
}
// -- stream_oriented::upper_layer implementation ------------------------------
// -- octet_stream::upper_layer implementation ---------------------------------
error server::start(stream_oriented::lower_layer* down_ptr) {
error server::start(octet_stream::lower_layer* down_ptr) {
framing_.start(down_ptr);
down().configure_read(receive_policy::up_to(handshake::max_http_size));
return none;
......
#pragma once
#include "caf/error.hpp"
#include "caf/net/octet_stream/lower_layer.hpp"
#include "caf/net/octet_stream/upper_layer.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/stream_oriented.hpp"
#include "caf/settings.hpp"
#include "caf/span.hpp"
#include "caf/string_view.hpp"
#include "caf/test/bdd_dsl.hpp"
class mock_stream_transport : public caf::net::stream_oriented::lower_layer {
class mock_stream_transport : public caf::net::octet_stream::lower_layer {
public:
// -- member types -----------------------------------------------------------
using upper_layer_ptr
= std::unique_ptr<caf::net::stream_oriented::upper_layer>;
using upper_layer_ptr = std::unique_ptr<caf::net::octet_stream::upper_layer>;
// -- constructors, destructors, and assignment operators --------------------
......@@ -28,7 +28,7 @@ public:
return std::make_unique<mock_stream_transport>(std::move(ptr));
}
// -- implementation of stream_oriented::lower_layer -------------------------
// -- implementation of octet_stream::lower_layer ----------------------------
bool can_send_more() const noexcept override;
......
......@@ -13,9 +13,9 @@
#include "caf/callback.hpp"
#include "caf/net/make_actor_shell.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/octet_stream/transport.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_transport.hpp"
using namespace caf;
......@@ -25,7 +25,7 @@ namespace {
using svec = std::vector<std::string>;
class app_t : public net::stream_oriented::upper_layer {
class app_t : public net::octet_stream::upper_layer {
public:
explicit app_t(actor_system& sys, async::execution_context_ptr loop,
actor hdl = {})
......@@ -38,7 +38,7 @@ public:
return std::make_unique<app_t>(sys, std::move(loop), std::move(hdl));
}
error start(net::stream_oriented::lower_layer* down) override {
error start(net::octet_stream::lower_layer* down) override {
this->down = down;
self->set_behavior([this](std::string& line) {
CAF_MESSAGE("received an asynchronous message: " << line);
......@@ -119,7 +119,7 @@ public:
}
// Pointer to the layer below.
net::stream_oriented::lower_layer* down;
net::octet_stream::lower_layer* down;
// Handle to the worker-under-test.
actor worker;
......@@ -200,7 +200,7 @@ CAF_TEST(actor shells expose their mailbox to their owners) {
auto fd = testee_socket_guard.release();
auto app_uptr = app_t::make(sys, mpx);
auto app = app_uptr.get();
auto transport = net::stream_transport::make(fd, std::move(app_uptr));
auto transport = net::octet_stream::transport::make(fd, std::move(app_uptr));
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
if (auto err = mgr->start())
CAF_FAIL("mgr->init() failed: " << err);
......@@ -222,7 +222,7 @@ CAF_TEST(actor shells can send requests and receive responses) {
auto fd = testee_socket_guard.release();
auto app_uptr = app_t::make(sys, mpx, worker);
auto app = app_uptr.get();
auto transport = net::stream_transport::make(fd, std::move(app_uptr));
auto transport = net::octet_stream::transport::make(fd, std::move(app_uptr));
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
if (auto err = mgr->start())
CAF_FAIL("mgr->start() failed: " << err);
......
......@@ -21,10 +21,10 @@
#include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/octet_stream/transport.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/span.hpp"
......@@ -195,7 +195,8 @@ SCENARIO("calling suspend_reading temporarily halts receiving of messages") {
auto app = app_t<true>::make(mpx, buf);
auto app_ptr = app.get();
auto framing = net::lp::framing::make(std::move(app));
auto transport = net::stream_transport::make(fd2, std::move(framing));
auto transport = net::octet_stream::transport::make(fd2,
std::move(framing));
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
CHECK_EQ(mgr->start(), none);
mpx->apply_updates();
......
......@@ -2,9 +2,9 @@
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE net.stream_transport
#define CAF_SUITE net.octet_stream.transport
#include "caf/net/stream_transport.hpp"
#include "caf/net/octet_stream/transport.hpp"
#include "net-test.hpp"
......@@ -22,6 +22,8 @@
using namespace caf;
namespace os = net::octet_stream;
namespace {
constexpr std::string_view hello_manager = "hello manager!";
......@@ -59,7 +61,7 @@ struct fixture : test_coordinator_fixture<> {
byte_buffer_ptr shared_send_buf;
};
class mock_application : public net::stream_oriented::upper_layer {
class mock_application : public os::upper_layer {
public:
using byte_buffer_ptr = std::shared_ptr<byte_buffer>;
......@@ -73,9 +75,9 @@ public:
std::move(send_buf));
}
net::stream_oriented::lower_layer* down;
os::lower_layer* down;
error start(net::stream_oriented::lower_layer* down_ptr) override {
error start(os::lower_layer* down_ptr) override {
down = down_ptr;
down->configure_read(net::receive_policy::exactly(hello_manager.size()));
return none;
......@@ -115,8 +117,8 @@ BEGIN_FIXTURE_SCOPE(fixture)
CAF_TEST(receive) {
auto mock = mock_application::make(shared_recv_buf, shared_send_buf);
auto transport = net::stream_transport::make(recv_socket_guard.release(),
std::move(mock));
auto transport = os::transport::make(recv_socket_guard.release(),
std::move(mock));
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
CHECK_EQ(mgr->start(), none);
mpx->apply_updates();
......@@ -133,8 +135,8 @@ CAF_TEST(receive) {
CAF_TEST(send) {
auto mock = mock_application::make(shared_recv_buf, shared_send_buf);
auto transport = net::stream_transport::make(recv_socket_guard.release(),
std::move(mock));
auto transport = os::transport::make(recv_socket_guard.release(),
std::move(mock));
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
CHECK_EQ(mgr->start(), none);
mpx->apply_updates();
......
......@@ -42,7 +42,7 @@ struct fixture {
}
};
class mock_application : public net::stream_oriented::upper_layer {
class mock_application : public net::octet_stream::upper_layer {
public:
explicit mock_application(std::shared_ptr<bool> done,
byte_buffer_ptr recv_buf)
......@@ -59,7 +59,7 @@ public:
*done_ = true;
}
error start(net::stream_oriented::lower_layer* down) override {
error start(net::octet_stream::lower_layer* down) override {
MESSAGE("initialize dummy app");
down_ = down;
down->configure_read(receive_policy::exactly(4));
......@@ -97,7 +97,7 @@ public:
}
private:
net::stream_oriented::lower_layer* down_ = nullptr;
net::octet_stream::lower_layer* down_ = nullptr;
std::shared_ptr<bool> done_;
byte_buffer_ptr recv_buf_;
};
......
......@@ -4,7 +4,7 @@
#define CAF_SUITE net.string_application
#include "caf/net/stream_transport.hpp"
#include "caf/net/octet_stream/transport.hpp"
#include "caf/test/dsl.hpp"
......@@ -186,7 +186,7 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) {
using application_type
= extend<string_application>::with<stream_string_application>;
using transport_type = stream_transport<application_type>;
using transport_type = octet_stream::transport<application_type>;
byte_buffer read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<byte_buffer>();
......
......@@ -13,9 +13,9 @@
#include "caf/callback.hpp"
#include "caf/net/make_actor_shell.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/octet_stream/transport.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/typed_actor.hpp"
using namespace caf;
......@@ -28,7 +28,7 @@ using svec = std::vector<std::string>;
using string_consumer = typed_actor<result<void>(std::string)>;
class app_t : public net::stream_oriented::upper_layer {
class app_t : public net::octet_stream::upper_layer {
public:
explicit app_t(actor_system& sys, async::execution_context_ptr loop,
actor hdl = {})
......@@ -41,7 +41,7 @@ public:
return std::make_unique<app_t>(sys, std::move(loop), std::move(hdl));
}
error start(net::stream_oriented::lower_layer* down) override {
error start(net::octet_stream::lower_layer* down) override {
this->down = down;
self->set_behavior([this](std::string& line) {
CAF_MESSAGE("received an asynchronous message: " << line);
......@@ -119,7 +119,7 @@ public:
}
// Pointer to the layer below.
net::stream_oriented::lower_layer* down;
net::octet_stream::lower_layer* down;
// Handle to the worker-under-test.
actor worker;
......@@ -200,7 +200,7 @@ CAF_TEST(actor shells expose their mailbox to their owners) {
auto fd = testee_socket_guard.release();
auto app_uptr = app_t::make(sys, mpx);
auto app = app_uptr.get();
auto transport = net::stream_transport::make(fd, std::move(app_uptr));
auto transport = net::octet_stream::transport::make(fd, std::move(app_uptr));
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
if (auto err = mgr->start())
CAF_FAIL("mgr->start() failed: " << err);
......@@ -222,7 +222,7 @@ CAF_TEST(actor shells can send requests and receive responses) {
auto fd = testee_socket_guard.release();
auto app_uptr = app_t::make(sys, mpx, worker);
auto app = app_uptr.get();
auto transport = net::stream_transport::make(fd, std::move(app_uptr));
auto transport = net::octet_stream::transport::make(fd, std::move(app_uptr));
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
if (auto err = mgr->start())
CAF_FAIL("mgr->start() failed: " << err);
......
......@@ -12,7 +12,6 @@
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
using namespace caf;
using namespace std::literals;
......
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