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
72ed1769
Commit
72ed1769
authored
Aug 31, 2019
by
Dominik Charousset
Committed by
GitHub
Aug 31, 2019
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #18
Refactor scribe to stream-transport
parents
5bff4f92
a708f150
Changes
13
Hide whitespace changes
Inline
Side-by-side
Showing
13 changed files
with
523 additions
and
105 deletions
+523
-105
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+0
-1
libcaf_net/caf/net/endpoint_manager.hpp
libcaf_net/caf/net/endpoint_manager.hpp
+7
-0
libcaf_net/caf/net/endpoint_manager_impl.hpp
libcaf_net/caf/net/endpoint_manager_impl.hpp
+18
-10
libcaf_net/caf/net/fwd.hpp
libcaf_net/caf/net/fwd.hpp
+2
-0
libcaf_net/caf/net/make_endpoint_manager.hpp
libcaf_net/caf/net/make_endpoint_manager.hpp
+4
-5
libcaf_net/caf/net/stream_transport.hpp
libcaf_net/caf/net/stream_transport.hpp
+108
-41
libcaf_net/caf/net/transport_worker.hpp
libcaf_net/caf/net/transport_worker.hpp
+98
-0
libcaf_net/caf/net/write_packet_decorator.hpp
libcaf_net/caf/net/write_packet_decorator.hpp
+27
-14
libcaf_net/test/endpoint_manager.cpp
libcaf_net/test/endpoint_manager.cpp
+11
-12
libcaf_net/test/socket_guard.cpp
libcaf_net/test/socket_guard.cpp
+2
-0
libcaf_net/test/stream_transport.cpp
libcaf_net/test/stream_transport.cpp
+14
-11
libcaf_net/test/string_application.cpp
libcaf_net/test/string_application.cpp
+12
-11
libcaf_net/test/transport_worker.cpp
libcaf_net/test/transport_worker.cpp
+220
-0
No files found.
libcaf_net/CMakeLists.txt
View file @
72ed1769
...
...
@@ -19,7 +19,6 @@ set(LIBCAF_NET_SRCS
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/scribe.cpp
src/tcp_stream_socket.cpp
src/tcp_accept_socket.cpp
)
...
...
libcaf_net/caf/net/endpoint_manager.hpp
View file @
72ed1769
...
...
@@ -23,6 +23,7 @@
#include <memory>
#include "caf/actor.hpp"
#include "caf/actor_clock.hpp"
#include "caf/byte.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
...
...
@@ -137,6 +138,9 @@ public:
/// Enqueues a message to the endpoint.
void
enqueue
(
mailbox_element_ptr
msg
,
std
::
vector
<
byte
>
payload
);
/// Enqueues a timeout to the endpoint.
void
enqueue
(
timeout_msg
msg
);
// -- pure virtual member functions ------------------------------------------
/// Initializes the manager before adding it to the multiplexer's event loop.
...
...
@@ -154,6 +158,9 @@ protected:
/// Stores outbound messages.
message_queue_type
messages_
;
/// Stores a proxy for interacting with the actor clock.
actor
timeout_proxy_
;
};
using
endpoint_manager_ptr
=
intrusive_ptr
<
endpoint_manager
>
;
...
...
libcaf_net/caf/net/endpoint_manager_impl.hpp
View file @
72ed1769
...
...
@@ -23,7 +23,7 @@
namespace
caf
{
namespace
net
{
template
<
class
Transport
,
class
Application
>
template
<
class
Transport
>
class
endpoint_manager_impl
:
public
endpoint_manager
{
public:
// -- member types -----------------------------------------------------------
...
...
@@ -32,15 +32,13 @@ public:
using
transport_type
=
Transport
;
using
application_type
=
Application
;
using
application_type
=
typename
transport_type
::
application_type
;
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager_impl
(
const
multiplexer_ptr
&
parent
,
actor_system
&
sys
,
Transport
trans
,
Application
app
)
:
super
(
trans
.
handle
(),
parent
,
sys
),
transport_
(
std
::
move
(
trans
)),
application_
(
std
::
move
(
app
))
{
Transport
trans
)
:
super
(
trans
.
handle
(),
parent
,
sys
),
transport_
(
std
::
move
(
trans
))
{
// nop
}
...
...
@@ -54,8 +52,16 @@ public:
return
transport_
;
}
application_type
&
application
()
{
return
application_
;
// -- timeout management -----------------------------------------------------
template
<
class
...
Ts
>
uint64_t
set_timeout
(
actor_clock
::
time_point
tp
,
atom_value
type
,
Ts
&&
...
xs
)
{
auto
act
=
actor_cast
<
abstract_actor
*>
(
timeout_proxy_
);
CAF_ASSERT
(
act
!=
nullptr
);
sys_
.
clock
().
set_multi_timeout
(
tp
,
act
,
type
,
next_timeout_id_
);
transport_
.
set_timeout
(
next_timeout_id_
,
std
::
forward
<
Ts
>
(
xs
)...);
return
next_timeout_id_
++
;
}
// -- interface functions ----------------------------------------------------
...
...
@@ -91,7 +97,7 @@ public:
}
void
handle_error
(
sec
code
)
override
{
transport_
.
handle_error
(
application_
,
code
);
transport_
.
handle_error
(
code
);
}
serialize_fun_type
serialize_fun
()
const
noexcept
override
{
...
...
@@ -100,7 +106,9 @@ public:
private:
transport_type
transport_
;
application_type
application_
;
/// Stores the id for the next timeout.
uint64_t
next_timeout_id_
;
};
}
// namespace net
...
...
libcaf_net/caf/net/fwd.hpp
View file @
72ed1769
...
...
@@ -27,6 +27,8 @@ namespace net {
class
multiplexer
;
class
socket_manager
;
template
<
class
Application
,
class
IdType
=
unit_t
>
class
transport_worker
;
struct
network_socket
;
struct
pipe_socket
;
...
...
libcaf_net/caf/net/make_endpoint_manager.hpp
View file @
72ed1769
...
...
@@ -25,12 +25,11 @@
namespace
caf
{
namespace
net
{
template
<
class
Transport
,
class
Application
>
template
<
class
Transport
>
endpoint_manager_ptr
make_endpoint_manager
(
const
multiplexer_ptr
&
mpx
,
actor_system
&
sys
,
Transport
trans
,
Application
app
)
{
using
impl
=
endpoint_manager_impl
<
Transport
,
Application
>
;
return
make_counted
<
impl
>
(
mpx
,
sys
,
std
::
move
(
trans
),
std
::
move
(
app
));
actor_system
&
sys
,
Transport
trans
)
{
using
impl
=
endpoint_manager_impl
<
Transport
>
;
return
make_counted
<
impl
>
(
mpx
,
sys
,
std
::
move
(
trans
));
}
}
// namespace net
...
...
libcaf_net/caf/
policy/scribe
.hpp
→
libcaf_net/caf/
net/stream_transport
.hpp
View file @
72ed1769
...
...
@@ -25,26 +25,49 @@
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace
caf
{
namespace
policy
{
namespace
net
{
/// Implements a scribe policy that manages a stream socket.
class
scribe
{
/// Implements a stream_transport that manages a stream socket.
template
<
class
Application
>
class
stream_transport
{
public:
explicit
scribe
(
net
::
stream_socket
handle
);
// -- member types -----------------------------------------------------------
using
application_type
=
Application
;
// -- constructors, destructors, and assignment operators --------------------
stream_transport
(
stream_socket
handle
,
application_type
application
)
:
worker_
(
std
::
move
(
application
)),
handle_
(
handle
),
// max_consecutive_reads_(0),
read_threshold_
(
1024
),
collected_
(
0
),
max_
(
1024
),
rd_flag_
(
net
::
receive_policy_flag
::
exactly
),
written_
(
0
)
{
// nop
}
// -- properties -------------------------------------------------------------
net
::
stream_socket
handle
()
const
noexcept
{
stream_socket
handle
()
const
noexcept
{
return
handle_
;
}
// -- member functions -------------------------------------------------------
template
<
class
Parent
>
error
init
(
Parent
&
parent
)
{
parent
.
application
().
init
(
parent
);
parent
.
mask_add
(
net
::
operation
::
read
);
if
(
auto
err
=
worker_
.
init
(
parent
))
return
err
;
parent
.
mask_add
(
operation
::
read
);
return
none
;
}
...
...
@@ -59,14 +82,15 @@ public:
CAF_LOG_DEBUG
(
CAF_ARG
(
len
)
<<
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
collected_
+=
*
num_bytes
;
if
(
collected_
>=
read_threshold_
)
{
parent
.
application
().
handle_data
(
*
this
,
read_buf_
);
auto
decorator
=
make_write_packet_decorator
(
*
this
,
parent
);
worker_
.
handle_data
(
decorator
,
read_buf_
);
prepare_next_read
();
}
return
true
;
}
else
{
auto
err
=
get
<
sec
>
(
ret
);
CAF_LOG_DEBUG
(
"receive failed"
<<
CAF_ARG
(
err
));
parent
.
application
()
.
handle_error
(
err
);
worker_
.
handle_error
(
err
);
return
false
;
}
}
...
...
@@ -74,25 +98,88 @@ public:
template
<
class
Parent
>
bool
handle_write_event
(
Parent
&
parent
)
{
// Try to write leftover data.
write_some
(
parent
);
write_some
();
// Get new data from parent.
// TODO: dont read all messages at once - get one by one.
for
(
auto
msg
=
parent
.
next_message
();
msg
!=
nullptr
;
msg
=
parent
.
next_message
())
{
parent
.
application
().
write_message
(
*
this
,
std
::
move
(
msg
));
auto
decorator
=
make_write_packet_decorator
(
*
this
,
parent
);
worker_
.
write_message
(
decorator
,
std
::
move
(
msg
));
}
// Write prepared data.
return
write_some
(
parent
);
return
write_some
();
}
template
<
class
Parent
>
void
resolve
(
Parent
&
parent
,
const
std
::
string
&
path
,
actor
listener
)
{
worker_
.
resolve
(
parent
,
path
,
listener
);
}
template
<
class
...
Ts
>
void
set_timeout
(
uint64_t
,
Ts
&&
...)
{
// nop
}
template
<
class
Parent
>
bool
write_some
(
Parent
&
parent
)
{
void
timeout
(
Parent
&
parent
,
atom_value
value
,
uint64_t
id
)
{
worker_
.
timeout
(
parent
,
value
,
id
);
}
void
handle_error
(
sec
code
)
{
worker_
.
handle_error
(
code
);
}
void
prepare_next_read
()
{
read_buf_
.
clear
();
collected_
=
0
;
// This cast does nothing, but prevents a weird compiler error on GCC
// <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
switch
(
static_cast
<
net
::
receive_policy_flag
>
(
rd_flag_
))
{
case
net
:
:
receive_policy_flag
::
exactly
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
read_threshold_
=
max_
;
break
;
case
net
:
:
receive_policy_flag
::
at_most
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
read_threshold_
=
1
;
break
;
case
net
:
:
receive_policy_flag
::
at_least
:
{
// read up to 10% more, but at least allow 100 bytes more
auto
max_size
=
max_
+
std
::
max
<
size_t
>
(
100
,
max_
/
10
);
if
(
read_buf_
.
size
()
!=
max_size
)
read_buf_
.
resize
(
max_size
);
read_threshold_
=
max_
;
break
;
}
}
}
void
configure_read
(
receive_policy
::
config
cfg
)
{
rd_flag_
=
cfg
.
first
;
max_
=
cfg
.
second
;
prepare_next_read
();
}
template
<
class
Parent
>
void
write_packet
(
Parent
&
,
span
<
const
byte
>
header
,
span
<
const
byte
>
payload
,
unit_t
)
{
write_buf_
.
insert
(
write_buf_
.
end
(),
header
.
begin
(),
header
.
end
());
write_buf_
.
insert
(
write_buf_
.
end
(),
payload
.
begin
(),
payload
.
end
());
}
private:
// -- private member functions -----------------------------------------------
bool
write_some
()
{
if
(
write_buf_
.
empty
())
return
false
;
auto
len
=
write_buf_
.
size
()
-
written_
;
auto
buf
=
write_buf_
.
data
()
+
written_
;
CAF_LOG_TRACE
(
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG
(
len
));
auto
ret
=
net
::
write
(
handle_
,
make_span
(
buf
,
len
));
auto
ret
=
write
(
handle_
,
make_span
(
buf
,
len
));
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
ret
))
{
CAF_LOG_DEBUG
(
CAF_ARG
(
len
)
<<
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
// Update state.
...
...
@@ -105,47 +192,27 @@ public:
}
else
{
auto
err
=
get
<
sec
>
(
ret
);
CAF_LOG_DEBUG
(
"send failed"
<<
CAF_ARG
(
err
));
parent
.
application
()
.
handle_error
(
err
);
worker_
.
handle_error
(
err
);
return
false
;
}
return
true
;
}
template
<
class
Parent
>
void
resolve
(
Parent
&
parent
,
const
std
::
string
&
path
,
actor
listener
)
{
parent
.
application
().
resolve
(
parent
,
path
,
listener
);
}
template
<
class
Parent
>
void
timeout
(
Parent
&
parent
,
atom_value
value
,
uint64_t
id
)
{
parent
.
application
().
timeout
(
*
this
,
value
,
id
);
}
template
<
class
Application
>
void
handle_error
(
Application
&
application
,
sec
code
)
{
application
.
handle_error
(
code
);
}
void
prepare_next_read
();
void
configure_read
(
net
::
receive_policy
::
config
cfg
);
void
write_packet
(
span
<
const
byte
>
buf
);
private:
net
::
stream_socket
handle_
;
transport_worker
<
application_type
>
worker_
;
stream_socket
handle_
;
std
::
vector
<
byte
>
read_buf_
;
std
::
vector
<
byte
>
write_buf_
;
size_t
max_consecutive_reads_
;
// TODO use this field!
// TODO implement retries using this member!
// size_t max_consecutive_reads_;
size_t
read_threshold_
;
size_t
collected_
;
size_t
max_
;
net
::
receive_policy_flag
rd_flag_
;
receive_policy_flag
rd_flag_
;
size_t
written_
;
};
}
// namespace
policy
}
// namespace
net
}
// namespace caf
libcaf_net/
src/scribe.c
pp
→
libcaf_net/
caf/net/transport_worker.h
pp
View file @
72ed1769
...
...
@@ -16,64 +16,83 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/policy/scribe.hpp"
#include <system_error>
#pragma once
#include "caf/byte.hpp"
#include "caf/config.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/write_packet_decorator.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
namespace
caf
{
namespace
policy
{
scribe
::
scribe
(
caf
::
net
::
stream_socket
handle
)
:
handle_
(
handle
),
max_consecutive_reads_
(
0
),
read_threshold_
(
1024
),
collected_
(
0
),
max_
(
1024
),
rd_flag_
(
net
::
receive_policy_flag
::
exactly
),
written_
(
0
)
{
// nop
}
void
scribe
::
prepare_next_read
()
{
read_buf_
.
clear
();
collected_
=
0
;
// This cast does nothing, but prevents a weird compiler error on GCC <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
switch
(
static_cast
<
net
::
receive_policy_flag
>
(
rd_flag_
))
{
case
net
:
:
receive_policy_flag
::
exactly
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
read_threshold_
=
max_
;
break
;
case
net
:
:
receive_policy_flag
::
at_most
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
read_threshold_
=
1
;
break
;
case
net
:
:
receive_policy_flag
::
at_least
:
{
// read up to 10% more, but at least allow 100 bytes more
auto
max_size
=
max_
+
std
::
max
<
size_t
>
(
100
,
max_
/
10
);
if
(
read_buf_
.
size
()
!=
max_size
)
read_buf_
.
resize
(
max_size
);
read_threshold_
=
max_
;
break
;
}
namespace
net
{
/// Implements a worker for transport protocols.
template
<
class
Application
,
class
IdType
>
class
transport_worker
{
public:
// -- member types -----------------------------------------------------------
using
id_type
=
IdType
;
using
application_type
=
Application
;
// -- constructors, destructors, and assignment operators --------------------
transport_worker
(
application_type
application
,
id_type
id
=
id_type
{})
:
application_
(
std
::
move
(
application
)),
id_
(
std
::
move
(
id
))
{
// nop
}
// -- member functions -------------------------------------------------------
template
<
class
Parent
>
error
init
(
Parent
&
parent
)
{
return
application_
.
init
(
parent
);
}
template
<
class
Parent
>
void
handle_data
(
Parent
&
parent
,
span
<
const
byte
>
data
)
{
application_
.
handle_data
(
parent
,
data
);
}
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
net
::
endpoint_manager
::
message
>
msg
)
{
auto
decorator
=
make_write_packet_decorator
(
*
this
,
parent
);
application_
.
write_message
(
decorator
,
std
::
move
(
msg
));
}
template
<
class
Parent
>
void
write_packet
(
Parent
&
parent
,
span
<
const
byte
>
header
,
span
<
const
byte
>
payload
)
{
parent
.
write_packet
(
header
,
payload
,
id_
);
}
template
<
class
Parent
>
void
resolve
(
Parent
&
parent
,
const
std
::
string
&
path
,
actor
listener
)
{
application_
.
resolve
(
parent
,
path
,
listener
);
}
template
<
class
Parent
>
void
timeout
(
Parent
&
parent
,
atom_value
value
,
uint64_t
id
)
{
application_
.
timeout
(
parent
,
value
,
id
);
}
void
handle_error
(
sec
error
)
{
application_
.
handle_error
(
error
);
}
}
void
scribe
::
configure_read
(
net
::
receive_policy
::
config
cfg
)
{
rd_flag_
=
cfg
.
first
;
max_
=
cfg
.
second
;
prepare_next_read
();
}
private:
application_type
application_
;
id_type
id_
;
};
void
scribe
::
write_packet
(
span
<
const
byte
>
buf
)
{
write_buf_
.
insert
(
write_buf_
.
end
(),
buf
.
begin
(),
buf
.
end
());
}
template
<
class
Application
,
class
IdType
=
unit_t
>
using
transport_worker_ptr
=
std
::
shared_ptr
<
transport_worker
<
Application
,
IdType
>>
;
}
// namespace
policy
}
// namespace
net
}
// namespace caf
libcaf_net/caf/net/write_packet_decorator.hpp
View file @
72ed1769
...
...
@@ -18,6 +18,9 @@
#pragma once
#include "caf/byte.hpp"
#include "caf/span.hpp"
namespace
caf
{
namespace
net
{
...
...
@@ -26,34 +29,44 @@ namespace net {
template
<
class
Object
,
class
Parent
>
class
write_packet_decorator
{
public:
dispatcher
(
Object
&
object
,
Parent
&
parent
)
// -- member types -----------------------------------------------------------
using
transport_type
=
typename
Parent
::
transport_type
;
using
application_type
=
typename
Parent
::
application_type
;
// -- constructors, destructors, and assignment operators --------------------
write_packet_decorator
(
Object
&
object
,
Parent
&
parent
)
:
object_
(
object
),
parent_
(
parent
)
{
// nop
}
template
<
class
Header
>
void
write_packet
(
const
Header
&
header
,
span
<
const
byte
>
payload
)
{
object_
.
write_packet
(
parent_
,
header
,
payload
);
}
// -- properties -------------------------------------------------------------
actor_system
&
system
()
{
parent_
.
system
();
return
parent_
.
system
();
}
void
cancel_timeout
(
atom_value
type
,
uint64_t
id
)
{
parent_
.
cancel_timeout
(
type
,
id
);
transport_type
&
transport
(
)
{
return
parent_
.
transport
(
);
}
void
set_timeout
(
timestamp
tout
,
atom_value
type
,
uint64_t
id
)
{
parent_
.
set_timeout
(
tout
,
type
,
id
);
// -- member functions -------------------------------------------------------
template
<
class
...
Ts
>
void
write_packet
(
span
<
const
byte
>
header
,
span
<
const
byte
>
payload
,
Ts
&&
...
xs
)
{
object_
.
write_packet
(
parent_
,
header
,
payload
,
std
::
forward
<
Ts
>
(
xs
)...);
}
typename
parent
::
transport_type
&
transport
(
)
{
return
parent_
.
transport_
;
void
cancel_timeout
(
atom_value
type
,
uint64_t
id
)
{
parent_
.
cancel_timeout
(
type
,
id
)
;
}
typename
parent
::
application_type
&
application
()
{
return
parent_
.
application_
;
template
<
class
...
Ts
>
uint64_t
set_timeout
(
timestamp
tout
,
atom_value
type
,
Ts
&&
...
xs
)
{
return
parent_
.
set_timeout
(
tout
,
type
,
std
::
forward
<
Ts
>
(
xs
)...);
}
private:
...
...
libcaf_net/test/endpoint_manager.cpp
View file @
72ed1769
...
...
@@ -72,6 +72,8 @@ public:
class
dummy_transport
{
public:
using
application_type
=
dummy_application
;
dummy_transport
(
stream_socket
handle
,
std
::
shared_ptr
<
std
::
vector
<
byte
>>
data
)
:
handle_
(
handle
),
data_
(
data
),
read_buf_
(
1024
)
{
// nop
...
...
@@ -84,7 +86,7 @@ public:
template
<
class
Manager
>
error
init
(
Manager
&
manager
)
{
auto
test_bytes
=
as_bytes
(
make_span
(
hello_test
));
write_buf_
.
insert
(
write_
buf_
.
end
(),
test_bytes
.
begin
(),
test_bytes
.
end
());
buf_
.
insert
(
buf_
.
end
(),
test_bytes
.
begin
(),
test_bytes
.
end
());
CAF_CHECK
(
manager
.
mask_add
(
operation
::
read_write
));
return
none
;
}
...
...
@@ -104,18 +106,17 @@ public:
bool
handle_write_event
(
Manager
&
mgr
)
{
for
(
auto
x
=
mgr
.
next_message
();
x
!=
nullptr
;
x
=
mgr
.
next_message
())
{
auto
&
payload
=
x
->
payload
;
write_buf_
.
insert
(
write_
buf_
.
end
(),
payload
.
begin
(),
payload
.
end
());
buf_
.
insert
(
buf_
.
end
(),
payload
.
begin
(),
payload
.
end
());
}
auto
res
=
write
(
handle_
,
make_span
(
write_
buf_
));
auto
res
=
write
(
handle_
,
make_span
(
buf_
));
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
res
))
{
write_buf_
.
erase
(
write_buf_
.
begin
(),
write_
buf_
.
begin
()
+
*
num_bytes
);
return
write_
buf_
.
size
()
>
0
;
buf_
.
erase
(
buf_
.
begin
(),
buf_
.
begin
()
+
*
num_bytes
);
return
buf_
.
size
()
>
0
;
}
return
get
<
sec
>
(
res
)
==
sec
::
unavailable_or_would_block
;
}
template
<
class
Manager
>
void
handle_error
(
Manager
&
,
sec
)
{
void
handle_error
(
sec
)
{
// nop
}
...
...
@@ -143,7 +144,7 @@ private:
std
::
vector
<
byte
>
read_buf_
;
std
::
vector
<
byte
>
write_
buf_
;
std
::
vector
<
byte
>
buf_
;
};
}
// namespace
...
...
@@ -160,8 +161,7 @@ CAF_TEST(send and receive) {
sec
::
unavailable_or_would_block
);
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
close
(
sockets
.
second
);
});
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
dummy_transport
{
sockets
.
first
,
buf
},
dummy_application
{});
dummy_transport
{
sockets
.
first
,
buf
});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
mpx
->
handle_updates
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
2u
);
...
...
@@ -184,8 +184,7 @@ CAF_TEST(resolve and proxy communication) {
nonblocking
(
sockets
.
second
,
true
);
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
close
(
sockets
.
second
);
});
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
dummy_transport
{
sockets
.
first
,
buf
},
dummy_application
{});
dummy_transport
{
sockets
.
first
,
buf
});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
mpx
->
handle_updates
();
run
();
...
...
libcaf_net/test/socket_guard.cpp
View file @
72ed1769
...
...
@@ -37,6 +37,8 @@ struct dummy_socket {
// nop
}
dummy_socket
(
const
dummy_socket
&
)
=
default
;
dummy_socket
&
operator
=
(
const
dummy_socket
&
other
)
{
id
=
other
.
id
;
closed
=
other
.
closed
;
...
...
libcaf_net/test/s
cribe
.cpp
→
libcaf_net/test/s
tream_transport
.cpp
View file @
72ed1769
...
...
@@ -16,9 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE s
cribe_policy
#define CAF_SUITE s
tream_transport
#include "caf/
policy/scribe
.hpp"
#include "caf/
net/stream_transport
.hpp"
#include "caf/test/dsl.hpp"
...
...
@@ -75,7 +75,7 @@ public:
template
<
class
Transport
>
void
write_message
(
Transport
&
transport
,
std
::
unique_ptr
<
endpoint_manager
::
message
>
msg
)
{
transport
.
write_packet
(
msg
->
payload
);
transport
.
write_packet
(
span
<
byte
>
{},
msg
->
payload
);
}
template
<
class
Parent
>
...
...
@@ -123,24 +123,25 @@ private:
CAF_TEST_FIXTURE_SCOPE
(
endpoint_manager_tests
,
fixture
)
CAF_TEST
(
receive
)
{
using
transport_type
=
stream_transport
<
dummy_application
>
;
std
::
vector
<
byte
>
read_buf
(
1024
);
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
auto
buf
=
std
::
make_shared
<
std
::
vector
<
byte
>>
();
auto
sockets
=
unbox
(
make_stream_socket_pair
());
nonblocking
(
sockets
.
second
,
true
);
if
(
auto
err
=
nonblocking
(
sockets
.
second
,
true
))
CAF_FAIL
(
"nonblocking() returned an error: "
<<
err
);
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
make_span
(
read_buf
)),
sec
::
unavailable_or_would_block
);
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
close
(
sockets
.
second
);
});
CAF_MESSAGE
(
"configure scribe_policy"
);
policy
::
scribe
scribe
{
sockets
.
first
};
scribe
.
configure_read
(
net
::
receive_policy
::
exactly
(
hello_manager
.
size
()));
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
scribe
,
dummy_application
{
buf
});
transport_type
transport
{
sockets
.
first
,
dummy_application
{
buf
}};
transport
.
configure_read
(
net
::
receive_policy
::
exactly
(
hello_manager
.
size
()));
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
transport
);
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
mpx
->
handle_updates
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
2u
);
CAF_MESSAGE
(
"sending data to scribe_policy"
);
CAF_CHECK_EQUAL
(
write
(
sockets
.
second
,
as_bytes
(
make_span
(
hello_manager
))),
hello_manager
.
size
());
CAF_MESSAGE
(
"wrote "
<<
hello_manager
.
size
()
<<
" bytes."
);
run
();
CAF_CHECK_EQUAL
(
string_view
(
reinterpret_cast
<
char
*>
(
buf
->
data
()),
buf
->
size
()),
...
...
@@ -148,13 +149,15 @@ CAF_TEST(receive) {
}
CAF_TEST
(
resolve
and
proxy
communication
)
{
using
transport_type
=
stream_transport
<
dummy_application
>
;
std
::
vector
<
byte
>
read_buf
(
1024
);
auto
buf
=
std
::
make_shared
<
std
::
vector
<
byte
>>
();
auto
sockets
=
unbox
(
make_stream_socket_pair
());
nonblocking
(
sockets
.
second
,
true
);
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
close
(
sockets
.
second
);
});
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
policy
::
scribe
{
sockets
.
first
},
dummy_application
{
buf
});
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
sockets
.
first
,
dummy_application
{
buf
}});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
mpx
->
handle_updates
();
run
();
...
...
libcaf_net/test/string_application.cpp
View file @
72ed1769
...
...
@@ -18,7 +18,7 @@
#define CAF_SUITE string_application
#include "caf/
policy/scribe
.hpp"
#include "caf/
net/stream_transport
.hpp"
#include "caf/test/dsl.hpp"
...
...
@@ -100,12 +100,9 @@ public:
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
net
::
endpoint_manager
::
message
>
msg
)
{
std
::
vector
<
byte
>
buf
;
header_type
header
{
static_cast
<
uint32_t
>
(
msg
->
payload
.
size
())};
buf
.
resize
(
sizeof
(
header_type
));
memcpy
(
buf
.
data
(),
&
header
,
buf
.
size
());
buf
.
insert
(
buf
.
end
(),
msg
->
payload
.
begin
(),
msg
->
payload
.
end
());
parent
.
write_packet
(
buf
);
std
::
vector
<
byte
>
payload
(
msg
->
payload
.
begin
(),
msg
->
payload
.
end
());
parent
.
write_packet
(
as_bytes
(
make_span
(
&
header
,
1
)),
make_span
(
payload
));
}
static
expected
<
std
::
vector
<
byte
>>
serialize
(
actor_system
&
sys
,
...
...
@@ -152,7 +149,8 @@ public:
if
(
header_
.
payload
==
0
)
Base
::
handle_packet
(
parent
,
header_
,
span
<
const
byte
>
{});
else
parent
.
configure_read
(
net
::
receive_policy
::
exactly
(
header_
.
payload
));
parent
.
transport
().
configure_read
(
net
::
receive_policy
::
exactly
(
header_
.
payload
));
await_payload_
=
true
;
}
}
...
...
@@ -191,6 +189,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
>
;
std
::
vector
<
byte
>
read_buf
(
1024
);
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
auto
buf
=
std
::
make_shared
<
std
::
vector
<
byte
>>
();
...
...
@@ -199,13 +198,15 @@ CAF_TEST(receive) {
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
make_span
(
read_buf
)),
sec
::
unavailable_or_would_block
);
CAF_MESSAGE
(
"adding both endpoint managers"
);
auto
mgr1
=
make_endpoint_manager
(
mpx
,
sys
,
policy
::
scribe
{
sockets
.
first
},
application_type
{
sys
,
buf
});
auto
mgr1
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
sockets
.
first
,
application_type
{
sys
,
buf
}});
CAF_CHECK_EQUAL
(
mgr1
->
init
(),
none
);
mpx
->
handle_updates
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
2u
);
auto
mgr2
=
make_endpoint_manager
(
mpx
,
sys
,
policy
::
scribe
{
sockets
.
second
},
application_type
{
sys
,
buf
});
auto
mgr2
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
sockets
.
second
,
application_type
{
sys
,
buf
}});
CAF_CHECK_EQUAL
(
mgr2
->
init
(),
none
);
mpx
->
handle_updates
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
3u
);
...
...
libcaf_net/test/transport_worker.cpp
0 → 100644
View file @
72ed1769
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE transport_worker
#include "caf/net/transport_worker.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
namespace
{
constexpr
string_view
hello_test
=
"hello test!"
;
struct
application_result
{
bool
initialized
;
std
::
vector
<
byte
>
data_buffer
;
std
::
string
resolve_path
;
actor
resolve_listener
;
atom_value
timeout_value
;
uint64_t
timeout_id
;
sec
err
;
};
struct
transport_result
{
std
::
vector
<
byte
>
packet_buffer
;
ip_endpoint
ep
;
};
class
dummy_application
{
public:
dummy_application
(
std
::
shared_ptr
<
application_result
>
res
)
:
res_
(
std
::
move
(
res
)){
// nop
};
~
dummy_application
()
=
default
;
template
<
class
Parent
>
error
init
(
Parent
&
)
{
res_
->
initialized
=
true
;
return
none
;
}
template
<
class
Transport
>
void
write_message
(
Transport
&
transport
,
std
::
unique_ptr
<
endpoint_manager
::
message
>
msg
)
{
transport
.
write_packet
(
span
<
byte
>
{},
msg
->
payload
);
}
template
<
class
Parent
>
void
handle_data
(
Parent
&
,
span
<
const
byte
>
data
)
{
auto
&
buf
=
res_
->
data_buffer
;
buf
.
clear
();
buf
.
insert
(
buf
.
begin
(),
data
.
begin
(),
data
.
end
());
}
template
<
class
Parent
>
void
resolve
(
Parent
&
,
const
std
::
string
&
path
,
actor
listener
)
{
res_
->
resolve_path
=
path
;
res_
->
resolve_listener
=
std
::
move
(
listener
);
}
template
<
class
Parent
>
void
timeout
(
Parent
&
,
atom_value
value
,
uint64_t
id
)
{
res_
->
timeout_value
=
value
;
res_
->
timeout_id
=
id
;
}
void
handle_error
(
sec
err
)
{
res_
->
err
=
err
;
}
static
expected
<
std
::
vector
<
byte
>>
serialize
(
actor_system
&
sys
,
const
type_erased_tuple
&
x
)
{
std
::
vector
<
byte
>
result
;
serializer_impl
<
std
::
vector
<
byte
>>
sink
{
sys
,
result
};
if
(
auto
err
=
message
::
save
(
sink
,
x
))
return
err
;
return
result
;
}
private:
std
::
shared_ptr
<
application_result
>
res_
;
};
class
dummy_transport
{
public:
using
transport_type
=
dummy_transport
;
using
application_type
=
dummy_application
;
dummy_transport
(
std
::
shared_ptr
<
transport_result
>
res
)
:
res_
(
res
)
{
// nop
}
void
write_packet
(
span
<
const
byte
>
header
,
span
<
const
byte
>
payload
,
ip_endpoint
ep
)
{
auto
&
buf
=
res_
->
packet_buffer
;
buf
.
insert
(
buf
.
begin
(),
header
.
begin
(),
header
.
end
());
buf
.
insert
(
buf
.
begin
(),
payload
.
begin
(),
payload
.
end
());
res_
->
ep
=
ep
;
}
private:
std
::
shared_ptr
<
transport_result
>
res_
;
};
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
using
worker_type
=
transport_worker
<
dummy_application
,
ip_endpoint
>
;
fixture
()
:
transport_results
{
std
::
make_shared
<
transport_result
>
()},
application_results
{
std
::
make_shared
<
application_result
>
()},
transport
(
transport_results
),
worker
{
dummy_application
{
application_results
}}
{
mpx
=
std
::
make_shared
<
multiplexer
>
();
if
(
auto
err
=
mpx
->
init
())
CAF_FAIL
(
"mpx->init failed: "
<<
sys
.
render
(
err
));
if
(
auto
err
=
parse
(
"[::1]:12345"
,
ep
))
CAF_FAIL
(
"parse returned an error: "
<<
err
);
worker
=
worker_type
{
dummy_application
{
application_results
},
ep
};
}
bool
handle_io_event
()
override
{
mpx
->
handle_updates
();
return
mpx
->
poll_once
(
false
);
}
multiplexer_ptr
mpx
;
std
::
shared_ptr
<
transport_result
>
transport_results
;
std
::
shared_ptr
<
application_result
>
application_results
;
dummy_transport
transport
;
worker_type
worker
;
ip_endpoint
ep
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
endpoint_manager_tests
,
fixture
)
CAF_TEST
(
construction
and
initialization
)
{
CAF_CHECK_EQUAL
(
worker
.
init
(
transport
),
none
);
CAF_CHECK_EQUAL
(
application_results
->
initialized
,
true
);
}
CAF_TEST
(
handle_data
)
{
auto
test_span
=
as_bytes
(
make_span
(
hello_test
));
worker
.
handle_data
(
transport
,
test_span
);
auto
&
buf
=
application_results
->
data_buffer
;
string_view
result
{
reinterpret_cast
<
char
*>
(
buf
.
data
()),
buf
.
size
()};
CAF_CHECK_EQUAL
(
result
,
hello_test
);
}
CAF_TEST
(
write_message
)
{
actor
act
;
auto
strong_actor
=
actor_cast
<
strong_actor_ptr
>
(
act
);
mailbox_element
::
forwarding_stack
stack
;
auto
msg
=
make_message
();
auto
elem
=
make_mailbox_element
(
strong_actor
,
make_message_id
(
12345
),
stack
,
msg
);
auto
test_span
=
make_span
(
reinterpret_cast
<
byte
*>
(
const_cast
<
char
*>
(
hello_test
.
data
())),
hello_test
.
size
());
std
::
vector
<
byte
>
payload
(
test_span
.
begin
(),
test_span
.
end
());
auto
message
=
detail
::
make_unique
<
endpoint_manager
::
message
>
(
std
::
move
(
elem
),
payload
);
worker
.
write_message
(
transport
,
std
::
move
(
message
));
auto
&
buf
=
transport_results
->
packet_buffer
;
string_view
result
{
reinterpret_cast
<
char
*>
(
buf
.
data
()),
buf
.
size
()};
CAF_CHECK_EQUAL
(
result
,
hello_test
);
CAF_CHECK_EQUAL
(
transport_results
->
ep
,
ep
);
}
CAF_TEST
(
resolve
)
{
worker
.
resolve
(
transport
,
"foo"
,
self
);
CAF_CHECK_EQUAL
(
application_results
->
resolve_path
,
"foo"
);
CAF_CHECK_EQUAL
(
application_results
->
resolve_listener
,
self
);
}
CAF_TEST
(
timeout
)
{
worker
.
timeout
(
transport
,
atom
(
"bar"
),
42u
);
CAF_CHECK_EQUAL
(
application_results
->
timeout_value
,
atom
(
"bar"
));
CAF_CHECK_EQUAL
(
application_results
->
timeout_id
,
42u
);
}
CAF_TEST
(
handle_error
)
{
worker
.
handle_error
(
sec
::
feature_disabled
);
CAF_CHECK_EQUAL
(
application_results
->
err
,
sec
::
feature_disabled
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
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