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
37fc6104
Unverified
Commit
37fc6104
authored
Jun 13, 2020
by
Dominik Charousset
Committed by
GitHub
Jun 13, 2020
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #73
Address nodiscard warnings
parents
7e99ad85
bf07f2a1
Changes
15
Hide whitespace changes
Inline
Side-by-side
Showing
15 changed files
with
58 additions
and
36 deletions
+58
-36
CMakeLists.txt
CMakeLists.txt
+1
-1
libcaf_net/caf/net/datagram_transport.hpp
libcaf_net/caf/net/datagram_transport.hpp
+6
-1
libcaf_net/caf/net/stream_transport.hpp
libcaf_net/caf/net/stream_transport.hpp
+2
-1
libcaf_net/caf/net/transport_worker.hpp
libcaf_net/caf/net/transport_worker.hpp
+7
-5
libcaf_net/src/net/backend/test.cpp
libcaf_net/src/net/backend/test.cpp
+2
-1
libcaf_net/src/pollset_updater.cpp
libcaf_net/src/pollset_updater.cpp
+2
-1
libcaf_net/src/tcp_accept_socket.cpp
libcaf_net/src/tcp_accept_socket.cpp
+5
-3
libcaf_net/src/tcp_stream_socket.cpp
libcaf_net/src/tcp_stream_socket.cpp
+5
-3
libcaf_net/test/datagram_transport.cpp
libcaf_net/test/datagram_transport.cpp
+4
-3
libcaf_net/test/doorman.cpp
libcaf_net/test/doorman.cpp
+4
-2
libcaf_net/test/endpoint_manager.cpp
libcaf_net/test/endpoint_manager.cpp
+2
-2
libcaf_net/test/stream_transport.cpp
libcaf_net/test/stream_transport.cpp
+3
-2
libcaf_net/test/string_application.cpp
libcaf_net/test/string_application.cpp
+5
-4
libcaf_net/test/transport_worker.cpp
libcaf_net/test/transport_worker.cpp
+5
-4
libcaf_net/test/transport_worker_dispatcher.cpp
libcaf_net/test/transport_worker_dispatcher.cpp
+5
-3
No files found.
CMakeLists.txt
View file @
37fc6104
...
...
@@ -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
94e26b315
GIT_TAG
c5f6c835c
)
FetchContent_Populate
(
actor_framework
)
set
(
CAF_ENABLE_EXAMPLES OFF CACHE BOOL
""
FORCE
)
...
...
libcaf_net/caf/net/datagram_transport.hpp
View file @
37fc6104
...
...
@@ -22,6 +22,7 @@
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
...
...
@@ -85,7 +86,11 @@ public:
auto
&
[
num_bytes
,
ep
]
=
*
res
;
CAF_LOG_DEBUG
(
"received "
<<
num_bytes
<<
" bytes"
);
this
->
read_buf_
.
resize
(
num_bytes
);
this
->
next_layer_
.
handle_data
(
*
this
,
this
->
read_buf_
,
std
::
move
(
ep
));
if
(
auto
err
=
this
->
next_layer_
.
handle_data
(
*
this
,
this
->
read_buf_
,
std
::
move
(
ep
)))
{
CAF_LOG_ERROR
(
"handle_data failed: "
<<
err
);
return
false
;
}
prepare_next_read
();
}
else
{
auto
err
=
get
<
sec
>
(
ret
);
...
...
libcaf_net/caf/net/stream_transport.hpp
View file @
37fc6104
...
...
@@ -65,7 +65,8 @@ public:
max_
(
1024
),
rd_flag_
(
net
::
receive_policy_flag
::
exactly
)
{
CAF_ASSERT
(
handle
!=
invalid_socket
);
nodelay
(
handle
,
true
);
if
(
auto
err
=
nodelay
(
handle
,
true
))
CAF_LOG_ERROR
(
"nodelay failed: "
<<
err
);
}
// -- member functions -------------------------------------------------------
...
...
libcaf_net/caf/net/transport_worker.hpp
View file @
37fc6104
...
...
@@ -18,6 +18,7 @@
#pragma once
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_decorator.hpp"
...
...
@@ -75,7 +76,8 @@ public:
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
msg
)
{
auto
writer
=
make_packet_writer_decorator
(
*
this
,
parent
);
application_
.
write_message
(
writer
,
std
::
move
(
msg
));
if
(
auto
err
=
application_
.
write_message
(
writer
,
std
::
move
(
msg
)))
CAF_LOG_ERROR
(
"write_message failed: "
<<
err
);
}
template
<
class
Parent
>
...
...
@@ -91,8 +93,8 @@ public:
}
template
<
class
Parent
>
void
local_actor_down
(
Parent
&
parent
,
const
node_id
&
,
actor_id
id
,
error
reason
)
{
void
local_actor_down
(
Parent
&
parent
,
const
node_id
&
,
actor_id
id
,
error
reason
)
{
auto
writer
=
make_packet_writer_decorator
(
*
this
,
parent
);
application_
.
local_actor_down
(
writer
,
id
,
std
::
move
(
reason
));
}
...
...
@@ -113,7 +115,7 @@ private:
};
template
<
class
Application
,
class
IdType
=
unit_t
>
using
transport_worker_ptr
=
std
::
shared_ptr
<
transport_worker
<
Application
,
IdType
>>
;
using
transport_worker_ptr
=
std
::
shared_ptr
<
transport_worker
<
Application
,
IdType
>>
;
}
// namespace caf::net
libcaf_net/src/net/backend/test.cpp
View file @
37fc6104
...
...
@@ -77,7 +77,8 @@ void test::set_last_hop(node_id*) {
test
::
peer_entry
&
test
::
emplace
(
const
node_id
&
peer_id
,
stream_socket
first
,
stream_socket
second
)
{
using
transport_type
=
stream_transport
<
basp
::
application
>
;
nonblocking
(
second
,
true
);
if
(
auto
err
=
nonblocking
(
second
,
true
))
CAF_LOG_ERROR
(
"nonblocking failed: "
<<
err
);
auto
mpx
=
mm_
.
mpx
();
basp
::
application
app
{
proxies_
};
auto
mgr
=
make_endpoint_manager
(
mpx
,
mm_
.
system
(),
...
...
libcaf_net/src/pollset_updater.cpp
View file @
37fc6104
...
...
@@ -32,7 +32,8 @@ pollset_updater::pollset_updater(pipe_socket read_handle,
const
multiplexer_ptr
&
parent
)
:
super
(
read_handle
,
parent
),
buf_size_
(
0
)
{
mask_
=
operation
::
read
;
nonblocking
(
read_handle
,
true
);
if
(
auto
err
=
nonblocking
(
read_handle
,
true
))
CAF_LOG_ERROR
(
"nonblocking failed: "
<<
err
);
}
pollset_updater
::~
pollset_updater
()
{
...
...
libcaf_net/src/tcp_accept_socket.cpp
View file @
37fc6104
...
...
@@ -65,7 +65,8 @@ expected<tcp_accept_socket> new_tcp_acceptor_impl(uint16_t port,
tcp_accept_socket
sock
{
fd
};
// sguard closes the socket in case of exception
auto
sguard
=
make_socket_guard
(
tcp_accept_socket
{
fd
});
child_process_inherit
(
sock
,
false
);
if
(
auto
err
=
child_process_inherit
(
sock
,
false
))
return
err
;
if
(
reuse_addr
)
{
int
on
=
1
;
CAF_NET_SYSCALL
(
"setsockopt"
,
tmp1
,
!=
,
0
,
...
...
@@ -73,8 +74,9 @@ expected<tcp_accept_socket> new_tcp_acceptor_impl(uint16_t port,
reinterpret_cast
<
setsockopt_ptr
>
(
&
on
),
static_cast
<
socket_size_type
>
(
sizeof
(
on
))));
}
using
sockaddr_type
=
typename
std
::
conditional
<
Family
==
AF_INET
,
sockaddr_in
,
sockaddr_in6
>::
type
;
using
sockaddr_type
=
typename
std
::
conditional
<
Family
==
AF_INET
,
sockaddr_in
,
sockaddr_in6
>::
type
;
sockaddr_type
sa
;
memset
(
&
sa
,
0
,
sizeof
(
sockaddr_type
));
detail
::
family_of
(
sa
)
=
Family
;
...
...
libcaf_net/src/tcp_stream_socket.cpp
View file @
37fc6104
...
...
@@ -38,8 +38,9 @@ bool ip_connect(stream_socket fd, std::string host, uint16_t port) {
CAF_LOG_TRACE
(
"Family ="
<<
(
Family
==
AF_INET
?
"AF_INET"
:
"AF_INET6"
)
<<
CAF_ARG
(
fd
.
id
)
<<
CAF_ARG
(
host
)
<<
CAF_ARG
(
port
));
static_assert
(
Family
==
AF_INET
||
Family
==
AF_INET6
,
"invalid family"
);
using
sockaddr_type
=
typename
std
::
conditional
<
Family
==
AF_INET
,
sockaddr_in
,
sockaddr_in6
>::
type
;
using
sockaddr_type
=
typename
std
::
conditional
<
Family
==
AF_INET
,
sockaddr_in
,
sockaddr_in6
>::
type
;
sockaddr_type
sa
;
memset
(
&
sa
,
0
,
sizeof
(
sockaddr_type
));
inet_pton
(
Family
,
host
.
c_str
(),
&
detail
::
addr_of
(
sa
));
...
...
@@ -60,7 +61,8 @@ expected<tcp_stream_socket> make_connected_tcp_stream_socket(ip_endpoint node) {
#endif
CAF_NET_SYSCALL
(
"socket"
,
fd
,
==
,
-
1
,
::
socket
(
proto
,
socktype
,
0
));
tcp_stream_socket
sock
{
fd
};
child_process_inherit
(
sock
,
false
);
if
(
auto
err
=
child_process_inherit
(
sock
,
false
))
return
err
;
auto
sguard
=
make_socket_guard
(
sock
);
if
(
proto
==
AF_INET6
)
{
if
(
ip_connect
<
AF_INET6
>
(
sock
,
to_string
(
node
.
address
()),
node
.
port
()))
{
...
...
libcaf_net/test/datagram_transport.cpp
View file @
37fc6104
...
...
@@ -115,13 +115,14 @@ public:
}
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
msg
)
{
error
write_message
(
Parent
&
parent
,
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
);
return
none
;
}
template
<
class
Parent
>
...
...
@@ -221,7 +222,7 @@ CAF_TEST(resolve and proxy communication) {
auto
mgr_impl
=
mgr
.
downcast
<
endpoint_manager_impl
<
transport_type
>>
();
CAF_CHECK
(
mgr_impl
!=
nullptr
);
auto
&
transport
=
mgr_impl
->
transport
();
transport
.
add_new_worker
(
make_node_id
(
uri
),
ep
);
CAF_CHECK_EQUAL
(
transport
.
add_new_worker
(
make_node_id
(
uri
),
ep
),
none
);
run
();
mgr
->
resolve
(
uri
,
self
);
run
();
...
...
libcaf_net/test/doorman.cpp
View file @
37fc6104
...
...
@@ -21,6 +21,7 @@
#include "caf/net/doorman.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/error.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
...
...
@@ -66,13 +67,14 @@ public:
}
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
msg
)
{
error
write_message
(
Parent
&
parent
,
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
);
return
none
;
}
template
<
class
Parent
>
...
...
libcaf_net/test/endpoint_manager.cpp
View file @
37fc6104
...
...
@@ -163,7 +163,7 @@ CAF_TEST(send and receive) {
byte_buffer
read_buf
(
1024
);
auto
buf
=
std
::
make_shared
<
byte_buffer
>
();
auto
sockets
=
unbox
(
make_stream_socket_pair
());
nonblocking
(
sockets
.
second
,
tru
e
);
CAF_CHECK_EQUAL
(
nonblocking
(
sockets
.
second
,
true
),
non
e
);
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
),
sec
::
unavailable_or_would_block
);
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
close
(
sockets
.
second
);
});
...
...
@@ -189,7 +189,7 @@ CAF_TEST(resolve and proxy communication) {
byte_buffer
read_buf
(
1024
);
auto
buf
=
std
::
make_shared
<
byte_buffer
>
();
auto
sockets
=
unbox
(
make_stream_socket_pair
());
nonblocking
(
sockets
.
second
,
tru
e
);
CAF_CHECK_EQUAL
(
nonblocking
(
sockets
.
second
,
true
),
non
e
);
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
close
(
sockets
.
second
);
});
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
dummy_transport
{
sockets
.
first
,
buf
});
...
...
libcaf_net/test/stream_transport.cpp
View file @
37fc6104
...
...
@@ -88,13 +88,14 @@ public:
}
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
msg
)
{
error
write_message
(
Parent
&
parent
,
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
);
return
none
;
}
template
<
class
Parent
>
...
...
libcaf_net/test/string_application.cpp
View file @
37fc6104
...
...
@@ -97,11 +97,11 @@ public:
}
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
ptr
)
{
error
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
ptr
)
{
// Ignore proxy announcement messages.
if
(
ptr
->
msg
==
nullptr
)
return
;
return
none
;
auto
header_buf
=
parent
.
next_header_buffer
();
auto
payload_buf
=
parent
.
next_payload_buffer
();
binary_serializer
payload_sink
{
parent
.
system
(),
payload_buf
};
...
...
@@ -112,6 +112,7 @@ public:
=
header_sink
(
header_type
{
static_cast
<
uint32_t
>
(
payload_buf
.
size
())}))
CAF_FAIL
(
"serializing failed: "
<<
err
);
parent
.
write_packet
(
header_buf
,
payload_buf
);
return
none
;
}
private:
...
...
@@ -206,7 +207,7 @@ CAF_TEST(receive) {
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
auto
buf
=
std
::
make_shared
<
byte_buffer
>
();
auto
sockets
=
unbox
(
make_stream_socket_pair
());
nonblocking
(
sockets
.
second
,
tru
e
);
CAF_CHECK_EQUAL
(
nonblocking
(
sockets
.
second
,
true
),
non
e
);
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
),
sec
::
unavailable_or_would_block
);
CAF_MESSAGE
(
"adding both endpoint managers"
);
...
...
libcaf_net/test/transport_worker.cpp
View file @
37fc6104
...
...
@@ -71,14 +71,15 @@ public:
}
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
ptr
)
{
error
write_message
(
Parent
&
parent
,
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
);
return
none
;
}
template
<
class
Parent
>
...
...
@@ -188,7 +189,7 @@ CAF_TEST(construction and initialization) {
CAF_TEST
(
handle_data
)
{
auto
test_span
=
as_bytes
(
make_span
(
hello_test
));
worker
.
handle_data
(
transport
,
test_span
);
CAF_CHECK_EQUAL
(
worker
.
handle_data
(
transport
,
test_span
),
none
);
auto
&
buf
=
application_results
->
data_buffer
;
string_view
result
{
reinterpret_cast
<
char
*>
(
buf
.
data
()),
buf
.
size
()};
CAF_CHECK_EQUAL
(
result
,
hello_test
);
...
...
@@ -208,7 +209,7 @@ CAF_TEST(write_message) {
auto
&
buf
=
transport_results
->
packet_buffer
;
binary_deserializer
source
{
sys
,
buf
};
caf
::
message
received_msg
;
CAF_CHECK
(
!
source
(
received_msg
)
);
CAF_CHECK
_EQUAL
(
source
(
received_msg
),
none
);
CAF_MESSAGE
(
CAF_ARG
(
received_msg
));
auto
received_str
=
received_msg
.
get_as
<
std
::
string
>
(
0
);
string_view
result
{
received_str
};
...
...
libcaf_net/test/transport_worker_dispatcher.cpp
View file @
37fc6104
...
...
@@ -66,11 +66,12 @@ public:
}
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
ptr
)
{
error
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
ptr
)
{
rec_buf_
->
push_back
(
static_cast
<
byte
>
(
id_
));
auto
data
=
ptr
->
msg
->
content
().
get_as
<
byte_buffer
>
(
0
);
parent
.
write_packet
(
data
);
return
none
;
}
template
<
class
Parent
>
...
...
@@ -249,7 +250,8 @@ struct fixture : host_fixture {
};
#define CHECK_HANDLE_DATA(testcase) \
dispatcher.handle_data(dummy, span<const byte>{}, testcase.ep); \
CAF_CHECK_EQUAL( \
dispatcher.handle_data(dummy, span<const byte>{}, testcase.ep), none); \
CAF_CHECK_EQUAL(buf->size(), 1u); \
CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \
buf->clear();
...
...
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