Commit 586922e6 authored by Jakob Otto's avatar Jakob Otto

WIP: stream_transport test

parent 4f70e957
diff --git a/libcaf_net/CMakeLists.txt b/libcaf_net/CMakeLists.txt
index 8c98f80..be7333c 100644
--- a/libcaf_net/CMakeLists.txt
+++ b/libcaf_net/CMakeLists.txt
@@ -50,8 +50,8 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/ip.cpp
src/message_queue.cpp
src/multiplexer.cpp
- src/net/backend/test.cpp
- src/net/backend/tcp.cpp
+ #src/net/backend/test.cpp
+ #src/net/backend/tcp.cpp
src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp
src/net/middleman_backend.cpp
@@ -124,28 +124,28 @@ target_link_libraries(caf-net-test PRIVATE CAF::test)
caf_incubator_add_test_suites(caf-net-test
accept_socket
- application
+ #application
convert_ip_endpoint
datagram_socket
- datagram_transport
- doorman
- endpoint_manager
+ #datagram_transport
+ #doorman
+ #endpoint_manager
header
ip
multiplexer
- net.backend.tcp
- net.basp.message_queue
- net.basp.ping_pong
- net.basp.worker
+ #net.backend.tcp
+ #net.basp.message_queue
+ #net.basp.ping_pong
+ #net.basp.worker
net.length_prefix_framing
network_socket
pipe_socket
socket
socket_guard
- stream_application
+ #stream_application
stream_socket
- stream_transport
- string_application
+ #stream_transport
+ #string_application
tcp_sockets
transport_worker
transport_worker_dispatcher
diff --git a/libcaf_net/src/net/backend/tcp.cpp b/libcaf_net/src/net/backend/tcp.cpp
index d1d2bd9..e84b011 100644
--- a/libcaf_net/src/net/backend/tcp.cpp
+++ b/libcaf_net/src/net/backend/tcp.cpp
@@ -25,6 +25,7 @@
#include "caf/net/basp/application.hpp"
#include "caf/net/basp/application_factory.hpp"
#include "caf/net/basp/ec.hpp"
+#include "caf/net/defaults.hpp"
#include "caf/net/doorman.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
diff --git a/libcaf_net/src/udp_datagram_socket.cpp b/libcaf_net/src/udp_datagram_socket.cpp
index d6362f3..58fe231 100644
--- a/libcaf_net/src/udp_datagram_socket.cpp
+++ b/libcaf_net/src/udp_datagram_socket.cpp
@@ -114,8 +114,9 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
ip_endpoint ep) {
sockaddr_storage addr = {};
detail::convert(ep, addr);
- auto len = ep.address().embeds_v4() ? sizeof(sockaddr_in)
- : sizeof(sockaddr_in6);
+ auto len = static_cast<socklen_t>(ep.address().embeds_v4()
+ ? sizeof(sockaddr_in)
+ : sizeof(sockaddr_in6));
auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), len);
auto ret = check_udp_datagram_socket_io_res(res);
@@ -172,7 +173,7 @@ variant<size_t, sec> write(udp_datagram_socket x, span<byte_buffer*> bufs,
message.msg_namelen = ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6);
message.msg_iov = buf_array;
- message.msg_iovlen = bufs.size();
+ message.msg_iovlen = static_cast<int>(bufs.size());
auto res = sendmsg(x.id, &message, 0);
return check_udp_datagram_socket_io_res(res);
}
...@@ -50,8 +50,8 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS} ...@@ -50,8 +50,8 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/ip.cpp src/ip.cpp
src/message_queue.cpp src/message_queue.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/net/backend/test.cpp #src/net/backend/test.cpp
src/net/backend/tcp.cpp #src/net/backend/tcp.cpp
src/net/endpoint_manager_queue.cpp src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp src/net/middleman.cpp
src/net/middleman_backend.cpp src/net/middleman_backend.cpp
...@@ -124,28 +124,28 @@ target_link_libraries(caf-net-test PRIVATE CAF::test) ...@@ -124,28 +124,28 @@ target_link_libraries(caf-net-test PRIVATE CAF::test)
caf_incubator_add_test_suites(caf-net-test caf_incubator_add_test_suites(caf-net-test
accept_socket accept_socket
application #application
convert_ip_endpoint convert_ip_endpoint
datagram_socket datagram_socket
datagram_transport #datagram_transport
doorman #doorman
endpoint_manager #endpoint_manager
header header
ip ip
multiplexer multiplexer
net.backend.tcp #net.backend.tcp
net.basp.message_queue #net.basp.message_queue
net.basp.ping_pong #net.basp.ping_pong
net.basp.worker #net.basp.worker
net.length_prefix_framing net.length_prefix_framing
network_socket network_socket
pipe_socket pipe_socket
socket socket
socket_guard socket_guard
stream_application #stream_application
stream_socket stream_socket
stream_transport stream_transport
string_application #string_application
tcp_sockets tcp_sockets
transport_worker transport_worker
transport_worker_dispatcher transport_worker_dispatcher
......
...@@ -40,8 +40,8 @@ public: ...@@ -40,8 +40,8 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
endpoint_manager_impl(const multiplexer_ptr& parent, actor_system& sys, endpoint_manager_impl(const multiplexer_ptr& parent, actor_system& sys,
Transport trans) socket handle, Transport trans)
: super(trans.handle(), parent, sys), transport_(std::move(trans)) { : super(handle, parent, sys), transport_(std::move(trans)) {
// nop // nop
} }
......
...@@ -94,6 +94,8 @@ public: ...@@ -94,6 +94,8 @@ public:
// -- pure virtual member functions ------------------------------------------ // -- pure virtual member functions ------------------------------------------
virtual error init(const settings& config) = 0;
/// Called whenever the socket received new data. /// Called whenever the socket received new data.
virtual bool handle_read_event() = 0; virtual bool handle_read_event() = 0;
...@@ -120,10 +122,19 @@ template <class Protocol> ...@@ -120,10 +122,19 @@ template <class Protocol>
class socket_manager_impl : public socket_manager { class socket_manager_impl : public socket_manager {
public: public:
template <class... Ts> template <class... Ts>
socket_manager_impl(Ts&&... xs) : protocol_(std::forward<Ts>(xs)...) { socket_manager_impl(socket handle, const multiplexer_ptr& mpx, Ts&&... xs)
: socket_manager{handle, mpx}, protocol_(std::forward<Ts>(xs)...) {
// nop // nop
} }
// -- initialization ---------------------------------------------------------
error init(const settings& config) override {
protocol_.init(*this, config);
}
// -- event callbacks --------------------------------------------------------
bool handle_read_event() override { bool handle_read_event() override {
return protocol_.handle_read_event(*this); return protocol_.handle_read_event(*this);
} }
...@@ -152,6 +163,24 @@ private: ...@@ -152,6 +163,24 @@ private:
/// @relates socket_manager /// @relates socket_manager
using socket_manager_ptr = intrusive_ptr<socket_manager>; using socket_manager_ptr = intrusive_ptr<socket_manager>;
template <class B, template <class> class... Layers>
struct socket_type_helper;
template <class B>
struct socket_type_helper<B> {
using type = typename B::socket_type;
};
template <class B, template <class> class Layer,
template <class> class... Layers>
struct socket_type_helper<B, Layer, Layers...>
: socket_type_helper<Layer<B>, Layers...> {
// no content
};
template <class B, template <class> class... Layers>
using socket_type_t = typename socket_type_helper<B, Layers...>::type;
template <class B, template <class> class... Layers> template <class B, template <class> class... Layers>
struct make_socket_manager_helper; struct make_socket_manager_helper;
...@@ -167,10 +196,17 @@ struct make_socket_manager_helper<B, Layer, Layers...> ...@@ -167,10 +196,17 @@ struct make_socket_manager_helper<B, Layer, Layers...>
// no content // no content
}; };
template <class B, template <class> class... Layers>
using make_socket_manager_helper_t =
typename make_socket_manager_helper<B, Layers...>::type;
template <class App, template <class> class... Layers, class... Ts> template <class App, template <class> class... Layers, class... Ts>
auto make_socket_manager(Ts&&... xs) { auto make_socket_manager(socket_type_t<App, Layers...> handle,
using impl = make_socket_manager_helper<App, Layers..., socket_manager_impl>; const multiplexer_ptr& mpx, Ts&&... xs) {
return make_counted<impl>(std::forward<Ts>(xs)...); static_assert(std::is_base_of<socket, socket_type_t<App, Layers...>>::value);
using impl
= make_socket_manager_helper_t<App, Layers..., socket_manager_impl>;
return make_counted<impl>(std::move(handle), mpx, std::forward<Ts>(xs)...);
} }
} // namespace caf::net } // namespace caf::net
...@@ -42,6 +42,8 @@ public: ...@@ -42,6 +42,8 @@ public:
using output_tag = tag::stream_oriented; using output_tag = tag::stream_oriented;
using socket_type = stream_socket;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
template <class... Ts> template <class... Ts>
......
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include "caf/net/basp/application.hpp" #include "caf/net/basp/application.hpp"
#include "caf/net/basp/application_factory.hpp" #include "caf/net/basp/application_factory.hpp"
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/doorman.hpp" #include "caf/net/doorman.hpp"
#include "caf/net/ip.hpp" #include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
......
...@@ -114,8 +114,9 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf, ...@@ -114,8 +114,9 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
ip_endpoint ep) { ip_endpoint ep) {
sockaddr_storage addr = {}; sockaddr_storage addr = {};
detail::convert(ep, addr); detail::convert(ep, addr);
auto len = ep.address().embeds_v4() ? sizeof(sockaddr_in) auto len = static_cast<socklen_t>(ep.address().embeds_v4()
: sizeof(sockaddr_in6); ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6));
auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), len); buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), len);
auto ret = check_udp_datagram_socket_io_res(res); auto ret = check_udp_datagram_socket_io_res(res);
...@@ -172,7 +173,7 @@ variant<size_t, sec> write(udp_datagram_socket x, span<byte_buffer*> bufs, ...@@ -172,7 +173,7 @@ variant<size_t, sec> write(udp_datagram_socket x, span<byte_buffer*> bufs,
message.msg_namelen = ep.address().embeds_v4() ? sizeof(sockaddr_in) message.msg_namelen = ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6); : sizeof(sockaddr_in6);
message.msg_iov = buf_array; message.msg_iov = buf_array;
message.msg_iovlen = bufs.size(); message.msg_iovlen = static_cast<int>(bufs.size());
auto res = sendmsg(x.id, &message, 0); auto res = sendmsg(x.id, &message, 0);
return check_udp_datagram_socket_io_res(res); return check_udp_datagram_socket_io_res(res);
} }
......
...@@ -147,13 +147,13 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture) ...@@ -147,13 +147,13 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) { CAF_TEST(receive) {
using transport_type = stream_transport<dummy_application>; using transport_type = stream_transport<dummy_application>;
auto mgr = make_endpoint_manager( auto mgr = make_socket_manager<dummy_application, stream_transport>(
mpx, sys, recv_socket_guard.release(), mpx, shared_buf);
transport_type{recv_socket_guard.release(), dummy_application{shared_buf}}); settings config;
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(config), none);
auto mgr_impl = mgr.downcast<endpoint_manager_impl<transport_type>>(); // auto mgr_impl = mgr.downcast<endpoint_manager_impl<transport_type>>();
CAF_CHECK(mgr_impl != nullptr); // CAF_CHECK(mgr_impl != nullptr);
auto& transport = mgr_impl->transport(); auto& transport = mgr->protocol();
transport.configure_read(receive_policy::exactly(hello_manager.size())); transport.configure_read(receive_policy::exactly(hello_manager.size()));
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_CHECK_EQUAL(write(send_socket_guard.socket(), CAF_CHECK_EQUAL(write(send_socket_guard.socket(),
......
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