Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
A
Actor Framework
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Operations
Operations
Metrics
Environments
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
cpp-libs
Actor Framework
Commits
9e2cab29
Commit
9e2cab29
authored
May 13, 2020
by
Dominik Charousset
Committed by
GitHub
May 13, 2020
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #69
Refactor byte_buffer
parents
0bd912b4
a192ed03
Changes
30
Show whitespace changes
Inline
Side-by-side
Showing
30 changed files
with
157 additions
and
182 deletions
+157
-182
libcaf_net/caf/net/basp/application.hpp
libcaf_net/caf/net/basp/application.hpp
+2
-3
libcaf_net/caf/net/basp/header.hpp
libcaf_net/caf/net/basp/header.hpp
+1
-1
libcaf_net/caf/net/basp/worker.hpp
libcaf_net/caf/net/basp/worker.hpp
+6
-8
libcaf_net/caf/net/datagram_transport.hpp
libcaf_net/caf/net/datagram_transport.hpp
+11
-12
libcaf_net/caf/net/doorman.hpp
libcaf_net/caf/net/doorman.hpp
+3
-6
libcaf_net/caf/net/endpoint_manager.hpp
libcaf_net/caf/net/endpoint_manager.hpp
+0
-1
libcaf_net/caf/net/packet_writer.hpp
libcaf_net/caf/net/packet_writer.hpp
+5
-9
libcaf_net/caf/net/packet_writer_decorator.hpp
libcaf_net/caf/net/packet_writer_decorator.hpp
+5
-4
libcaf_net/caf/net/stream_transport.hpp
libcaf_net/caf/net/stream_transport.hpp
+10
-12
libcaf_net/caf/net/transport_base.hpp
libcaf_net/caf/net/transport_base.hpp
+7
-8
libcaf_net/caf/net/udp_datagram_socket.hpp
libcaf_net/caf/net/udp_datagram_socket.hpp
+4
-3
libcaf_net/src/application.cpp
libcaf_net/src/application.cpp
+2
-2
libcaf_net/src/endpoint_manager.cpp
libcaf_net/src/endpoint_manager.cpp
+0
-1
libcaf_net/src/header.cpp
libcaf_net/src/header.cpp
+2
-1
libcaf_net/src/ip.cpp
libcaf_net/src/ip.cpp
+2
-7
libcaf_net/src/udp_datagram_socket.cpp
libcaf_net/src/udp_datagram_socket.cpp
+5
-4
libcaf_net/test/application.cpp
libcaf_net/test/application.cpp
+8
-10
libcaf_net/test/datagram_transport.cpp
libcaf_net/test/datagram_transport.cpp
+12
-18
libcaf_net/test/endpoint_manager.cpp
libcaf_net/test/endpoint_manager.cpp
+11
-9
libcaf_net/test/header.cpp
libcaf_net/test/header.cpp
+2
-1
libcaf_net/test/multiplexer.cpp
libcaf_net/test/multiplexer.cpp
+3
-3
libcaf_net/test/net/basp/worker.cpp
libcaf_net/test/net/basp/worker.cpp
+2
-1
libcaf_net/test/pipe_socket.cpp
libcaf_net/test/pipe_socket.cpp
+4
-5
libcaf_net/test/stream_application.cpp
libcaf_net/test/stream_application.cpp
+5
-7
libcaf_net/test/stream_socket.cpp
libcaf_net/test/stream_socket.cpp
+7
-6
libcaf_net/test/stream_transport.cpp
libcaf_net/test/stream_transport.cpp
+8
-11
libcaf_net/test/string_application.cpp
libcaf_net/test/string_application.cpp
+7
-9
libcaf_net/test/transport_worker.cpp
libcaf_net/test/transport_worker.cpp
+2
-1
libcaf_net/test/transport_worker_dispatcher.cpp
libcaf_net/test/transport_worker_dispatcher.cpp
+14
-13
libcaf_net/test/udp_datagram_socket.cpp
libcaf_net/test/udp_datagram_socket.cpp
+7
-6
No files found.
libcaf_net/caf/net/basp/application.hpp
View file @
9e2cab29
...
@@ -33,6 +33,7 @@
...
@@ -33,6 +33,7 @@
#include "caf/defaults.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/error.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/net/basp/header.hpp"
...
@@ -55,8 +56,6 @@ class application {
...
@@ -55,8 +56,6 @@ class application {
public:
public:
// -- member types -----------------------------------------------------------
// -- member types -----------------------------------------------------------
using
buffer_type
=
std
::
vector
<
byte
>
;
using
byte_span
=
span
<
const
byte
>
;
using
byte_span
=
span
<
const
byte
>
;
using
hub_type
=
detail
::
worker_hub
<
worker
>
;
using
hub_type
=
detail
::
worker_hub
<
worker
>
;
...
@@ -175,7 +174,7 @@ private:
...
@@ -175,7 +174,7 @@ private:
byte_span
received
);
byte_span
received
);
/// Writes the handshake payload to `buf_`.
/// Writes the handshake payload to `buf_`.
error
generate_handshake
(
b
uffer_type
&
buf
);
error
generate_handshake
(
b
yte_buffer
&
buf
);
// -- member variables -------------------------------------------------------
// -- member variables -------------------------------------------------------
...
...
libcaf_net/caf/net/basp/header.hpp
View file @
9e2cab29
...
@@ -78,7 +78,7 @@ std::array<byte, header_size> to_bytes(header x);
...
@@ -78,7 +78,7 @@ std::array<byte, header_size> to_bytes(header x);
/// Serializes a header to a byte representation.
/// Serializes a header to a byte representation.
/// @relates header
/// @relates header
void
to_bytes
(
header
x
,
std
::
vector
<
byte
>
&
buf
);
void
to_bytes
(
header
x
,
byte_buffer
&
buf
);
/// @relates header
/// @relates header
template
<
class
Inspector
>
template
<
class
Inspector
>
...
...
libcaf_net/caf/net/basp/worker.hpp
View file @
9e2cab29
...
@@ -20,11 +20,12 @@
...
@@ -20,11 +20,12 @@
#include <atomic>
#include <atomic>
#include <cstdint>
#include <cstdint>
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/config.hpp"
#include "caf/config.hpp"
#include "caf/detail/abstract_worker.hpp"
#include "caf/detail/abstract_worker.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/fwd.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/net/basp/remote_message_handler.hpp"
#include "caf/net/basp/remote_message_handler.hpp"
...
@@ -48,8 +49,6 @@ public:
...
@@ -48,8 +49,6 @@ public:
using
scheduler_type
=
scheduler
::
abstract_coordinator
;
using
scheduler_type
=
scheduler
::
abstract_coordinator
;
using
buffer_type
=
std
::
vector
<
byte
>
;
using
hub_type
=
detail
::
worker_hub
<
worker
>
;
using
hub_type
=
detail
::
worker_hub
<
worker
>
;
// -- constructors, destructors, and assignment operators --------------------
// -- constructors, destructors, and assignment operators --------------------
...
@@ -72,9 +71,8 @@ private:
...
@@ -72,9 +71,8 @@ private:
// -- constants and assertions -----------------------------------------------
// -- constants and assertions -----------------------------------------------
/// Stores how many bytes the "first half" of this object requires.
/// Stores how many bytes the "first half" of this object requires.
static
constexpr
size_t
pointer_members_size
=
sizeof
(
hub_type
*
)
static
constexpr
size_t
pointer_members_size
+
sizeof
(
message_queue
*
)
=
sizeof
(
hub_type
*
)
+
sizeof
(
message_queue
*
)
+
sizeof
(
proxy_registry
*
)
+
sizeof
(
proxy_registry
*
)
+
sizeof
(
actor_system
*
);
+
sizeof
(
actor_system
*
);
static_assert
(
CAF_CACHE_LINE_SIZE
>
pointer_members_size
,
static_assert
(
CAF_CACHE_LINE_SIZE
>
pointer_members_size
,
...
@@ -108,7 +106,7 @@ private:
...
@@ -108,7 +106,7 @@ private:
header
hdr_
;
header
hdr_
;
/// Contains whatever this worker deserializes next.
/// Contains whatever this worker deserializes next.
b
uffer_type
payload_
;
b
yte_buffer
payload_
;
};
};
}
// namespace caf::net::basp
}
// namespace caf::net::basp
libcaf_net/caf/net/datagram_transport.hpp
View file @
9e2cab29
...
@@ -21,6 +21,7 @@
...
@@ -21,6 +21,7 @@
#include <deque>
#include <deque>
#include <vector>
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/fwd.hpp"
#include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
#include "caf/logger.hpp"
...
@@ -34,18 +35,18 @@
...
@@ -34,18 +35,18 @@
namespace
caf
::
net
{
namespace
caf
::
net
{
template
<
class
Factory
>
template
<
class
Factory
>
using
datagram_transport_base
=
transport_base
<
using
datagram_transport_base
datagram_transport
<
Factory
>
,
=
transport_base
<
datagram_transport
<
Factory
>
,
transport_worker_dispatcher
<
Factory
,
ip_endpoint
>
,
udp_datagram_socket
,
transport_worker_dispatcher
<
Factory
,
ip_endpoint
>
,
Factory
,
ip_endpoint
>
;
udp_datagram_socket
,
Factory
,
ip_endpoint
>
;
/// Implements a udp_transport policy that manages a datagram socket.
/// Implements a udp_transport policy that manages a datagram socket.
template
<
class
Factory
>
template
<
class
Factory
>
class
datagram_transport
:
public
datagram_transport_base
<
Factory
>
{
class
datagram_transport
:
public
datagram_transport_base
<
Factory
>
{
public:
public:
// Maximal UDP-packet size
// Maximal UDP-packet size
static
constexpr
size_t
max_datagram_size
=
std
::
numeric_limits
<
static
constexpr
size_t
max_datagram_size
uint16_t
>::
max
();
=
std
::
numeric_limits
<
uint16_t
>::
max
();
// -- member types -----------------------------------------------------------
// -- member types -----------------------------------------------------------
...
@@ -57,8 +58,6 @@ public:
...
@@ -57,8 +58,6 @@ public:
using
super
=
datagram_transport_base
<
factory_type
>
;
using
super
=
datagram_transport_base
<
factory_type
>
;
using
buffer_type
=
typename
super
::
buffer_type
;
using
buffer_cache_type
=
typename
super
::
buffer_cache_type
;
using
buffer_cache_type
=
typename
super
::
buffer_cache_type
;
// -- constructors, destructors, and assignment operators --------------------
// -- constructors, destructors, and assignment operators --------------------
...
@@ -127,7 +126,7 @@ public:
...
@@ -127,7 +126,7 @@ public:
return
none
;
return
none
;
}
}
void
write_packet
(
id_type
id
,
span
<
b
uffer_type
*>
buffers
)
override
{
void
write_packet
(
id_type
id
,
span
<
b
yte_buffer
*>
buffers
)
override
{
CAF_LOG_TRACE
(
""
);
CAF_LOG_TRACE
(
""
);
CAF_ASSERT
(
!
buffers
.
empty
());
CAF_ASSERT
(
!
buffers
.
empty
());
if
(
packet_queue_
.
empty
())
if
(
packet_queue_
.
empty
())
...
@@ -143,7 +142,7 @@ public:
...
@@ -143,7 +142,7 @@ public:
buffer_cache_type
bytes
;
buffer_cache_type
bytes
;
size_t
size
;
size_t
size
;
packet
(
id_type
id
,
span
<
b
uffer_type
*>
bufs
)
:
id
(
id
)
{
packet
(
id_type
id
,
span
<
b
yte_buffer
*>
bufs
)
:
id
(
id
)
{
size
=
0
;
size
=
0
;
for
(
auto
buf
:
bufs
)
{
for
(
auto
buf
:
bufs
)
{
size
+=
buf
->
size
();
size
+=
buf
->
size
();
...
@@ -151,8 +150,8 @@ public:
...
@@ -151,8 +150,8 @@ public:
}
}
}
}
std
::
vector
<
std
::
vector
<
byte
>
*>
get_buffer_ptrs
()
{
std
::
vector
<
byte_buffer
*>
get_buffer_ptrs
()
{
std
::
vector
<
std
::
vector
<
byte
>
*>
ptrs
;
std
::
vector
<
byte_buffer
*>
ptrs
;
for
(
auto
&
buf
:
bytes
)
for
(
auto
&
buf
:
bytes
)
ptrs
.
emplace_back
(
&
buf
);
ptrs
.
emplace_back
(
&
buf
);
return
ptrs
;
return
ptrs
;
...
...
libcaf_net/caf/net/doorman.hpp
View file @
9e2cab29
...
@@ -18,8 +18,6 @@
...
@@ -18,8 +18,6 @@
#pragma once
#pragma once
#include <vector>
#include "caf/logger.hpp"
#include "caf/logger.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/socket.hpp"
...
@@ -77,10 +75,9 @@ public:
...
@@ -77,10 +75,9 @@ public:
CAF_LOG_DEBUG
(
"unable to get multiplexer from parent"
);
CAF_LOG_DEBUG
(
"unable to get multiplexer from parent"
);
return
false
;
return
false
;
}
}
auto
child
=
make_endpoint_manager
(
mpx
,
parent
.
system
(),
auto
child
=
make_endpoint_manager
(
stream_transport
<
mpx
,
parent
.
system
(),
application_type
>
{
*
x
,
stream_transport
<
application_type
>
{
*
x
,
factory_
.
make
()});
factory_
.
make
()});
if
(
auto
err
=
child
->
init
())
if
(
auto
err
=
child
->
init
())
return
false
;
return
false
;
return
true
;
return
true
;
...
...
libcaf_net/caf/net/endpoint_manager.hpp
View file @
9e2cab29
...
@@ -24,7 +24,6 @@
...
@@ -24,7 +24,6 @@
#include "caf/actor.hpp"
#include "caf/actor.hpp"
#include "caf/actor_clock.hpp"
#include "caf/actor_clock.hpp"
#include "caf/byte.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/drr_queue.hpp"
...
...
libcaf_net/caf/net/packet_writer.hpp
View file @
9e2cab29
...
@@ -18,10 +18,8 @@
...
@@ -18,10 +18,8 @@
#pragma once
#pragma once
#include <vector>
#include "caf/byte.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/span.hpp"
#include "caf/span.hpp"
...
@@ -30,15 +28,13 @@ namespace caf::net {
...
@@ -30,15 +28,13 @@ namespace caf::net {
/// Implements an interface for packet writing in application-layers.
/// Implements an interface for packet writing in application-layers.
class
CAF_NET_EXPORT
packet_writer
{
class
CAF_NET_EXPORT
packet_writer
{
public:
public:
using
buffer_type
=
std
::
vector
<
byte
>
;
virtual
~
packet_writer
();
virtual
~
packet_writer
();
/// Returns a buffer for writing header information.
/// Returns a buffer for writing header information.
virtual
b
uffer_type
next_header_buffer
()
=
0
;
virtual
b
yte_buffer
next_header_buffer
()
=
0
;
/// Returns a buffer for writing payload content.
/// Returns a buffer for writing payload content.
virtual
b
uffer_type
next_payload_buffer
()
=
0
;
virtual
b
yte_buffer
next_payload_buffer
()
=
0
;
/// Convenience function to write a packet consisting of multiple buffers.
/// Convenience function to write a packet consisting of multiple buffers.
/// @param buffers all buffers for the packet. The first buffer is a header
/// @param buffers all buffers for the packet. The first buffer is a header
...
@@ -46,14 +42,14 @@ public:
...
@@ -46,14 +42,14 @@ public:
/// @warning this function takes ownership of `buffers`.
/// @warning this function takes ownership of `buffers`.
template
<
class
...
Ts
>
template
<
class
...
Ts
>
void
write_packet
(
Ts
&
...
buffers
)
{
void
write_packet
(
Ts
&
...
buffers
)
{
b
uffer_type
*
bufs
[]
=
{
&
buffers
...};
b
yte_buffer
*
bufs
[]
=
{
&
buffers
...};
write_impl
(
make_span
(
bufs
,
sizeof
...(
Ts
)));
write_impl
(
make_span
(
bufs
,
sizeof
...(
Ts
)));
}
}
protected:
protected:
/// Implementing function for `write_packet`.
/// Implementing function for `write_packet`.
/// @param buffers a `span` containing all buffers of a packet.
/// @param buffers a `span` containing all buffers of a packet.
virtual
void
write_impl
(
span
<
b
uffer_type
*>
buffers
)
=
0
;
virtual
void
write_impl
(
span
<
b
yte_buffer
*>
buffers
)
=
0
;
};
};
}
// namespace caf::net
}
// namespace caf::net
libcaf_net/caf/net/packet_writer_decorator.hpp
View file @
9e2cab29
...
@@ -20,7 +20,8 @@
...
@@ -20,7 +20,8 @@
#include "caf/net/packet_writer.hpp"
#include "caf/net/packet_writer.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/span.hpp"
namespace
caf
::
net
{
namespace
caf
::
net
{
...
@@ -56,11 +57,11 @@ public:
...
@@ -56,11 +57,11 @@ public:
return
parent_
.
manager
();
return
parent_
.
manager
();
}
}
b
uffer_type
next_header_buffer
()
override
{
b
yte_buffer
next_header_buffer
()
override
{
return
transport
().
next_header_buffer
();
return
transport
().
next_header_buffer
();
}
}
b
uffer_type
next_payload_buffer
()
override
{
b
yte_buffer
next_payload_buffer
()
override
{
return
transport
().
next_payload_buffer
();
return
transport
().
next_payload_buffer
();
}
}
...
@@ -76,7 +77,7 @@ public:
...
@@ -76,7 +77,7 @@ public:
}
}
protected:
protected:
void
write_impl
(
span
<
b
uffer_type
*>
buffers
)
override
{
void
write_impl
(
span
<
b
yte_buffer
*>
buffers
)
override
{
parent_
.
write_packet
(
object_
.
id
(),
buffers
);
parent_
.
write_packet
(
object_
.
id
(),
buffers
);
}
}
...
...
libcaf_net/caf/net/stream_transport.hpp
View file @
9e2cab29
...
@@ -19,8 +19,8 @@
...
@@ -19,8 +19,8 @@
#pragma once
#pragma once
#include <deque>
#include <deque>
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/fwd.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager.hpp"
...
@@ -35,9 +35,9 @@
...
@@ -35,9 +35,9 @@
namespace
caf
::
net
{
namespace
caf
::
net
{
template
<
class
Application
>
template
<
class
Application
>
using
stream_transport_base
=
transport_base
<
using
stream_transport_base
stream_transport
<
Application
>
,
transport_worker
<
Application
>
,
stream_socket
,
=
transport_base
<
stream_transport
<
Application
>
,
transport_worker
<
Application
>
,
Application
,
unit_t
>
;
stream_socket
,
Application
,
unit_t
>
;
/// Implements a stream_transport that manages a stream socket.
/// Implements a stream_transport that manages a stream socket.
template
<
class
Application
>
template
<
class
Application
>
...
@@ -53,9 +53,7 @@ public:
...
@@ -53,9 +53,7 @@ public:
using
id_type
=
typename
super
::
id_type
;
using
id_type
=
typename
super
::
id_type
;
using
buffer_type
=
typename
super
::
buffer_type
;
using
write_queue_type
=
std
::
deque
<
std
::
pair
<
bool
,
byte_buffer
>>
;
using
write_queue_type
=
std
::
deque
<
std
::
pair
<
bool
,
buffer_type
>>
;
// -- constructors, destructors, and assignment operators --------------------
// -- constructors, destructors, and assignment operators --------------------
...
@@ -85,9 +83,8 @@ public:
...
@@ -85,9 +83,8 @@ public:
<<
CAF_ARG
(
this
->
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
<<
CAF_ARG
(
this
->
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
this
->
collected_
+=
*
num_bytes
;
this
->
collected_
+=
*
num_bytes
;
if
(
this
->
collected_
>=
this
->
read_threshold_
)
{
if
(
this
->
collected_
>=
this
->
read_threshold_
)
{
if
(
auto
err
=
this
->
next_layer_
.
handle_data
(
*
this
,
if
(
auto
err
=
this
->
next_layer_
.
handle_data
(
make_span
(
*
this
,
make_span
(
this
->
read_buf_
)))
{
this
->
read_buf_
)))
{
CAF_LOG_ERROR
(
"handle_data failed: "
<<
CAF_ARG
(
err
));
CAF_LOG_ERROR
(
"handle_data failed: "
<<
CAF_ARG
(
err
));
return
false
;
return
false
;
}
}
...
@@ -121,7 +118,8 @@ public:
...
@@ -121,7 +118,8 @@ public:
if
(
is_header
)
{
if
(
is_header
)
{
if
(
this
->
header_bufs_
.
size
()
<
this
->
header_bufs_
.
capacity
())
if
(
this
->
header_bufs_
.
size
()
<
this
->
header_bufs_
.
capacity
())
this
->
header_bufs_
.
emplace_back
(
std
::
move
(
buf
));
this
->
header_bufs_
.
emplace_back
(
std
::
move
(
buf
));
}
else
if
(
this
->
payload_bufs_
.
size
()
<
this
->
payload_bufs_
.
capacity
())
{
}
else
if
(
this
->
payload_bufs_
.
size
()
<
this
->
payload_bufs_
.
capacity
())
{
this
->
payload_bufs_
.
emplace_back
(
std
::
move
(
buf
));
this
->
payload_bufs_
.
emplace_back
(
std
::
move
(
buf
));
}
}
write_queue_
.
pop_front
();
write_queue_
.
pop_front
();
...
@@ -166,7 +164,7 @@ public:
...
@@ -166,7 +164,7 @@ public:
return
false
;
return
false
;
}
}
void
write_packet
(
id_type
,
span
<
b
uffer_type
*>
buffers
)
override
{
void
write_packet
(
id_type
,
span
<
b
yte_buffer
*>
buffers
)
override
{
CAF_LOG_TRACE
(
""
);
CAF_LOG_TRACE
(
""
);
CAF_ASSERT
(
!
buffers
.
empty
());
CAF_ASSERT
(
!
buffers
.
empty
());
if
(
this
->
write_queue_
.
empty
())
if
(
this
->
write_queue_
.
empty
())
...
...
libcaf_net/caf/net/transport_base.hpp
View file @
9e2cab29
...
@@ -18,6 +18,7 @@
...
@@ -18,6 +18,7 @@
#pragma once
#pragma once
#include "caf/byte_buffer.hpp"
#include "caf/defaults.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/overload.hpp"
#include "caf/detail/overload.hpp"
#include "caf/fwd.hpp"
#include "caf/fwd.hpp"
...
@@ -51,9 +52,7 @@ public:
...
@@ -51,9 +52,7 @@ public:
using
id_type
=
IdType
;
using
id_type
=
IdType
;
using
buffer_type
=
std
::
vector
<
byte
>
;
using
buffer_cache_type
=
std
::
vector
<
byte_buffer
>
;
using
buffer_cache_type
=
std
::
vector
<
buffer_type
>
;
// -- constructors, destructors, and assignment operators --------------------
// -- constructors, destructors, and assignment operators --------------------
...
@@ -191,26 +190,26 @@ public:
...
@@ -191,26 +190,26 @@ public:
/// transport.
/// transport.
/// @param id The id of the destination endpoint.
/// @param id The id of the destination endpoint.
/// @param buffers Pointers to the buffers that make up the packet content.
/// @param buffers Pointers to the buffers that make up the packet content.
virtual
void
write_packet
(
id_type
id
,
span
<
b
uffer_type
*>
buffers
)
=
0
;
virtual
void
write_packet
(
id_type
id
,
span
<
b
yte_buffer
*>
buffers
)
=
0
;
// -- buffer management ------------------------------------------------------
// -- buffer management ------------------------------------------------------
/// Returns the next cached header buffer or creates a new one if no buffers
/// Returns the next cached header buffer or creates a new one if no buffers
/// are cached.
/// are cached.
b
uffer_type
next_header_buffer
()
{
b
yte_buffer
next_header_buffer
()
{
return
next_buffer_impl
(
header_bufs_
);
return
next_buffer_impl
(
header_bufs_
);
}
}
/// Returns the next cached payload buffer or creates a new one if no buffers
/// Returns the next cached payload buffer or creates a new one if no buffers
/// are cached.
/// are cached.
b
uffer_type
next_payload_buffer
()
{
b
yte_buffer
next_payload_buffer
()
{
return
next_buffer_impl
(
payload_bufs_
);
return
next_buffer_impl
(
payload_bufs_
);
}
}
private:
private:
// -- utility functions ------------------------------------------------------
// -- utility functions ------------------------------------------------------
static
b
uffer_type
next_buffer_impl
(
buffer_cache_type
cache
)
{
static
b
yte_buffer
next_buffer_impl
(
buffer_cache_type
cache
)
{
if
(
cache
.
empty
())
if
(
cache
.
empty
())
return
{};
return
{};
auto
buf
=
std
::
move
(
cache
.
back
());
auto
buf
=
std
::
move
(
cache
.
back
());
...
@@ -225,7 +224,7 @@ protected:
...
@@ -225,7 +224,7 @@ protected:
buffer_cache_type
header_bufs_
;
buffer_cache_type
header_bufs_
;
buffer_cache_type
payload_bufs_
;
buffer_cache_type
payload_bufs_
;
b
uffer_type
read_buf_
;
b
yte_buffer
read_buf_
;
endpoint_manager
*
manager_
;
endpoint_manager
*
manager_
;
...
...
libcaf_net/caf/net/udp_datagram_socket.hpp
View file @
9e2cab29
...
@@ -38,8 +38,9 @@ struct CAF_NET_EXPORT udp_datagram_socket : network_socket {
...
@@ -38,8 +38,9 @@ struct CAF_NET_EXPORT udp_datagram_socket : network_socket {
/// specific port that was bound.
/// specific port that was bound.
/// @returns The connected socket or an error.
/// @returns The connected socket or an error.
/// @relates udp_datagram_socket
/// @relates udp_datagram_socket
expected
<
std
::
pair
<
udp_datagram_socket
,
uint16_t
>>
CAF_NET_EXPORT
expected
<
std
::
pair
<
udp_datagram_socket
,
uint16_t
>>
make_udp_datagram_socket
(
ip_endpoint
ep
,
bool
reuse_addr
=
false
);
CAF_NET_EXPORT
make_udp_datagram_socket
(
ip_endpoint
ep
,
bool
reuse_addr
=
false
);
/// Enables or disables `SIO_UDP_CONNRESET` error on `x`.
/// Enables or disables `SIO_UDP_CONNRESET` error on `x`.
/// @relates udp_datagram_socket
/// @relates udp_datagram_socket
...
@@ -66,7 +67,7 @@ variant<std::pair<size_t, ip_endpoint>, sec>
...
@@ -66,7 +67,7 @@ variant<std::pair<size_t, ip_endpoint>, sec>
/// @relates udp_datagram_socket
/// @relates udp_datagram_socket
/// @pre `bufs.size() < 10`
/// @pre `bufs.size() < 10`
variant
<
size_t
,
sec
>
CAF_NET_EXPORT
write
(
udp_datagram_socket
x
,
variant
<
size_t
,
sec
>
CAF_NET_EXPORT
write
(
udp_datagram_socket
x
,
span
<
std
::
vector
<
byte
>
*>
bufs
,
span
<
byte_buffer
*>
bufs
,
ip_endpoint
ep
);
ip_endpoint
ep
);
/// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`.
/// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`.
...
...
libcaf_net/src/application.cpp
View file @
9e2cab29
...
@@ -24,7 +24,7 @@
...
@@ -24,7 +24,7 @@
#include "caf/actor_system_config.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/byte
_buffer
.hpp"
#include "caf/defaults.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/detail/parse.hpp"
#include "caf/detail/parse.hpp"
...
@@ -379,7 +379,7 @@ error application::handle_down_message(packet_writer&, header received_hdr,
...
@@ -379,7 +379,7 @@ error application::handle_down_message(packet_writer&, header received_hdr,
return
none
;
return
none
;
}
}
error
application
::
generate_handshake
(
std
::
vector
<
byte
>
&
buf
)
{
error
application
::
generate_handshake
(
byte_buffer
&
buf
)
{
binary_serializer
sink
{
&
executor_
,
buf
};
binary_serializer
sink
{
&
executor_
,
buf
};
return
sink
(
system
().
node
(),
return
sink
(
system
().
node
(),
get_or
(
system
().
config
(),
"middleman.app-identifiers"
,
get_or
(
system
().
config
(),
"middleman.app-identifiers"
,
...
...
libcaf_net/src/endpoint_manager.cpp
View file @
9e2cab29
...
@@ -18,7 +18,6 @@
...
@@ -18,7 +18,6 @@
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/byte.hpp"
#include "caf/intrusive/inbox_result.hpp"
#include "caf/intrusive/inbox_result.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp"
#include "caf/sec.hpp"
...
...
libcaf_net/src/header.cpp
View file @
9e2cab29
...
@@ -21,6 +21,7 @@
...
@@ -21,6 +21,7 @@
#include <cstring>
#include <cstring>
#include "caf/byte.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/span.hpp"
#include "caf/span.hpp"
...
@@ -64,7 +65,7 @@ std::array<byte, header_size> to_bytes(header x) {
...
@@ -64,7 +65,7 @@ std::array<byte, header_size> to_bytes(header x) {
return
result
;
return
result
;
}
}
void
to_bytes
(
header
x
,
std
::
vector
<
byte
>
&
buf
)
{
void
to_bytes
(
header
x
,
byte_buffer
&
buf
)
{
buf
.
resize
(
header_size
);
buf
.
resize
(
header_size
);
to_bytes_impl
(
x
,
buf
.
data
());
to_bytes_impl
(
x
,
buf
.
data
());
}
}
...
...
libcaf_net/src/ip.cpp
View file @
9e2cab29
...
@@ -50,10 +50,6 @@
...
@@ -50,10 +50,6 @@
# define HOST_NAME_MAX 255
# define HOST_NAME_MAX 255
#endif
#endif
using
std
::
pair
;
using
std
::
string
;
using
std
::
vector
;
namespace
caf
::
net
::
ip
{
namespace
caf
::
net
::
ip
{
namespace
{
namespace
{
...
@@ -91,9 +87,8 @@ void for_each_adapter(F f, bool is_link_local = false) {
...
@@ -91,9 +87,8 @@ void for_each_adapter(F f, bool is_link_local = false) {
CAF_LOG_ERROR
(
"failed to get adapter addresses buffer length"
);
CAF_LOG_ERROR
(
"failed to get adapter addresses buffer length"
);
return
;
return
;
}
}
auto
adapters
=
adapters_ptr
{
reinterpret_cast
<
IP_ADAPTER_ADDRESSES
*>
(
auto
adapters
=
adapters_ptr
{
::
malloc
(
len
)),
reinterpret_cast
<
IP_ADAPTER_ADDRESSES
*>
(
::
malloc
(
len
)),
free
};
free
};
if
(
!
adapters
)
{
if
(
!
adapters
)
{
CAF_LOG_ERROR
(
"malloc failed"
);
CAF_LOG_ERROR
(
"malloc failed"
);
return
;
return
;
...
...
libcaf_net/src/udp_datagram_socket.cpp
View file @
9e2cab29
...
@@ -19,6 +19,7 @@
...
@@ -19,6 +19,7 @@
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/byte.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/convert_ip_endpoint.hpp"
#include "caf/detail/convert_ip_endpoint.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
...
@@ -126,11 +127,11 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
...
@@ -126,11 +127,11 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
#ifdef CAF_WINDOWS
#ifdef CAF_WINDOWS
variant
<
size_t
,
sec
>
write
(
udp_datagram_socket
x
,
span
<
std
::
vector
<
byte
>
*>
bufs
,
variant
<
size_t
,
sec
>
write
(
udp_datagram_socket
x
,
span
<
byte_buffer
*>
bufs
,
ip_endpoint
ep
)
{
ip_endpoint
ep
)
{
CAF_ASSERT
(
bufs
.
size
()
<
10
);
CAF_ASSERT
(
bufs
.
size
()
<
10
);
WSABUF
buf_array
[
10
];
WSABUF
buf_array
[
10
];
auto
convert
=
[](
std
::
vector
<
byte
>
*
buf
)
{
auto
convert
=
[](
byte_buffer
*
buf
)
{
return
WSABUF
{
static_cast
<
ULONG
>
(
buf
->
size
()),
return
WSABUF
{
static_cast
<
ULONG
>
(
buf
->
size
()),
reinterpret_cast
<
CHAR
*>
(
buf
->
data
())};
reinterpret_cast
<
CHAR
*>
(
buf
->
data
())};
};
};
...
@@ -155,10 +156,10 @@ variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
...
@@ -155,10 +156,10 @@ variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
#else // CAF_WINDOWS
#else // CAF_WINDOWS
variant
<
size_t
,
sec
>
write
(
udp_datagram_socket
x
,
span
<
std
::
vector
<
byte
>
*>
bufs
,
variant
<
size_t
,
sec
>
write
(
udp_datagram_socket
x
,
span
<
byte_buffer
*>
bufs
,
ip_endpoint
ep
)
{
ip_endpoint
ep
)
{
CAF_ASSERT
(
bufs
.
size
()
<
10
);
CAF_ASSERT
(
bufs
.
size
()
<
10
);
auto
convert
=
[](
std
::
vector
<
byte
>
*
buf
)
{
auto
convert
=
[](
byte_buffer
*
buf
)
{
return
iovec
{
buf
->
data
(),
buf
->
size
()};
return
iovec
{
buf
->
data
(),
buf
->
size
()};
};
};
sockaddr_storage
addr
=
{};
sockaddr_storage
addr
=
{};
...
...
libcaf_net/test/application.cpp
View file @
9e2cab29
...
@@ -24,7 +24,7 @@
...
@@ -24,7 +24,7 @@
#include <vector>
#include <vector>
#include "caf/byte.hpp"
#include "caf/byte
_buffer
.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/constants.hpp"
...
@@ -46,8 +46,6 @@ struct fixture : test_coordinator_fixture<>,
...
@@ -46,8 +46,6 @@ struct fixture : test_coordinator_fixture<>,
proxy_registry
::
backend
,
proxy_registry
::
backend
,
basp
::
application
::
test_tag
,
basp
::
application
::
test_tag
,
public
packet_writer
{
public
packet_writer
{
using
buffer_type
=
std
::
vector
<
byte
>
;
fixture
()
:
proxies
(
sys
,
*
this
),
app
(
proxies
)
{
fixture
()
:
proxies
(
sys
,
*
this
),
app
(
proxies
)
{
REQUIRE_OK
(
app
.
init
(
*
this
));
REQUIRE_OK
(
app
.
init
(
*
this
));
uri
mars_uri
;
uri
mars_uri
;
...
@@ -56,8 +54,8 @@ struct fixture : test_coordinator_fixture<>,
...
@@ -56,8 +54,8 @@ struct fixture : test_coordinator_fixture<>,
}
}
template
<
class
...
Ts
>
template
<
class
...
Ts
>
b
uffer_type
to_buf
(
const
Ts
&
...
xs
)
{
b
yte_buffer
to_buf
(
const
Ts
&
...
xs
)
{
b
uffer_type
buf
;
b
yte_buffer
buf
;
binary_serializer
sink
{
system
(),
buf
};
binary_serializer
sink
{
system
(),
buf
};
REQUIRE_OK
(
sink
(
xs
...));
REQUIRE_OK
(
sink
(
xs
...));
return
buf
;
return
buf
;
...
@@ -111,11 +109,11 @@ struct fixture : test_coordinator_fixture<>,
...
@@ -111,11 +109,11 @@ struct fixture : test_coordinator_fixture<>,
CAF_FAIL
(
"unexpected function call"
);
CAF_FAIL
(
"unexpected function call"
);
}
}
b
uffer_type
next_payload_buffer
()
override
{
b
yte_buffer
next_payload_buffer
()
override
{
return
{};
return
{};
}
}
b
uffer_type
next_header_buffer
()
override
{
b
yte_buffer
next_header_buffer
()
override
{
return
{};
return
{};
}
}
...
@@ -136,14 +134,14 @@ struct fixture : test_coordinator_fixture<>,
...
@@ -136,14 +134,14 @@ struct fixture : test_coordinator_fixture<>,
}
}
protected:
protected:
void
write_impl
(
span
<
b
uffer_type
*>
buffers
)
override
{
void
write_impl
(
span
<
b
yte_buffer
*>
buffers
)
override
{
for
(
auto
buf
:
buffers
)
for
(
auto
buf
:
buffers
)
output
.
insert
(
output
.
end
(),
buf
->
begin
(),
buf
->
end
());
output
.
insert
(
output
.
end
(),
buf
->
begin
(),
buf
->
end
());
}
}
b
uffer_type
input
;
b
yte_buffer
input
;
b
uffer_type
output
;
b
yte_buffer
output
;
node_id
mars
;
node_id
mars
;
...
...
libcaf_net/test/datagram_transport.cpp
View file @
9e2cab29
...
@@ -25,6 +25,7 @@
...
@@ -25,6 +25,7 @@
#include "caf/binary_serializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/make_actor.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/actor_proxy_impl.hpp"
...
@@ -42,16 +43,14 @@ using namespace caf::net::ip;
...
@@ -42,16 +43,14 @@ using namespace caf::net::ip;
namespace
{
namespace
{
using
byte_buffer_ptr
=
std
::
shared_ptr
<
byte_buffer
>
;
constexpr
string_view
hello_manager
=
"hello manager!"
;
constexpr
string_view
hello_manager
=
"hello manager!"
;
class
dummy_application_factory
;
class
dummy_application_factory
;
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
using
buffer_type
=
std
::
vector
<
byte
>
;
fixture
()
:
shared_buf
(
std
::
make_shared
<
byte_buffer
>
(
1024
))
{
using
buffer_ptr
=
std
::
shared_ptr
<
buffer_type
>
;
fixture
()
:
shared_buf
(
std
::
make_shared
<
buffer_type
>
(
1024
))
{
mpx
=
std
::
make_shared
<
multiplexer
>
();
mpx
=
std
::
make_shared
<
multiplexer
>
();
if
(
auto
err
=
mpx
->
init
())
if
(
auto
err
=
mpx
->
init
())
CAF_FAIL
(
"mpx->init failed: "
<<
err
);
CAF_FAIL
(
"mpx->init failed: "
<<
err
);
...
@@ -79,7 +78,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
...
@@ -79,7 +78,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
return
mpx
->
poll_once
(
false
);
return
mpx
->
poll_once
(
false
);
}
}
error
read_from_socket
(
udp_datagram_socket
sock
,
b
uffer_type
&
buf
)
{
error
read_from_socket
(
udp_datagram_socket
sock
,
b
yte_buffer
&
buf
)
{
uint8_t
receive_attempts
=
0
;
uint8_t
receive_attempts
=
0
;
variant
<
std
::
pair
<
size_t
,
ip_endpoint
>
,
sec
>
read_ret
;
variant
<
std
::
pair
<
size_t
,
ip_endpoint
>
,
sec
>
read_ret
;
do
{
do
{
...
@@ -97,19 +96,15 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
...
@@ -97,19 +96,15 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
}
}
multiplexer_ptr
mpx
;
multiplexer_ptr
mpx
;
buffer_ptr
shared_buf
;
b
yte_b
uffer_ptr
shared_buf
;
ip_endpoint
ep
;
ip_endpoint
ep
;
udp_datagram_socket
send_socket
;
udp_datagram_socket
send_socket
;
udp_datagram_socket
recv_socket
;
udp_datagram_socket
recv_socket
;
};
};
class
dummy_application
{
class
dummy_application
{
using
buffer_type
=
std
::
vector
<
byte
>
;
using
buffer_ptr
=
std
::
shared_ptr
<
buffer_type
>
;
public:
public:
explicit
dummy_application
(
buffer_ptr
rec_buf
)
explicit
dummy_application
(
b
yte_b
uffer_ptr
rec_buf
)
:
rec_buf_
(
std
::
move
(
rec_buf
)){
:
rec_buf_
(
std
::
move
(
rec_buf
)){
// nop
// nop
};
};
...
@@ -169,16 +164,15 @@ public:
...
@@ -169,16 +164,15 @@ public:
}
}
private:
private:
buffer_ptr
rec_buf_
;
b
yte_b
uffer_ptr
rec_buf_
;
};
};
class
dummy_application_factory
{
class
dummy_application_factory
{
using
buffer_ptr
=
std
::
shared_ptr
<
std
::
vector
<
byte
>>
;
public:
public:
using
application_type
=
dummy_application
;
using
application_type
=
dummy_application
;
explicit
dummy_application_factory
(
buffer_ptr
buf
)
:
buf_
(
std
::
move
(
buf
))
{
explicit
dummy_application_factory
(
byte_buffer_ptr
buf
)
:
buf_
(
std
::
move
(
buf
))
{
// nop
// nop
}
}
...
@@ -187,7 +181,7 @@ public:
...
@@ -187,7 +181,7 @@ public:
}
}
private:
private:
buffer_ptr
buf_
;
b
yte_b
uffer_ptr
buf_
;
};
};
}
// namespace
}
// namespace
...
@@ -218,7 +212,7 @@ CAF_TEST(receive) {
...
@@ -218,7 +212,7 @@ CAF_TEST(receive) {
CAF_TEST
(
resolve
and
proxy
communication
)
{
CAF_TEST
(
resolve
and
proxy
communication
)
{
using
transport_type
=
datagram_transport
<
dummy_application_factory
>
;
using
transport_type
=
datagram_transport
<
dummy_application_factory
>
;
b
uffer_type
recv_buf
(
1024
);
b
yte_buffer
recv_buf
(
1024
);
auto
uri
=
unbox
(
make_uri
(
"test:/id/42"
));
auto
uri
=
unbox
(
make_uri
(
"test:/id/42"
));
auto
mgr
=
make_endpoint_manager
(
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
mpx
,
sys
,
...
...
libcaf_net/test/endpoint_manager.cpp
View file @
9e2cab29
...
@@ -25,7 +25,7 @@
...
@@ -25,7 +25,7 @@
#include "caf/binary_deserializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/byte
_buffer
.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/actor_proxy_impl.hpp"
...
@@ -40,6 +40,8 @@ using namespace caf::net;
...
@@ -40,6 +40,8 @@ using namespace caf::net;
namespace
{
namespace
{
using
byte_buffer_ptr
=
std
::
shared_ptr
<
byte_buffer
>
;
string_view
hello_manager
{
"hello manager!"
};
string_view
hello_manager
{
"hello manager!"
};
string_view
hello_test
{
"hello test!"
};
string_view
hello_test
{
"hello test!"
};
...
@@ -69,7 +71,7 @@ class dummy_transport {
...
@@ -69,7 +71,7 @@ class dummy_transport {
public:
public:
using
application_type
=
dummy_application
;
using
application_type
=
dummy_application
;
dummy_transport
(
stream_socket
handle
,
std
::
shared_ptr
<
std
::
vector
<
byte
>>
data
)
dummy_transport
(
stream_socket
handle
,
byte_buffer_ptr
data
)
:
handle_
(
handle
),
data_
(
data
),
read_buf_
(
1024
)
{
:
handle_
(
handle
),
data_
(
data
),
read_buf_
(
1024
)
{
// nop
// nop
}
}
...
@@ -146,11 +148,11 @@ public:
...
@@ -146,11 +148,11 @@ public:
private:
private:
stream_socket
handle_
;
stream_socket
handle_
;
std
::
shared_ptr
<
std
::
vector
<
byte
>>
data_
;
byte_buffer_ptr
data_
;
std
::
vector
<
byte
>
read_buf_
;
byte_buffer
read_buf_
;
std
::
vector
<
byte
>
buf_
;
byte_buffer
buf_
;
};
};
}
// namespace
}
// namespace
...
@@ -158,8 +160,8 @@ private:
...
@@ -158,8 +160,8 @@ private:
CAF_TEST_FIXTURE_SCOPE
(
endpoint_manager_tests
,
fixture
)
CAF_TEST_FIXTURE_SCOPE
(
endpoint_manager_tests
,
fixture
)
CAF_TEST
(
send
and
receive
)
{
CAF_TEST
(
send
and
receive
)
{
std
::
vector
<
byte
>
read_buf
(
1024
);
byte_buffer
read_buf
(
1024
);
auto
buf
=
std
::
make_shared
<
std
::
vector
<
byte
>
>
();
auto
buf
=
std
::
make_shared
<
byte_buffer
>
();
auto
sockets
=
unbox
(
make_stream_socket_pair
());
auto
sockets
=
unbox
(
make_stream_socket_pair
());
nonblocking
(
sockets
.
second
,
true
);
nonblocking
(
sockets
.
second
,
true
);
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
),
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
),
...
@@ -184,8 +186,8 @@ CAF_TEST(send and receive) {
...
@@ -184,8 +186,8 @@ CAF_TEST(send and receive) {
}
}
CAF_TEST
(
resolve
and
proxy
communication
)
{
CAF_TEST
(
resolve
and
proxy
communication
)
{
std
::
vector
<
byte
>
read_buf
(
1024
);
byte_buffer
read_buf
(
1024
);
auto
buf
=
std
::
make_shared
<
std
::
vector
<
byte
>
>
();
auto
buf
=
std
::
make_shared
<
byte_buffer
>
();
auto
sockets
=
unbox
(
make_stream_socket_pair
());
auto
sockets
=
unbox
(
make_stream_socket_pair
());
nonblocking
(
sockets
.
second
,
true
);
nonblocking
(
sockets
.
second
,
true
);
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
close
(
sockets
.
second
);
});
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
close
(
sockets
.
second
);
});
...
...
libcaf_net/test/header.cpp
View file @
9e2cab29
...
@@ -24,13 +24,14 @@
...
@@ -24,13 +24,14 @@
#include "caf/binary_deserializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte_buffer.hpp"
using
namespace
caf
;
using
namespace
caf
;
using
namespace
caf
::
net
;
using
namespace
caf
::
net
;
CAF_TEST
(
serialization
)
{
CAF_TEST
(
serialization
)
{
basp
::
header
x
{
basp
::
message_type
::
handshake
,
42
,
4
};
basp
::
header
x
{
basp
::
message_type
::
handshake
,
42
,
4
};
std
::
vector
<
byte
>
buf
;
byte_buffer
buf
;
{
{
binary_serializer
sink
{
nullptr
,
buf
};
binary_serializer
sink
{
nullptr
,
buf
};
CAF_CHECK_EQUAL
(
sink
(
x
),
none
);
CAF_CHECK_EQUAL
(
sink
(
x
),
none
);
...
...
libcaf_net/test/multiplexer.cpp
View file @
9e2cab29
...
@@ -25,9 +25,9 @@
...
@@ -25,9 +25,9 @@
#include <new>
#include <new>
#include <tuple>
#include <tuple>
#include <vector>
#include "caf/byte.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/span.hpp"
#include "caf/span.hpp"
...
@@ -113,9 +113,9 @@ private:
...
@@ -113,9 +113,9 @@ private:
size_t
rd_buf_pos_
;
size_t
rd_buf_pos_
;
std
::
vector
<
byte
>
wr_buf_
;
byte_buffer
wr_buf_
;
std
::
vector
<
byte
>
rd_buf_
;
byte_buffer
rd_buf_
;
};
};
using
dummy_manager_ptr
=
intrusive_ptr
<
dummy_manager
>
;
using
dummy_manager_ptr
=
intrusive_ptr
<
dummy_manager
>
;
...
...
libcaf_net/test/net/basp/worker.cpp
View file @
9e2cab29
...
@@ -26,6 +26,7 @@
...
@@ -26,6 +26,7 @@
#include "caf/actor_control_block.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/make_actor.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/proxy_registry.hpp"
...
@@ -107,7 +108,7 @@ CAF_TEST(deliver serialized message) {
...
@@ -107,7 +108,7 @@ CAF_TEST(deliver serialized message) {
CAF_REQUIRE_NOT_EQUAL
(
hub
.
peek
(),
nullptr
);
CAF_REQUIRE_NOT_EQUAL
(
hub
.
peek
(),
nullptr
);
auto
w
=
hub
.
pop
();
auto
w
=
hub
.
pop
();
CAF_MESSAGE
(
"create a fake message + BASP header"
);
CAF_MESSAGE
(
"create a fake message + BASP header"
);
std
::
vector
<
byte
>
payload
;
byte_buffer
payload
;
std
::
vector
<
strong_actor_ptr
>
stages
;
std
::
vector
<
strong_actor_ptr
>
stages
;
binary_serializer
sink
{
sys
,
payload
};
binary_serializer
sink
{
sys
,
payload
};
if
(
auto
err
=
sink
(
node_id
{},
self
->
id
(),
testee
.
id
(),
stages
,
if
(
auto
err
=
sink
(
node_id
{},
self
->
id
(),
testee
.
id
(),
stages
,
...
...
libcaf_net/test/pipe_socket.cpp
View file @
9e2cab29
...
@@ -23,9 +23,8 @@
...
@@ -23,9 +23,8 @@
#include "caf/net/test/host_fixture.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "caf/test/dsl.hpp"
#include <vector>
#include "caf/byte.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
using
namespace
caf
;
using
namespace
caf
;
using
namespace
caf
::
net
;
using
namespace
caf
::
net
;
...
@@ -33,9 +32,9 @@ using namespace caf::net;
...
@@ -33,9 +32,9 @@ using namespace caf::net;
CAF_TEST_FIXTURE_SCOPE
(
pipe_socket_tests
,
host_fixture
)
CAF_TEST_FIXTURE_SCOPE
(
pipe_socket_tests
,
host_fixture
)
CAF_TEST
(
send
and
receive
)
{
CAF_TEST
(
send
and
receive
)
{
std
::
vector
<
byte
>
send_buf
{
byte
(
1
),
byte
(
2
),
byte
(
3
),
byte
(
4
),
byte_buffer
send_buf
{
byte
(
1
),
byte
(
2
),
byte
(
3
),
byte
(
4
),
byte
(
5
),
byte
(
6
),
byte
(
7
),
byte
(
8
)};
byte
(
5
),
byte
(
6
),
byte
(
7
),
byte
(
8
)};
std
::
vector
<
byte
>
receive_buf
;
byte_buffer
receive_buf
;
receive_buf
.
resize
(
100
);
receive_buf
.
resize
(
100
);
pipe_socket
rd_sock
;
pipe_socket
rd_sock
;
pipe_socket
wr_sock
;
pipe_socket
wr_sock
;
...
...
libcaf_net/test/stream_application.cpp
View file @
9e2cab29
...
@@ -25,7 +25,7 @@
...
@@ -25,7 +25,7 @@
#include <vector>
#include <vector>
#include "caf/byte.hpp"
#include "caf/byte
_buffer
.hpp"
#include "caf/net/backend/test.hpp"
#include "caf/net/backend/test.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/constants.hpp"
...
@@ -46,8 +46,6 @@ using namespace caf::net;
...
@@ -46,8 +46,6 @@ using namespace caf::net;
namespace
{
namespace
{
using
buffer_type
=
std
::
vector
<
byte
>
;
using
transport_type
=
stream_transport
<
basp
::
application
>
;
using
transport_type
=
stream_transport
<
basp
::
application
>
;
size_t
fetch_size
(
variant
<
size_t
,
sec
>
x
)
{
size_t
fetch_size
(
variant
<
size_t
,
sec
>
x
)
{
...
@@ -80,8 +78,8 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
...
@@ -80,8 +78,8 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
}
}
template
<
class
...
Ts
>
template
<
class
...
Ts
>
b
uffer_type
to_buf
(
const
Ts
&
...
xs
)
{
b
yte_buffer
to_buf
(
const
Ts
&
...
xs
)
{
b
uffer_type
buf
;
b
yte_buffer
buf
;
binary_serializer
sink
{
system
(),
buf
};
binary_serializer
sink
{
system
(),
buf
};
REQUIRE_OK
(
sink
(
xs
...));
REQUIRE_OK
(
sink
(
xs
...));
return
buf
;
return
buf
;
...
@@ -108,7 +106,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
...
@@ -108,7 +106,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
}
}
void
consume_handshake
()
{
void
consume_handshake
()
{
b
uffer_type
buf
(
basp
::
header_size
);
b
yte_buffer
buf
(
basp
::
header_size
);
if
(
fetch_size
(
read
(
sock
,
buf
))
!=
basp
::
header_size
)
if
(
fetch_size
(
read
(
sock
,
buf
))
!=
basp
::
header_size
)
CAF_FAIL
(
"unable to read "
<<
basp
::
header_size
<<
" bytes"
);
CAF_FAIL
(
"unable to read "
<<
basp
::
header_size
<<
" bytes"
);
auto
hdr
=
basp
::
header
::
from_bytes
(
buf
);
auto
hdr
=
basp
::
header
::
from_bytes
(
buf
);
...
@@ -159,7 +157,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
...
@@ -159,7 +157,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
#define RECEIVE(msg_type, op_data, ...) \
#define RECEIVE(msg_type, op_data, ...) \
do { \
do { \
CAF_MESSAGE("receive " << msg_type); \
CAF_MESSAGE("receive " << msg_type); \
b
uffer_type
buf(basp::header_size); \
b
yte_buffer
buf(basp::header_size); \
if (fetch_size(read(sock, buf)) != basp::header_size) \
if (fetch_size(read(sock, buf)) != basp::header_size) \
CAF_FAIL("unable to read " << basp::header_size << " bytes"); \
CAF_FAIL("unable to read " << basp::header_size << " bytes"); \
auto hdr = basp::header::from_bytes(buf); \
auto hdr = basp::header::from_bytes(buf); \
...
...
libcaf_net/test/stream_socket.cpp
View file @
9e2cab29
...
@@ -24,6 +24,7 @@
...
@@ -24,6 +24,7 @@
#include "caf/test/dsl.hpp"
#include "caf/test/dsl.hpp"
#include "caf/byte.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/span.hpp"
#include "caf/span.hpp"
using
namespace
caf
;
using
namespace
caf
;
...
@@ -66,7 +67,7 @@ struct fixture : host_fixture {
...
@@ -66,7 +67,7 @@ struct fixture : host_fixture {
stream_socket
first
;
stream_socket
first
;
stream_socket
second
;
stream_socket
second
;
std
::
vector
<
byte
>
rd_buf
;
byte_buffer
rd_buf
;
};
};
}
// namespace
}
// namespace
...
@@ -79,7 +80,7 @@ CAF_TEST(read on empty sockets) {
...
@@ -79,7 +80,7 @@ CAF_TEST(read on empty sockets) {
}
}
CAF_TEST
(
transfer
data
from
first
to
second
socket
)
{
CAF_TEST
(
transfer
data
from
first
to
second
socket
)
{
std
::
vector
<
byte
>
wr_buf
{
1
_b
,
2
_b
,
4
_b
,
8
_b
,
16
_b
,
32
_b
,
64
_b
};
byte_buffer
wr_buf
{
1
_b
,
2
_b
,
4
_b
,
8
_b
,
16
_b
,
32
_b
,
64
_b
};
CAF_MESSAGE
(
"transfer data from first to second socket"
);
CAF_MESSAGE
(
"transfer data from first to second socket"
);
CAF_CHECK_EQUAL
(
write
(
first
,
wr_buf
),
wr_buf
.
size
());
CAF_CHECK_EQUAL
(
write
(
first
,
wr_buf
),
wr_buf
.
size
());
CAF_CHECK_EQUAL
(
read
(
second
,
rd_buf
),
wr_buf
.
size
());
CAF_CHECK_EQUAL
(
read
(
second
,
rd_buf
),
wr_buf
.
size
());
...
@@ -88,7 +89,7 @@ CAF_TEST(transfer data from first to second socket) {
...
@@ -88,7 +89,7 @@ CAF_TEST(transfer data from first to second socket) {
}
}
CAF_TEST
(
transfer
data
from
second
to
first
socket
)
{
CAF_TEST
(
transfer
data
from
second
to
first
socket
)
{
std
::
vector
<
byte
>
wr_buf
{
1
_b
,
2
_b
,
4
_b
,
8
_b
,
16
_b
,
32
_b
,
64
_b
};
byte_buffer
wr_buf
{
1
_b
,
2
_b
,
4
_b
,
8
_b
,
16
_b
,
32
_b
,
64
_b
};
CAF_CHECK_EQUAL
(
write
(
second
,
wr_buf
),
wr_buf
.
size
());
CAF_CHECK_EQUAL
(
write
(
second
,
wr_buf
),
wr_buf
.
size
());
CAF_CHECK_EQUAL
(
read
(
first
,
rd_buf
),
wr_buf
.
size
());
CAF_CHECK_EQUAL
(
read
(
first
,
rd_buf
),
wr_buf
.
size
());
CAF_CHECK
(
std
::
equal
(
wr_buf
.
begin
(),
wr_buf
.
end
(),
rd_buf
.
begin
()));
CAF_CHECK
(
std
::
equal
(
wr_buf
.
begin
(),
wr_buf
.
end
(),
rd_buf
.
begin
()));
...
@@ -101,9 +102,9 @@ CAF_TEST(shut down first socket and observe shutdown on the second one) {
...
@@ -101,9 +102,9 @@ CAF_TEST(shut down first socket and observe shutdown on the second one) {
}
}
CAF_TEST
(
transfer
data
using
multiple
buffers
)
{
CAF_TEST
(
transfer
data
using
multiple
buffers
)
{
std
::
vector
<
byte
>
wr_buf_1
{
1
_b
,
2
_b
,
4
_b
};
byte_buffer
wr_buf_1
{
1
_b
,
2
_b
,
4
_b
};
std
::
vector
<
byte
>
wr_buf_2
{
8
_b
,
16
_b
,
32
_b
,
64
_b
};
byte_buffer
wr_buf_2
{
8
_b
,
16
_b
,
32
_b
,
64
_b
};
std
::
vector
<
byte
>
full_buf
;
byte_buffer
full_buf
;
full_buf
.
insert
(
full_buf
.
end
(),
wr_buf_1
.
begin
(),
wr_buf_1
.
end
());
full_buf
.
insert
(
full_buf
.
end
(),
wr_buf_1
.
begin
(),
wr_buf_1
.
end
());
full_buf
.
insert
(
full_buf
.
end
(),
wr_buf_2
.
begin
(),
wr_buf_2
.
end
());
full_buf
.
insert
(
full_buf
.
end
(),
wr_buf_2
.
begin
(),
wr_buf_2
.
end
());
CAF_CHECK_EQUAL
(
write
(
second
,
{
make_span
(
wr_buf_1
),
make_span
(
wr_buf_2
)}),
CAF_CHECK_EQUAL
(
write
(
second
,
{
make_span
(
wr_buf_1
),
make_span
(
wr_buf_2
)}),
...
...
libcaf_net/test/stream_transport.cpp
View file @
9e2cab29
...
@@ -26,6 +26,7 @@
...
@@ -26,6 +26,7 @@
#include "caf/binary_deserializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/actor_proxy_impl.hpp"
...
@@ -44,11 +45,9 @@ namespace {
...
@@ -44,11 +45,9 @@ namespace {
constexpr
string_view
hello_manager
=
"hello manager!"
;
constexpr
string_view
hello_manager
=
"hello manager!"
;
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
using
b
uffer_type
=
std
::
vector
<
byte
>
;
using
b
yte_buffer_ptr
=
std
::
shared_ptr
<
byte_buffer
>
;
using
buffer_ptr
=
std
::
shared_ptr
<
buffer_type
>
;
fixture
()
:
recv_buf
(
1024
),
shared_buf
{
std
::
make_shared
<
byte_buffer
>
()}
{
fixture
()
:
recv_buf
(
1024
),
shared_buf
{
std
::
make_shared
<
buffer_type
>
()}
{
mpx
=
std
::
make_shared
<
multiplexer
>
();
mpx
=
std
::
make_shared
<
multiplexer
>
();
if
(
auto
err
=
mpx
->
init
())
if
(
auto
err
=
mpx
->
init
())
CAF_FAIL
(
"mpx->init failed: "
<<
err
);
CAF_FAIL
(
"mpx->init failed: "
<<
err
);
...
@@ -66,19 +65,17 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
...
@@ -66,19 +65,17 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
}
}
multiplexer_ptr
mpx
;
multiplexer_ptr
mpx
;
b
uffer_type
recv_buf
;
b
yte_buffer
recv_buf
;
socket_guard
<
stream_socket
>
send_socket_guard
;
socket_guard
<
stream_socket
>
send_socket_guard
;
socket_guard
<
stream_socket
>
recv_socket_guard
;
socket_guard
<
stream_socket
>
recv_socket_guard
;
buffer_ptr
shared_buf
;
b
yte_b
uffer_ptr
shared_buf
;
};
};
class
dummy_application
{
class
dummy_application
{
using
buffer_type
=
std
::
vector
<
byte
>
;
using
byte_buffer_ptr
=
std
::
shared_ptr
<
byte_buffer
>
;
using
buffer_ptr
=
std
::
shared_ptr
<
buffer_type
>
;
public:
public:
dummy_application
(
buffer_ptr
rec_buf
)
dummy_application
(
b
yte_b
uffer_ptr
rec_buf
)
:
rec_buf_
(
std
::
move
(
rec_buf
)){
:
rec_buf_
(
std
::
move
(
rec_buf
)){
// nop
// nop
};
};
...
@@ -140,7 +137,7 @@ public:
...
@@ -140,7 +137,7 @@ public:
}
}
private:
private:
buffer_ptr
rec_buf_
;
b
yte_b
uffer_ptr
rec_buf_
;
};
};
}
// namespace
}
// namespace
...
...
libcaf_net/test/string_application.cpp
View file @
9e2cab29
...
@@ -23,11 +23,10 @@
...
@@ -23,11 +23,10 @@
#include "caf/net/test/host_fixture.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "caf/test/dsl.hpp"
#include <vector>
#include "caf/binary_deserializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/actor_proxy_impl.hpp"
...
@@ -43,7 +42,7 @@ using namespace caf::policy;
...
@@ -43,7 +42,7 @@ using namespace caf::policy;
namespace
{
namespace
{
using
b
uffer_type
=
std
::
vector
<
byte
>
;
using
b
yte_buffer_ptr
=
std
::
shared_ptr
<
byte_buffer
>
;
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
fixture
()
{
fixture
()
{
...
@@ -75,8 +74,7 @@ class string_application {
...
@@ -75,8 +74,7 @@ class string_application {
public:
public:
using
header_type
=
string_application_header
;
using
header_type
=
string_application_header
;
string_application
(
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf
)
string_application
(
byte_buffer_ptr
buf
)
:
buf_
(
std
::
move
(
buf
))
{
:
buf_
(
std
::
move
(
buf
))
{
// nop
// nop
}
}
...
@@ -117,7 +115,7 @@ public:
...
@@ -117,7 +115,7 @@ public:
}
}
private:
private:
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf_
;
byte_buffer_ptr
buf_
;
};
};
template
<
class
Base
,
class
Subtype
>
template
<
class
Base
,
class
Subtype
>
...
@@ -125,7 +123,7 @@ class stream_string_application : public Base {
...
@@ -125,7 +123,7 @@ class stream_string_application : public Base {
public:
public:
using
header_type
=
typename
Base
::
header_type
;
using
header_type
=
typename
Base
::
header_type
;
stream_string_application
(
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf
)
stream_string_application
(
byte_buffer_ptr
buf
)
:
Base
(
std
::
move
(
buf
)),
await_payload_
(
false
)
{
:
Base
(
std
::
move
(
buf
)),
await_payload_
(
false
)
{
// nop
// nop
}
}
...
@@ -204,9 +202,9 @@ CAF_TEST(receive) {
...
@@ -204,9 +202,9 @@ CAF_TEST(receive) {
using
application_type
using
application_type
=
extend
<
string_application
>::
with
<
stream_string_application
>
;
=
extend
<
string_application
>::
with
<
stream_string_application
>
;
using
transport_type
=
stream_transport
<
application_type
>
;
using
transport_type
=
stream_transport
<
application_type
>
;
std
::
vector
<
byte
>
read_buf
(
1024
);
byte_buffer
read_buf
(
1024
);
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
auto
buf
=
std
::
make_shared
<
std
::
vector
<
byte
>
>
();
auto
buf
=
std
::
make_shared
<
byte_buffer
>
();
auto
sockets
=
unbox
(
make_stream_socket_pair
());
auto
sockets
=
unbox
(
make_stream_socket_pair
());
nonblocking
(
sockets
.
second
,
true
);
nonblocking
(
sockets
.
second
,
true
);
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
),
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
),
...
...
libcaf_net/test/transport_worker.cpp
View file @
9e2cab29
...
@@ -25,6 +25,7 @@
...
@@ -25,6 +25,7 @@
#include "caf/binary_serializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/make_actor.hpp"
#include "caf/make_actor.hpp"
...
@@ -50,7 +51,7 @@ struct application_result {
...
@@ -50,7 +51,7 @@ struct application_result {
};
};
struct
transport_result
{
struct
transport_result
{
std
::
vector
<
byte
>
packet_buffer
;
byte_buffer
packet_buffer
;
ip_endpoint
ep
;
ip_endpoint
ep
;
};
};
...
...
libcaf_net/test/transport_worker_dispatcher.cpp
View file @
9e2cab29
...
@@ -23,6 +23,7 @@
...
@@ -23,6 +23,7 @@
#include "caf/net/test/host_fixture.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "caf/test/dsl.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/make_actor.hpp"
#include "caf/make_actor.hpp"
#include "caf/monitorable_actor.hpp"
#include "caf/monitorable_actor.hpp"
...
@@ -34,7 +35,7 @@ using namespace caf::net;
...
@@ -34,7 +35,7 @@ using namespace caf::net;
namespace
{
namespace
{
using
b
uffer_type
=
std
::
vector
<
byte
>
;
using
b
yte_buffer_ptr
=
std
::
shared_ptr
<
byte_buffer
>
;
constexpr
string_view
hello_test
=
"hello_test"
;
constexpr
string_view
hello_test
=
"hello_test"
;
...
@@ -50,7 +51,7 @@ struct dummy_actor : public monitorable_actor {
...
@@ -50,7 +51,7 @@ struct dummy_actor : public monitorable_actor {
class
dummy_application
{
class
dummy_application
{
public:
public:
dummy_application
(
std
::
shared_ptr
<
buffer_type
>
rec_buf
,
uint8_t
id
)
dummy_application
(
byte_buffer_ptr
rec_buf
,
uint8_t
id
)
:
rec_buf_
(
std
::
move
(
rec_buf
)),
:
rec_buf_
(
std
::
move
(
rec_buf
)),
id_
(
id
){
id_
(
id
){
// nop
// nop
...
@@ -68,7 +69,7 @@ public:
...
@@ -68,7 +69,7 @@ public:
void
write_message
(
Parent
&
parent
,
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
ptr
)
{
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
ptr
)
{
rec_buf_
->
push_back
(
static_cast
<
byte
>
(
id_
));
rec_buf_
->
push_back
(
static_cast
<
byte
>
(
id_
));
auto
data
=
ptr
->
msg
->
content
().
get_as
<
std
::
vector
<
byte
>
>
(
0
);
auto
data
=
ptr
->
msg
->
content
().
get_as
<
byte_buffer
>
(
0
);
parent
.
write_packet
(
data
);
parent
.
write_packet
(
data
);
}
}
...
@@ -93,7 +94,7 @@ public:
...
@@ -93,7 +94,7 @@ public:
}
}
private:
private:
std
::
shared_ptr
<
buffer_type
>
rec_buf_
;
byte_buffer_ptr
rec_buf_
;
uint8_t
id_
;
uint8_t
id_
;
};
};
...
@@ -101,7 +102,7 @@ struct dummy_application_factory {
...
@@ -101,7 +102,7 @@ struct dummy_application_factory {
public:
public:
using
application_type
=
dummy_application
;
using
application_type
=
dummy_application
;
dummy_application_factory
(
std
::
shared_ptr
<
buffer_type
>
buf
)
dummy_application_factory
(
byte_buffer_ptr
buf
)
:
buf_
(
std
::
move
(
buf
)),
application_cnt_
(
0
)
{
:
buf_
(
std
::
move
(
buf
)),
application_cnt_
(
0
)
{
// nop
// nop
}
}
...
@@ -111,7 +112,7 @@ public:
...
@@ -111,7 +112,7 @@ public:
}
}
private:
private:
std
::
shared_ptr
<
buffer_type
>
buf_
;
byte_buffer_ptr
buf_
;
uint8_t
application_cnt_
;
uint8_t
application_cnt_
;
};
};
...
@@ -122,13 +123,13 @@ struct dummy_transport {
...
@@ -122,13 +123,13 @@ struct dummy_transport {
using
application_type
=
dummy_application
;
using
application_type
=
dummy_application
;
dummy_transport
(
actor_system
&
sys
,
std
::
shared_ptr
<
buffer_type
>
buf
)
dummy_transport
(
actor_system
&
sys
,
byte_buffer_ptr
buf
)
:
sys_
(
sys
),
buf_
(
std
::
move
(
buf
))
{
:
sys_
(
sys
),
buf_
(
std
::
move
(
buf
))
{
// nop
// nop
}
}
template
<
class
IdType
>
template
<
class
IdType
>
void
write_packet
(
IdType
,
span
<
b
uffer_type
*>
buffers
)
{
void
write_packet
(
IdType
,
span
<
b
yte_buffer
*>
buffers
)
{
for
(
auto
buf
:
buffers
)
for
(
auto
buf
:
buffers
)
buf_
->
insert
(
buf_
->
end
(),
buf
->
begin
(),
buf
->
end
());
buf_
->
insert
(
buf_
->
end
(),
buf
->
begin
(),
buf
->
end
());
}
}
...
@@ -141,17 +142,17 @@ struct dummy_transport {
...
@@ -141,17 +142,17 @@ struct dummy_transport {
return
*
this
;
return
*
this
;
}
}
b
uffer_type
next_header_buffer
()
{
b
yte_buffer
next_header_buffer
()
{
return
{};
return
{};
}
}
b
uffer_type
next_payload_buffer
()
{
b
yte_buffer
next_payload_buffer
()
{
return
{};
return
{};
}
}
private:
private:
actor_system
&
sys_
;
actor_system
&
sys_
;
std
::
shared_ptr
<
buffer_type
>
buf_
;
byte_buffer_ptr
buf_
;
};
};
struct
testdata
{
struct
testdata
{
...
@@ -188,7 +189,7 @@ struct fixture : host_fixture {
...
@@ -188,7 +189,7 @@ struct fixture : host_fixture {
=
transport_worker_dispatcher
<
dummy_application_factory
,
ip_endpoint
>
;
=
transport_worker_dispatcher
<
dummy_application_factory
,
ip_endpoint
>
;
fixture
()
fixture
()
:
buf
{
std
::
make_shared
<
b
uffer_type
>
()},
:
buf
{
std
::
make_shared
<
b
yte_buffer
>
()},
dispatcher
{
dummy_application_factory
{
buf
}},
dispatcher
{
dummy_application_factory
{
buf
}},
dummy
{
sys
,
buf
}
{
dummy
{
sys
,
buf
}
{
add_new_workers
();
add_new_workers
();
...
@@ -235,7 +236,7 @@ struct fixture : host_fixture {
...
@@ -235,7 +236,7 @@ struct fixture : host_fixture {
actor_system_config
cfg
{};
actor_system_config
cfg
{};
actor_system
sys
{
cfg
};
actor_system
sys
{
cfg
};
std
::
shared_ptr
<
buffer_type
>
buf
;
byte_buffer_ptr
buf
;
dispatcher_type
dispatcher
;
dispatcher_type
dispatcher
;
dummy_transport
dummy
;
dummy_transport
dummy
;
...
...
libcaf_net/test/udp_datagram_socket.cpp
View file @
9e2cab29
...
@@ -24,6 +24,7 @@
...
@@ -24,6 +24,7 @@
#include "caf/test/dsl.hpp"
#include "caf/test/dsl.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/ip_address.hpp"
#include "caf/ip_address.hpp"
...
@@ -62,10 +63,10 @@ struct fixture : host_fixture {
...
@@ -62,10 +63,10 @@ struct fixture : host_fixture {
ip_endpoint
ep
;
ip_endpoint
ep
;
udp_datagram_socket
send_socket
;
udp_datagram_socket
send_socket
;
udp_datagram_socket
receive_socket
;
udp_datagram_socket
receive_socket
;
std
::
vector
<
byte
>
buf
;
byte_buffer
buf
;
};
};
error
read_from_socket
(
udp_datagram_socket
sock
,
std
::
vector
<
byte
>
&
buf
)
{
error
read_from_socket
(
udp_datagram_socket
sock
,
byte_buffer
&
buf
)
{
uint8_t
receive_attempts
=
0
;
uint8_t
receive_attempts
=
0
;
variant
<
std
::
pair
<
size_t
,
ip_endpoint
>
,
sec
>
read_ret
;
variant
<
std
::
pair
<
size_t
,
ip_endpoint
>
,
sec
>
read_ret
;
do
{
do
{
...
@@ -115,17 +116,17 @@ CAF_TEST(read / write using span<byte>) {
...
@@ -115,17 +116,17 @@ CAF_TEST(read / write using span<byte>) {
CAF_CHECK_EQUAL
(
received
,
hello_test
);
CAF_CHECK_EQUAL
(
received
,
hello_test
);
}
}
CAF_TEST
(
read
/
write
using
span
<
std
::
vector
<
byte
>
*>
)
{
CAF_TEST
(
read
/
write
using
span
<
byte_buffer
*>
)
{
// generate header and payload in separate buffers
// generate header and payload in separate buffers
header
hdr
{
hello_test
.
size
()};
header
hdr
{
hello_test
.
size
()};
std
::
vector
<
byte
>
hdr_buf
;
byte_buffer
hdr_buf
;
binary_serializer
sink
(
sys
,
hdr_buf
);
binary_serializer
sink
(
sys
,
hdr_buf
);
if
(
auto
err
=
sink
(
hdr
))
if
(
auto
err
=
sink
(
hdr
))
CAF_FAIL
(
"serializing payload failed"
<<
err
);
CAF_FAIL
(
"serializing payload failed"
<<
err
);
auto
bytes
=
as_bytes
(
make_span
(
hello_test
));
auto
bytes
=
as_bytes
(
make_span
(
hello_test
));
std
::
vector
<
byte
>
payload_buf
(
bytes
.
begin
(),
bytes
.
end
());
byte_buffer
payload_buf
(
bytes
.
begin
(),
bytes
.
end
());
auto
packet_size
=
hdr_buf
.
size
()
+
payload_buf
.
size
();
auto
packet_size
=
hdr_buf
.
size
()
+
payload_buf
.
size
();
std
::
vector
<
std
::
vector
<
byte
>
*>
bufs
{
&
hdr_buf
,
&
payload_buf
};
std
::
vector
<
byte_buffer
*>
bufs
{
&
hdr_buf
,
&
payload_buf
};
CAF_CHECK_EQUAL
(
write
(
send_socket
,
bufs
,
ep
),
packet_size
);
CAF_CHECK_EQUAL
(
write
(
send_socket
,
bufs
,
ep
),
packet_size
);
// receive both as one single packet.
// receive both as one single packet.
buf
.
resize
(
packet_size
);
buf
.
resize
(
packet_size
);
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment