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