Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
A
actor-incubator
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
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-incubator
Commits
2e9fe41b
Unverified
Commit
2e9fe41b
authored
May 08, 2020
by
Dominik Charousset
Committed by
GitHub
May 08, 2020
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #67
Move serialization to multiplexer
parents
54240489
73853282
Changes
17
Hide whitespace changes
Inline
Side-by-side
Showing
17 changed files
with
135 additions
and
211 deletions
+135
-211
CMakeLists.txt
CMakeLists.txt
+1
-1
libcaf_net/caf/net/actor_proxy_impl.hpp
libcaf_net/caf/net/actor_proxy_impl.hpp
+0
-1
libcaf_net/caf/net/basp/application.hpp
libcaf_net/caf/net/basp/application.hpp
+0
-2
libcaf_net/caf/net/endpoint_manager.hpp
libcaf_net/caf/net/endpoint_manager.hpp
+1
-11
libcaf_net/caf/net/endpoint_manager_impl.hpp
libcaf_net/caf/net/endpoint_manager_impl.hpp
+2
-6
libcaf_net/caf/net/endpoint_manager_queue.hpp
libcaf_net/caf/net/endpoint_manager_queue.hpp
+5
-9
libcaf_net/src/actor_proxy_impl.cpp
libcaf_net/src/actor_proxy_impl.cpp
+5
-8
libcaf_net/src/application.cpp
libcaf_net/src/application.cpp
+6
-14
libcaf_net/src/endpoint_manager.cpp
libcaf_net/src/endpoint_manager.cpp
+2
-4
libcaf_net/src/net/endpoint_manager_queue.cpp
libcaf_net/src/net/endpoint_manager_queue.cpp
+2
-4
libcaf_net/test/datagram_transport.cpp
libcaf_net/test/datagram_transport.cpp
+15
-23
libcaf_net/test/doorman.cpp
libcaf_net/test/doorman.cpp
+5
-15
libcaf_net/test/endpoint_manager.cpp
libcaf_net/test/endpoint_manager.cpp
+7
-14
libcaf_net/test/stream_transport.cpp
libcaf_net/test/stream_transport.cpp
+14
-22
libcaf_net/test/string_application.cpp
libcaf_net/test/string_application.cpp
+22
-31
libcaf_net/test/transport_worker.cpp
libcaf_net/test/transport_worker.cpp
+28
-27
libcaf_net/test/transport_worker_dispatcher.cpp
libcaf_net/test/transport_worker_dispatcher.cpp
+20
-19
No files found.
CMakeLists.txt
View file @
2e9fe41b
...
...
@@ -50,7 +50,7 @@ if(CAF_INC_ENABLE_STANDALONE_BUILD)
FetchContent_Declare
(
actor_framework
GIT_REPOSITORY https://github.com/actor-framework/actor-framework.git
GIT_TAG
d7f70e997
GIT_TAG
94e26b315
)
FetchContent_Populate
(
actor_framework
)
set
(
CAF_ENABLE_EXAMPLES OFF CACHE BOOL
""
FORCE
)
...
...
libcaf_net/caf/net/actor_proxy_impl.hpp
View file @
2e9fe41b
...
...
@@ -37,7 +37,6 @@ public:
void
kill_proxy
(
execution_unit
*
ctx
,
error
rsn
)
override
;
private:
endpoint_manager
::
serialize_fun_type
sf_
;
endpoint_manager_ptr
dst_
;
};
...
...
libcaf_net/caf/net/basp/application.hpp
View file @
2e9fe41b
...
...
@@ -136,8 +136,6 @@ public:
// nop
}
static
expected
<
buffer_type
>
serialize
(
actor_system
&
sys
,
const
message
&
x
);
// -- utility functions ------------------------------------------------------
strong_actor_ptr
resolve_local_path
(
string_view
path
);
...
...
libcaf_net/caf/net/endpoint_manager.hpp
View file @
2e9fe41b
...
...
@@ -44,12 +44,6 @@ public:
using
super
=
socket_manager
;
/// Represents either an error or a serialized payload.
using
maybe_buffer
=
expected
<
std
::
vector
<
byte
>>
;
/// A function type for serializing message payloads.
using
serialize_fun_type
=
maybe_buffer
(
*
)(
actor_system
&
,
const
message
&
);
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager
(
socket
handle
,
const
multiplexer_ptr
&
parent
,
...
...
@@ -71,8 +65,7 @@ public:
void
resolve
(
uri
locator
,
actor
listener
);
/// Enqueues a message to the endpoint.
void
enqueue
(
mailbox_element_ptr
msg
,
strong_actor_ptr
receiver
,
std
::
vector
<
byte
>
payload
);
void
enqueue
(
mailbox_element_ptr
msg
,
strong_actor_ptr
receiver
);
/// Enqueues an event to the endpoint.
template
<
class
...
Ts
>
...
...
@@ -85,9 +78,6 @@ public:
/// Initializes the manager before adding it to the multiplexer's event loop.
virtual
error
init
()
=
0
;
/// @returns the protocol-specific function for serializing payloads.
virtual
serialize_fun_type
serialize_fun
()
const
noexcept
=
0
;
protected:
bool
enqueue
(
endpoint_manager_queue
::
element
*
ptr
);
...
...
libcaf_net/caf/net/endpoint_manager_impl.hpp
View file @
2e9fe41b
...
...
@@ -62,8 +62,8 @@ public:
// -- timeout management -----------------------------------------------------
template
<
class
...
Ts
>
uint64_t
set_timeout
(
actor_clock
::
time_point
tp
,
std
::
string
type
,
Ts
&&
...
xs
)
{
uint64_t
set_timeout
(
actor_clock
::
time_point
tp
,
std
::
string
type
,
Ts
&&
...
xs
)
{
auto
act
=
actor_cast
<
abstract_actor
*>
(
timeout_proxy_
);
CAF_ASSERT
(
act
!=
nullptr
);
sys_
.
clock
().
set_multi_timeout
(
tp
,
act
,
std
::
move
(
type
),
next_timeout_id_
);
...
...
@@ -119,10 +119,6 @@ public:
transport_
.
handle_error
(
code
);
}
serialize_fun_type
serialize_fun
()
const
noexcept
override
{
return
application_type
::
serialize
;
}
private:
transport_type
transport_
;
...
...
libcaf_net/caf/net/endpoint_manager_queue.hpp
View file @
2e9fe41b
...
...
@@ -21,6 +21,7 @@
#include <string>
#include "caf/actor.hpp"
#include "caf/detail/serialized_size.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
...
...
@@ -112,7 +113,7 @@ public:
// nop
}
task_size_type
task_size
(
const
event
&
)
const
noexcept
{
static
constexpr
task_size_type
task_size
(
const
event
&
)
noexcept
{
return
1
;
}
};
...
...
@@ -125,11 +126,7 @@ public:
/// ID of the receiving actor.
strong_actor_ptr
receiver
;
/// Serialized representation of of `msg->content()`.
std
::
vector
<
byte
>
payload
;
message
(
mailbox_element_ptr
msg
,
strong_actor_ptr
receiver
,
std
::
vector
<
byte
>
payload
);
message
(
mailbox_element_ptr
msg
,
strong_actor_ptr
receiver
);
~
message
()
override
;
...
...
@@ -153,9 +150,8 @@ public:
// nop
}
static
task_size_type
task_size
(
const
message
&
x
)
noexcept
{
// Return at least 1 if the payload is empty.
return
x
.
payload
.
size
()
+
static_cast
<
task_size_type
>
(
x
.
payload
.
empty
());
static
task_size_type
task_size
(
const
message
&
msg
)
noexcept
{
return
detail
::
serialized_size
(
msg
.
msg
->
content
());
}
};
...
...
libcaf_net/src/actor_proxy_impl.cpp
View file @
2e9fe41b
...
...
@@ -25,7 +25,7 @@
namespace
caf
::
net
{
actor_proxy_impl
::
actor_proxy_impl
(
actor_config
&
cfg
,
endpoint_manager_ptr
dst
)
:
super
(
cfg
),
sf_
(
dst
->
serialize_fun
()),
dst_
(
std
::
move
(
dst
))
{
:
super
(
cfg
),
dst_
(
std
::
move
(
dst
))
{
CAF_ASSERT
(
dst_
!=
nullptr
);
dst_
->
enqueue_event
(
node
(),
id
());
}
...
...
@@ -34,14 +34,11 @@ actor_proxy_impl::~actor_proxy_impl() {
// nop
}
void
actor_proxy_impl
::
enqueue
(
mailbox_element_ptr
what
,
execution_unit
*
)
{
void
actor_proxy_impl
::
enqueue
(
mailbox_element_ptr
msg
,
execution_unit
*
)
{
CAF_PUSH_AID
(
0
);
CAF_ASSERT
(
what
!=
nullptr
);
CAF_LOG_SEND_EVENT
(
what
);
if
(
auto
payload
=
sf_
(
home_system
(),
what
->
content
()))
dst_
->
enqueue
(
std
::
move
(
what
),
ctrl
(),
std
::
move
(
*
payload
));
else
CAF_LOG_ERROR
(
"unable to serialize payload: "
<<
payload
.
error
());
CAF_ASSERT
(
msg
!=
nullptr
);
CAF_LOG_SEND_EVENT
(
msg
);
dst_
->
enqueue
(
std
::
move
(
msg
),
ctrl
());
}
void
actor_proxy_impl
::
kill_proxy
(
execution_unit
*
ctx
,
error
rsn
)
{
...
...
libcaf_net/src/application.cpp
View file @
2e9fe41b
...
...
@@ -52,14 +52,14 @@ error application::write_message(
CAF_ASSERT
(
ptr
!=
nullptr
);
CAF_ASSERT
(
ptr
->
msg
!=
nullptr
);
CAF_LOG_TRACE
(
CAF_ARG2
(
"content"
,
ptr
->
msg
->
content
()));
auto
payload_prefix
=
writer
.
next_payload_buffer
();
binary_serializer
sink
{
system
(),
payload_prefix
};
const
auto
&
src
=
ptr
->
msg
->
sender
;
const
auto
&
dst
=
ptr
->
receiver
;
if
(
dst
==
nullptr
)
{
// TODO: valid?
return
none
;
}
auto
payload_buf
=
writer
.
next_payload_buffer
();
binary_serializer
sink
{
system
(),
payload_buf
};
if
(
src
!=
nullptr
)
{
auto
src_id
=
src
->
id
();
system
().
registry
().
put
(
src_id
,
src
);
...
...
@@ -69,13 +69,14 @@ error application::write_message(
if
(
auto
err
=
sink
(
node_id
{},
actor_id
{
0
},
dst
->
id
(),
ptr
->
msg
->
stages
))
return
err
;
}
if
(
auto
err
=
sink
(
ptr
->
msg
->
content
()))
return
err
;
auto
hdr
=
writer
.
next_header_buffer
();
to_bytes
(
header
{
message_type
::
actor_message
,
static_cast
<
uint32_t
>
(
payload_prefix
.
size
()
+
ptr
->
payload
.
size
()),
static_cast
<
uint32_t
>
(
payload_buf
.
size
()),
ptr
->
msg
->
mid
.
integer_value
()},
hdr
);
writer
.
write_packet
(
hdr
,
payload_
prefix
,
ptr
->
payload
);
writer
.
write_packet
(
hdr
,
payload_
buf
);
return
none
;
}
...
...
@@ -118,15 +119,6 @@ void application::local_actor_down(packet_writer& writer, actor_id id,
writer
.
write_packet
(
hdr
,
payload
);
}
expected
<
std
::
vector
<
byte
>>
application
::
serialize
(
actor_system
&
sys
,
const
message
&
x
)
{
std
::
vector
<
byte
>
result
;
binary_serializer
sink
{
sys
,
result
};
if
(
auto
err
=
x
.
save
(
sink
))
return
err
.
value
();
return
result
;
}
strong_actor_ptr
application
::
resolve_local_path
(
string_view
path
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
path
));
// We currently support two path formats: `id/<actor_id>` and `name/<atom>`.
...
...
libcaf_net/src/endpoint_manager.cpp
View file @
2e9fe41b
...
...
@@ -60,11 +60,9 @@ void endpoint_manager::resolve(uri locator, actor listener) {
}
void
endpoint_manager
::
enqueue
(
mailbox_element_ptr
msg
,
strong_actor_ptr
receiver
,
std
::
vector
<
byte
>
payload
)
{
strong_actor_ptr
receiver
)
{
using
message_type
=
endpoint_manager_queue
::
message
;
auto
ptr
=
new
message_type
(
std
::
move
(
msg
),
std
::
move
(
receiver
),
std
::
move
(
payload
));
auto
ptr
=
new
message_type
(
std
::
move
(
msg
),
std
::
move
(
receiver
));
enqueue
(
ptr
);
}
...
...
libcaf_net/src/net/endpoint_manager_queue.cpp
View file @
2e9fe41b
...
...
@@ -56,12 +56,10 @@ size_t endpoint_manager_queue::event::task_size() const noexcept {
}
endpoint_manager_queue
::
message
::
message
(
mailbox_element_ptr
msg
,
strong_actor_ptr
receiver
,
std
::
vector
<
byte
>
payload
)
strong_actor_ptr
receiver
)
:
element
(
element_type
::
message
),
msg
(
std
::
move
(
msg
)),
receiver
(
std
::
move
(
receiver
)),
payload
(
std
::
move
(
payload
))
{
receiver
(
std
::
move
(
receiver
))
{
// nop
}
...
...
libcaf_net/test/datagram_transport.cpp
View file @
2e9fe41b
...
...
@@ -119,10 +119,14 @@ public:
return
none
;
}
template
<
class
Transpor
t
>
void
write_message
(
Transport
&
transpor
t
,
template
<
class
Paren
t
>
void
write_message
(
Parent
&
paren
t
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
msg
)
{
transport
.
write_packet
(
msg
->
payload
);
auto
payload_buf
=
parent
.
next_payload_buffer
();
binary_serializer
sink
{
parent
.
system
(),
payload_buf
};
if
(
auto
err
=
sink
(
msg
->
msg
->
payload
))
CAF_FAIL
(
"serializing failed: "
<<
err
);
parent
.
write_packet
(
payload_buf
);
}
template
<
class
Parent
>
...
...
@@ -139,10 +143,8 @@ public:
auto
nid
=
make_node_id
(
uri
);
actor_config
cfg
;
endpoint_manager_ptr
ptr
{
&
parent
.
manager
()};
auto
p
=
make_actor
<
actor_proxy_impl
,
strong_actor_ptr
>
(
aid
,
nid
,
&
parent
.
system
(),
cfg
,
std
::
move
(
ptr
));
auto
p
=
make_actor
<
actor_proxy_impl
,
strong_actor_ptr
>
(
aid
,
nid
,
&
parent
.
system
(),
cfg
,
std
::
move
(
ptr
));
anon_send
(
listener
,
resolve_atom_v
,
std
::
string
{
path
.
begin
(),
path
.
end
()},
p
);
}
...
...
@@ -166,14 +168,6 @@ public:
CAF_FAIL
(
"handle_error called: "
<<
to_string
(
sec
));
}
static
expected
<
buffer_type
>
serialize
(
actor_system
&
sys
,
const
message
&
x
)
{
buffer_type
result
;
binary_serializer
sink
{
sys
,
result
};
if
(
auto
err
=
x
.
save
(
sink
))
return
err
.
value
();
return
result
;
}
private:
buffer_ptr
rec_buf_
;
};
...
...
@@ -204,10 +198,9 @@ CAF_TEST(receive) {
using
transport_type
=
datagram_transport
<
dummy_application_factory
>
;
if
(
auto
err
=
nonblocking
(
recv_socket
,
true
))
CAF_FAIL
(
"nonblocking() returned an error: "
<<
err
);
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
recv_socket
,
dummy_application_factory
{
shared_buf
}});
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
recv_socket
,
dummy_application_factory
{
shared_buf
}});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
auto
mgr_impl
=
mgr
.
downcast
<
endpoint_manager_impl
<
transport_type
>>
();
CAF_CHECK
(
mgr_impl
!=
nullptr
);
...
...
@@ -227,10 +220,9 @@ CAF_TEST(resolve and proxy communication) {
using
transport_type
=
datagram_transport
<
dummy_application_factory
>
;
buffer_type
recv_buf
(
1024
);
auto
uri
=
unbox
(
make_uri
(
"test:/id/42"
));
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
send_socket
,
dummy_application_factory
{
shared_buf
}});
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
send_socket
,
dummy_application_factory
{
shared_buf
}});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
auto
mgr_impl
=
mgr
.
downcast
<
endpoint_manager_impl
<
transport_type
>>
();
CAF_CHECK
(
mgr_impl
!=
nullptr
);
...
...
libcaf_net/test/doorman.cpp
View file @
2e9fe41b
...
...
@@ -60,15 +60,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
class
dummy_application
{
public:
static
expected
<
std
::
vector
<
byte
>>
serialize
(
actor_system
&
sys
,
const
message
&
x
)
{
std
::
vector
<
byte
>
result
;
binary_serializer
sink
{
sys
,
result
};
if
(
auto
err
=
x
.
save
(
sink
))
return
err
.
value
();
return
result
;
}
template
<
class
Parent
>
error
init
(
Parent
&
)
{
return
none
;
...
...
@@ -77,7 +68,11 @@ public:
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
msg
)
{
parent
.
write_packet
(
msg
->
payload
);
auto
payload_buf
=
parent
.
next_payload_buffer
();
binary_serializer
sink
{
parent
.
system
(),
payload_buf
};
if
(
auto
err
=
sink
(
msg
->
msg
->
payload
))
CAF_FAIL
(
"serializing failed: "
<<
err
);
parent
.
write_packet
(
payload_buf
);
}
template
<
class
Parent
>
...
...
@@ -116,11 +111,6 @@ class dummy_application_factory {
public:
using
application_type
=
dummy_application
;
static
expected
<
std
::
vector
<
byte
>>
serialize
(
actor_system
&
sys
,
const
message
&
x
)
{
return
dummy_application
::
serialize
(
sys
,
x
);
}
template
<
class
Parent
>
error
init
(
Parent
&
)
{
return
none
;
...
...
libcaf_net/test/endpoint_manager.cpp
View file @
2e9fe41b
...
...
@@ -23,6 +23,7 @@
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp"
...
...
@@ -61,15 +62,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
};
class
dummy_application
{
public:
static
expected
<
std
::
vector
<
byte
>>
serialize
(
actor_system
&
sys
,
const
message
&
x
)
{
std
::
vector
<
byte
>
result
;
binary_serializer
sink
{
sys
,
result
};
if
(
auto
err
=
x
.
save
(
sink
))
return
err
.
value
();
return
result
;
}
// nop
};
class
dummy_transport
{
...
...
@@ -107,8 +100,9 @@ public:
template
<
class
Manager
>
bool
handle_write_event
(
Manager
&
mgr
)
{
for
(
auto
x
=
mgr
.
next_message
();
x
!=
nullptr
;
x
=
mgr
.
next_message
())
{
auto
&
payload
=
x
->
payload
;
buf_
.
insert
(
buf_
.
end
(),
payload
.
begin
(),
payload
.
end
());
binary_serializer
sink
{
mgr
.
system
(),
buf_
};
if
(
auto
err
=
sink
(
x
->
msg
->
payload
))
CAF_FAIL
(
"serializing failed: "
<<
err
);
}
auto
res
=
write
(
handle_
,
buf_
);
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
res
))
{
...
...
@@ -128,9 +122,8 @@ public:
auto
hid
=
string_view
(
"0011223344556677889900112233445566778899"
);
auto
nid
=
unbox
(
make_node_id
(
42
,
hid
));
actor_config
cfg
;
auto
p
=
make_actor
<
actor_proxy_impl
,
strong_actor_ptr
>
(
aid
,
nid
,
&
mgr
.
system
(),
cfg
,
&
mgr
);
auto
p
=
make_actor
<
actor_proxy_impl
,
strong_actor_ptr
>
(
aid
,
nid
,
&
mgr
.
system
(),
cfg
,
&
mgr
);
std
::
string
path
{
locator
.
path
().
begin
(),
locator
.
path
().
end
()};
anon_send
(
listener
,
resolve_atom_v
,
std
::
move
(
path
),
p
);
}
...
...
libcaf_net/test/stream_transport.cpp
View file @
2e9fe41b
...
...
@@ -92,8 +92,12 @@ public:
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
ptr
)
{
parent
.
write_packet
(
ptr
->
payload
);
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
msg
)
{
auto
payload_buf
=
parent
.
next_payload_buffer
();
binary_serializer
sink
{
parent
.
system
(),
payload_buf
};
if
(
auto
err
=
sink
(
msg
->
msg
->
payload
))
CAF_FAIL
(
"serializing failed: "
<<
err
);
parent
.
write_packet
(
payload_buf
);
}
template
<
class
Parent
>
...
...
@@ -110,10 +114,8 @@ public:
auto
nid
=
unbox
(
make_node_id
(
42
,
hid
));
actor_config
cfg
;
endpoint_manager_ptr
ptr
{
&
parent
.
manager
()};
auto
p
=
make_actor
<
actor_proxy_impl
,
strong_actor_ptr
>
(
aid
,
nid
,
&
parent
.
system
(),
cfg
,
std
::
move
(
ptr
));
auto
p
=
make_actor
<
actor_proxy_impl
,
strong_actor_ptr
>
(
aid
,
nid
,
&
parent
.
system
(),
cfg
,
std
::
move
(
ptr
));
anon_send
(
listener
,
resolve_atom_v
,
std
::
string
{
path
.
begin
(),
path
.
end
()},
p
);
}
...
...
@@ -137,14 +139,6 @@ public:
// nop
}
static
expected
<
buffer_type
>
serialize
(
actor_system
&
sys
,
const
message
&
x
)
{
buffer_type
result
;
binary_serializer
sink
{
sys
,
result
};
if
(
auto
err
=
x
.
save
(
sink
))
return
err
.
value
();
return
result
;
}
private:
buffer_ptr
rec_buf_
;
};
...
...
@@ -155,10 +149,9 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST
(
receive
)
{
using
transport_type
=
stream_transport
<
dummy_application
>
;
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
recv_socket_guard
.
release
(),
dummy_application
{
shared_buf
}});
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
recv_socket_guard
.
release
(),
dummy_application
{
shared_buf
}});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
auto
mgr_impl
=
mgr
.
downcast
<
endpoint_manager_impl
<
transport_type
>>
();
CAF_CHECK
(
mgr_impl
!=
nullptr
);
...
...
@@ -177,10 +170,9 @@ CAF_TEST(receive) {
CAF_TEST
(
resolve
and
proxy
communication
)
{
using
transport_type
=
stream_transport
<
dummy_application
>
;
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
send_socket_guard
.
release
(),
dummy_application
{
shared_buf
}});
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
send_socket_guard
.
release
(),
dummy_application
{
shared_buf
}});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
run
();
mgr
->
resolve
(
unbox
(
make_uri
(
"test:/id/42"
)),
self
);
...
...
libcaf_net/test/string_application.cpp
View file @
2e9fe41b
...
...
@@ -66,8 +66,8 @@ struct string_application_header {
/// @relates header
template
<
class
Inspector
>
typename
Inspector
::
result_type
inspect
(
Inspector
&
f
,
string_application_header
&
hdr
)
{
typename
Inspector
::
result_type
inspect
(
Inspector
&
f
,
string_application_header
&
hdr
)
{
return
f
(
meta
::
type_name
(
"sa_header"
),
hdr
.
payload
);
}
...
...
@@ -75,8 +75,8 @@ class string_application {
public:
using
header_type
=
string_application_header
;
string_application
(
actor_system
&
sys
,
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf
)
:
sys_
(
sys
),
buf_
(
std
::
move
(
buf
))
{
string_application
(
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf
)
:
buf_
(
std
::
move
(
buf
))
{
// nop
}
...
...
@@ -86,8 +86,8 @@ public:
}
template
<
class
Parent
>
void
handle_packet
(
Parent
&
,
header_type
&
,
span
<
const
byte
>
payload
)
{
binary_deserializer
source
{
sys_
,
payload
};
void
handle_packet
(
Parent
&
parent
,
header_type
&
,
span
<
const
byte
>
payload
)
{
binary_deserializer
source
{
parent
.
system
()
,
payload
};
message
msg
;
if
(
auto
err
=
msg
.
load
(
source
))
CAF_FAIL
(
"unable to deserialize message: "
<<
err
);
...
...
@@ -105,24 +105,18 @@ public:
if
(
ptr
->
msg
==
nullptr
)
return
;
auto
header_buf
=
parent
.
next_header_buffer
();
binary_serializer
sink
{
sys_
,
header_buf
}
;
header_type
header
{
static_cast
<
uint32_t
>
(
ptr
->
payload
.
size
())
};
if
(
auto
err
=
sink
(
header
))
auto
payload_buf
=
parent
.
next_payload_buffer
()
;
binary_serializer
payload_sink
{
parent
.
system
(),
payload_buf
};
if
(
auto
err
=
payload_sink
(
ptr
->
msg
->
payload
))
CAF_FAIL
(
"serializing failed: "
<<
err
);
parent
.
write_packet
(
header_buf
,
ptr
->
payload
);
}
static
expected
<
std
::
vector
<
byte
>>
serialize
(
actor_system
&
sys
,
const
message
&
x
)
{
std
::
vector
<
byte
>
result
;
binary_serializer
sink
{
sys
,
result
};
if
(
auto
err
=
x
.
save
(
sink
))
return
err
.
value
();
return
result
;
binary_serializer
header_sink
{
parent
.
system
(),
header_buf
};
if
(
auto
err
=
header_sink
(
header_type
{
static_cast
<
uint32_t
>
(
payload_buf
.
size
())}))
CAF_FAIL
(
"serializing failed: "
<<
err
);
parent
.
write_packet
(
header_buf
,
payload_buf
);
}
private:
actor_system
&
sys_
;
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf_
;
};
...
...
@@ -131,9 +125,8 @@ class stream_string_application : public Base {
public:
using
header_type
=
typename
Base
::
header_type
;
stream_string_application
(
actor_system
&
sys
,
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf
)
:
Base
(
sys
,
std
::
move
(
buf
)),
await_payload_
(
false
)
{
stream_string_application
(
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf
)
:
Base
(
std
::
move
(
buf
)),
await_payload_
(
false
)
{
// nop
}
...
...
@@ -208,8 +201,8 @@ private:
CAF_TEST_FIXTURE_SCOPE
(
endpoint_manager_tests
,
fixture
)
CAF_TEST
(
receive
)
{
using
application_type
=
extend
<
string_application
>::
with
<
stream_string_application
>
;
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
);
...
...
@@ -219,14 +212,12 @@ CAF_TEST(receive) {
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
),
sec
::
unavailable_or_would_block
);
CAF_MESSAGE
(
"adding both endpoint managers"
);
auto
mgr1
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
sockets
.
first
,
application_type
{
sys
,
buf
}});
auto
mgr1
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
sockets
.
first
,
application_type
{
buf
}});
CAF_CHECK_EQUAL
(
mgr1
->
init
(),
none
);
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
2u
);
auto
mgr2
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
sockets
.
second
,
application_type
{
sys
,
buf
}});
auto
mgr2
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
sockets
.
second
,
application_type
{
buf
}});
CAF_CHECK_EQUAL
(
mgr2
->
init
(),
none
);
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
3u
);
CAF_MESSAGE
(
"resolve actor-proxy"
);
...
...
libcaf_net/test/transport_worker.cpp
View file @
2e9fe41b
...
...
@@ -37,13 +37,11 @@ using namespace caf::net;
namespace
{
using
buffer_type
=
std
::
vector
<
byte
>
;
constexpr
string_view
hello_test
=
"hello test!"
;
struct
application_result
{
bool
initialized
;
std
::
vector
<
byte
>
data_buffer
;
byte_buffer
data_buffer
;
std
::
string
resolve_path
;
actor
resolve_listener
;
std
::
string
timeout_value
;
...
...
@@ -73,9 +71,13 @@ public:
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
msg
)
{
auto
header_buffer
=
parent
.
next_header_buffer
();
parent
.
write_packet
(
header_buffer
,
msg
->
payload
);
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
ptr
)
{
auto
payload_buf
=
parent
.
next_payload_buffer
();
binary_serializer
sink
(
parent
.
system
(),
payload_buf
);
if
(
auto
err
=
sink
(
ptr
->
msg
->
content
()))
CAF_FAIL
(
"serializing failed: "
<<
err
);
CAF_MESSAGE
(
"before sending: "
<<
CAF_ARG
(
ptr
->
msg
->
content
()));
parent
.
write_packet
(
payload_buf
);
}
template
<
class
Parent
>
...
...
@@ -102,15 +104,6 @@ public:
res_
->
err
=
err
;
}
static
expected
<
std
::
vector
<
byte
>>
serialize
(
actor_system
&
sys
,
const
message
&
x
)
{
std
::
vector
<
byte
>
result
;
binary_serializer
sink
{
sys
,
result
};
if
(
auto
err
=
x
.
save
(
sink
))
return
err
.
value
();
return
result
;
}
private:
std
::
shared_ptr
<
application_result
>
res_
;
};
...
...
@@ -121,12 +114,12 @@ public:
using
application_type
=
dummy_application
;
dummy_transport
(
std
::
shared_ptr
<
transport_result
>
res
)
:
res_
(
std
::
move
(
res
))
{
dummy_transport
(
actor_system
&
sys
,
std
::
shared_ptr
<
transport_result
>
res
)
:
sys_
(
sys
),
res_
(
std
::
move
(
res
))
{
// nop
}
void
write_packet
(
ip_endpoint
ep
,
span
<
b
uffer_type
*>
buffers
)
{
void
write_packet
(
ip_endpoint
ep
,
span
<
b
yte_buffer
*>
buffers
)
{
res_
->
ep
=
ep
;
auto
&
packet_buf
=
res_
->
packet_buffer
;
packet_buf
.
clear
();
...
...
@@ -134,19 +127,24 @@ public:
packet_buf
.
insert
(
packet_buf
.
end
(),
buf
->
begin
(),
buf
->
end
());
}
actor_system
&
system
()
{
return
sys_
;
}
transport_type
&
transport
()
{
return
*
this
;
}
std
::
vector
<
byte
>
next_header_buffer
()
{
byte_buffer
next_header_buffer
()
{
return
{};
}
std
::
vector
<
byte
>
next_payload_buffer
()
{
byte_buffer
next_payload_buffer
()
{
return
{};
}
private:
actor_system
&
sys_
;
std
::
shared_ptr
<
transport_result
>
res_
;
};
...
...
@@ -156,7 +154,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
fixture
()
:
transport_results
{
std
::
make_shared
<
transport_result
>
()},
application_results
{
std
::
make_shared
<
application_result
>
()},
transport
(
transport_results
),
transport
(
sys
,
transport_results
),
worker
{
dummy_application
{
application_results
}}
{
mpx
=
std
::
make_shared
<
multiplexer
>
();
if
(
auto
err
=
mpx
->
init
())
...
...
@@ -196,20 +194,23 @@ CAF_TEST(handle_data) {
}
CAF_TEST
(
write_message
)
{
std
::
string
hello_test
{
"hello world!"
};
actor
act
;
auto
strong_actor
=
actor_cast
<
strong_actor_ptr
>
(
act
);
mailbox_element
::
forwarding_stack
stack
;
auto
msg
=
make_message
();
auto
msg
=
make_message
(
hello_test
);
auto
elem
=
make_mailbox_element
(
strong_actor
,
make_message_id
(
12345
),
stack
,
msg
);
auto
test_span
=
as_bytes
(
make_span
(
hello_test
));
std
::
vector
<
byte
>
payload
(
test_span
.
begin
(),
test_span
.
end
());
using
message_type
=
endpoint_manager_queue
::
message
;
auto
message
=
detail
::
make_unique
<
message_type
>
(
std
::
move
(
elem
),
nullptr
,
payload
);
auto
message
=
detail
::
make_unique
<
message_type
>
(
std
::
move
(
elem
),
nullptr
);
worker
.
write_message
(
transport
,
std
::
move
(
message
));
auto
&
buf
=
transport_results
->
packet_buffer
;
string_view
result
{
reinterpret_cast
<
char
*>
(
buf
.
data
()),
buf
.
size
()};
binary_deserializer
source
{
sys
,
buf
};
caf
::
message
received_msg
;
CAF_CHECK
(
!
source
(
received_msg
));
CAF_MESSAGE
(
CAF_ARG
(
received_msg
));
auto
received_str
=
received_msg
.
get_as
<
std
::
string
>
(
0
);
string_view
result
{
received_str
};
CAF_CHECK_EQUAL
(
result
,
hello_test
);
CAF_CHECK_EQUAL
(
transport_results
->
ep
,
ep
);
}
...
...
libcaf_net/test/transport_worker_dispatcher.cpp
View file @
2e9fe41b
...
...
@@ -66,10 +66,10 @@ public:
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
msg
)
{
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
ptr
)
{
rec_buf_
->
push_back
(
static_cast
<
byte
>
(
id_
));
auto
header_buf
=
parent
.
next_header_buffer
(
);
parent
.
write_packet
(
header_buf
,
msg
->
payload
);
auto
data
=
ptr
->
msg
->
content
().
get_as
<
std
::
vector
<
byte
>>
(
0
);
parent
.
write_packet
(
data
);
}
template
<
class
Parent
>
...
...
@@ -92,10 +92,6 @@ public:
rec_buf_
->
push_back
(
static_cast
<
byte
>
(
id_
));
}
static
expected
<
buffer_type
>
serialize
(
actor_system
&
,
const
message
&
)
{
return
buffer_type
{};
}
private:
std
::
shared_ptr
<
buffer_type
>
rec_buf_
;
uint8_t
id_
;
...
...
@@ -126,7 +122,8 @@ struct dummy_transport {
using
application_type
=
dummy_application
;
dummy_transport
(
std
::
shared_ptr
<
buffer_type
>
buf
)
:
buf_
(
std
::
move
(
buf
))
{
dummy_transport
(
actor_system
&
sys
,
std
::
shared_ptr
<
buffer_type
>
buf
)
:
sys_
(
sys
),
buf_
(
std
::
move
(
buf
))
{
// nop
}
...
...
@@ -136,6 +133,10 @@ struct dummy_transport {
buf_
->
insert
(
buf_
->
end
(),
buf
->
begin
(),
buf
->
end
());
}
actor_system
&
system
()
{
return
sys_
;
}
transport_type
&
transport
()
{
return
*
this
;
}
...
...
@@ -149,6 +150,7 @@ struct dummy_transport {
}
private:
actor_system
&
sys_
;
std
::
shared_ptr
<
buffer_type
>
buf_
;
};
...
...
@@ -182,32 +184,31 @@ uri operator"" _u(const char* cstr, size_t cstr_len) {
}
struct
fixture
:
host_fixture
{
using
dispatcher_type
=
transport_worker_dispatcher
<
dummy_application_factory
,
ip_endpoint
>
;
using
dispatcher_type
=
transport_worker_dispatcher
<
dummy_application_factory
,
ip_endpoint
>
;
fixture
()
:
buf
{
std
::
make_shared
<
buffer_type
>
()},
dispatcher
{
dummy_application_factory
{
buf
}},
dummy
{
buf
}
{
dummy
{
sys
,
buf
}
{
add_new_workers
();
}
std
::
unique_ptr
<
net
::
endpoint_manager_queue
::
message
>
make_dummy_message
(
node_id
nid
)
{
actor_id
aid
=
42
;
auto
test_span
=
as_bytes
(
make_span
(
hello_test
));
byte_buffer
payload
(
test_span
.
begin
(),
test_span
.
end
());
actor_config
cfg
;
auto
p
=
make_actor
<
dummy_actor
,
strong_actor_ptr
>
(
aid
,
nid
,
&
sys
,
cfg
);
auto
test_span
=
as_bytes
(
make_span
(
hello_test
));
buffer_type
payload
(
test_span
.
begin
(),
test_span
.
end
());
auto
receiver
=
actor_cast
<
strong_actor_ptr
>
(
p
);
if
(
!
receiver
)
CAF_FAIL
(
"failed to cast receiver to a strong_actor_ptr"
);
mailbox_element
::
forwarding_stack
stack
;
auto
elem
=
make_mailbox_element
(
nullptr
,
make_message_id
(
12345
),
std
::
move
(
stack
),
make_message
());
std
::
move
(
stack
),
make_message
(
payload
));
return
detail
::
make_unique
<
endpoint_manager_queue
::
message
>
(
std
::
move
(
elem
),
receiver
,
payload
);
receiver
);
}
bool
contains
(
byte
x
)
{
...
...
@@ -227,6 +228,7 @@ struct fixture : host_fixture {
auto
msg
=
make_dummy_message
(
testcase
.
nid
);
if
(
!
msg
->
receiver
)
CAF_FAIL
(
"receiver is null"
);
CAF_MESSAGE
(
CAF_ARG
(
msg
));
dispatcher
.
write_message
(
dummy
,
std
::
move
(
msg
));
}
...
...
@@ -255,9 +257,8 @@ struct fixture : host_fixture {
test_write_message(testcase); \
CAF_CHECK_EQUAL(buf->size(), hello_test.size() + 1u); \
CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \
CAF_CHECK_EQUAL(memcmp(buf->data() + 1, hello_test.data(), \
hello_test.size()), \
0); \
CAF_CHECK_EQUAL( \
memcmp(buf->data() + 1, hello_test.data(), hello_test.size()), 0); \
buf->clear();
#define CHECK_TIMEOUT(testcase) \
...
...
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