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
319ef793
Commit
319ef793
authored
Jul 31, 2018
by
Joseph Noir
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Fix UDP newb send and receive
parent
97849853
Changes
1
Hide whitespace changes
Inline
Side-by-side
Showing
1 changed file
with
616 additions
and
10 deletions
+616
-10
libcaf_io/test/protocol_policy.cpp
libcaf_io/test/protocol_policy.cpp
+616
-10
No files found.
libcaf_io/test/protocol_policy.cpp
View file @
319ef793
...
@@ -615,7 +615,7 @@ struct newb_acceptor : public network::event_handler {
...
@@ -615,7 +615,7 @@ struct newb_acceptor : public network::event_handler {
CAF_MESSAGE
(
"read event on newb acceptor"
);
CAF_MESSAGE
(
"read event on newb acceptor"
);
native_socket
sock
;
native_socket
sock
;
transport_policy_ptr
transport
;
transport_policy_ptr
transport
;
std
::
tie
(
sock
,
transport
)
=
acceptor
->
accept
(
this
);
;
std
::
tie
(
sock
,
transport
)
=
acceptor
->
accept
(
this
);
auto
en
=
create_newb
(
sock
,
std
::
move
(
transport
));
auto
en
=
create_newb
(
sock
,
std
::
move
(
transport
));
if
(
!
en
)
if
(
!
en
)
return
std
::
move
(
en
.
error
());
return
std
::
move
(
en
.
error
());
...
@@ -850,6 +850,38 @@ expected<native_socket> new_tcp_acceptor_impl(uint16_t port, const char* addr,
...
@@ -850,6 +850,38 @@ expected<native_socket> new_tcp_acceptor_impl(uint16_t port, const char* addr,
return
sguard
.
release
();
return
sguard
.
release
();
}
}
expected
<
native_socket
>
new_udp_endpoint_impl
(
uint16_t
port
,
const
char
*
addr
,
bool
reuse
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
port
)
<<
", addr = "
<<
(
addr
?
addr
:
"nullptr"
));
auto
addrs
=
network
::
interfaces
::
server_address
(
port
,
addr
);
auto
addr_str
=
std
::
string
{
addr
==
nullptr
?
""
:
addr
};
if
(
addrs
.
empty
())
return
make_error
(
sec
::
cannot_open_port
,
"No local interface available"
,
addr_str
);
bool
any
=
addr_str
.
empty
()
||
addr_str
==
"::"
||
addr_str
==
"0.0.0.0"
;
auto
fd
=
invalid_native_socket
;
for
(
auto
&
elem
:
addrs
)
{
auto
host
=
elem
.
first
.
c_str
();
auto
p
=
elem
.
second
==
ipv4
?
new_ip_acceptor_impl
<
AF_INET
,
SOCK_DGRAM
>
(
port
,
host
,
reuse
,
any
)
:
new_ip_acceptor_impl
<
AF_INET6
,
SOCK_DGRAM
>
(
port
,
host
,
reuse
,
any
);
if
(
!
p
)
{
CAF_LOG_DEBUG
(
p
.
error
());
continue
;
}
fd
=
*
p
;
break
;
}
if
(
fd
==
invalid_native_socket
)
{
CAF_LOG_WARNING
(
"could not open udp socket on:"
<<
CAF_ARG
(
port
)
<<
CAF_ARG
(
addr_str
));
return
make_error
(
sec
::
cannot_open_port
,
"udp socket creation failed"
,
port
,
addr_str
);
}
CAF_LOG_DEBUG
(
CAF_ARG
(
fd
));
return
fd
;
}
// -- create newbs -------------------------------------------------------------
// -- create newbs -------------------------------------------------------------
template
<
class
Newb
>
template
<
class
Newb
>
...
@@ -866,7 +898,8 @@ std::unique_ptr<NewbAcceptor> make_newb_acceptor(actor_system& sys,
...
@@ -866,7 +898,8 @@ std::unique_ptr<NewbAcceptor> make_newb_acceptor(actor_system& sys,
uint16_t
port
,
uint16_t
port
,
const
char
*
addr
=
nullptr
,
const
char
*
addr
=
nullptr
,
bool
reuse_addr
=
false
)
{
bool
reuse_addr
=
false
)
{
auto
sockfd
=
new_tcp_acceptor_impl
(
port
,
addr
,
reuse_addr
);
auto
sockfd
=
NewbAcceptor
::
create_socket
(
port
,
addr
,
reuse_addr
);
// new_tcp_acceptor_impl(port, addr, reuse_addr);
if
(
!
sockfd
)
{
if
(
!
sockfd
)
{
CAF_LOG_DEBUG
(
"Could not open "
<<
CAF_ARG
(
port
)
<<
CAF_ARG
(
addr
));
CAF_LOG_DEBUG
(
"Could not open "
<<
CAF_ARG
(
port
)
<<
CAF_ARG
(
addr
));
return
nullptr
;
return
nullptr
;
...
@@ -1195,6 +1228,12 @@ struct tcp_basp_acceptor
...
@@ -1195,6 +1228,12 @@ struct tcp_basp_acceptor
:
public
newb_acceptor
<
typename
ProtocolPolicy
::
message_type
>
{
:
public
newb_acceptor
<
typename
ProtocolPolicy
::
message_type
>
{
using
super
=
newb_acceptor
<
typename
ProtocolPolicy
::
message_type
>
;
using
super
=
newb_acceptor
<
typename
ProtocolPolicy
::
message_type
>
;
static
expected
<
native_socket
>
create_socket
(
uint16_t
port
,
const
char
*
host
,
bool
reuse
=
false
)
{
return
new_tcp_acceptor_impl
(
port
,
host
,
reuse
);
}
tcp_basp_acceptor
(
default_multiplexer
&
dm
,
native_socket
sockfd
)
tcp_basp_acceptor
(
default_multiplexer
&
dm
,
native_socket
sockfd
)
:
super
(
dm
,
sockfd
)
{
:
super
(
dm
,
sockfd
)
{
// nop
// nop
...
@@ -1220,6 +1259,457 @@ struct tcp_basp_acceptor
...
@@ -1220,6 +1259,457 @@ struct tcp_basp_acceptor
actor
responder
;
actor
responder
;
};
};
// -- udp impls ----------------------------------------------------------------
struct
udp_basp_header
{
uint32_t
payload_len
;
actor_id
from
;
actor_id
to
;
};
template
<
class
Inspector
>
typename
Inspector
::
result_type
inspect
(
Inspector
&
fun
,
udp_basp_header
&
hdr
)
{
return
fun
(
meta
::
type_name
(
"udp_basp_header"
),
hdr
.
payload_len
,
hdr
.
from
,
hdr
.
to
);
}
constexpr
size_t
udp_basp_header_len
=
sizeof
(
uint32_t
)
+
sizeof
(
actor_id
)
*
2
;
using
sequence_type
=
uint16_t
;
struct
udp_ordering_header
{
sequence_type
seq
;
};
template
<
class
Inspector
>
typename
Inspector
::
result_type
inspect
(
Inspector
&
fun
,
udp_ordering_header
&
hdr
)
{
return
fun
(
meta
::
type_name
(
"udp_basp_header"
),
hdr
.
seq
);
}
constexpr
size_t
udp_ordering_header_len
=
sizeof
(
sequence_type
);
struct
new_udp_basp_message
{
udp_basp_header
header
;
char
*
payload
;
size_t
payload_len
;
};
template
<
class
Inspector
>
typename
Inspector
::
result_type
inspect
(
Inspector
&
fun
,
new_udp_basp_message
&
msg
)
{
return
fun
(
meta
::
type_name
(
"new_udp_basp_message"
),
msg
.
header
,
msg
.
payload_len
);
}
struct
udp_basp
{
static
constexpr
size_t
header_size
=
udp_basp_header_len
;
using
message_type
=
new_udp_basp_message
;
using
result_type
=
optional
<
message_type
>
;
newb
<
message_type
>*
parent
;
message_type
msg
;
udp_basp
(
newb
<
message_type
>*
parent
)
:
parent
(
parent
)
{
// nop
}
error
read
(
char
*
bytes
,
size_t
count
)
{
CAF_MESSAGE
(
"reading basp udp header"
);
// Read header.
if
(
count
<
udp_basp_header_len
)
{
CAF_MESSAGE
(
"not enought bytes for basp header"
);
CAF_LOG_DEBUG
(
"buffer contains "
<<
count
<<
" bytes of expected "
<<
udp_basp_header_len
);
return
sec
::
unexpected_message
;
}
binary_deserializer
bd
{
&
parent
->
backend
(),
bytes
,
count
};
bd
(
msg
.
header
);
CAF_MESSAGE
(
"read basp header "
<<
CAF_ARG
(
msg
.
header
));
size_t
payload_len
=
static_cast
<
size_t
>
(
msg
.
header
.
payload_len
);
// Read payload.
auto
remaining
=
count
-
udp_basp_header_len
;
// TODO: Could be `!=` ?
if
(
remaining
<
payload_len
)
{
CAF_LOG_DEBUG
(
"only "
<<
remaining
<<
" bytes remaining of expected "
<<
msg
.
header
.
payload_len
);
return
sec
::
unexpected_message
;
}
msg
.
payload
=
bytes
+
udp_basp_header_len
;
msg
.
payload_len
=
msg
.
header
.
payload_len
;
parent
->
handle
(
msg
);
return
none
;
}
error
timeout
(
atom_value
,
uint32_t
)
{
return
none
;
}
size_t
write_header
(
byte_buffer
&
buf
,
header_writer
*
hw
)
{
CAF_ASSERT
(
hw
!=
nullptr
);
(
*
hw
)(
buf
);
return
header_size
;
}
void
prepare_for_sending
(
byte_buffer
&
buf
,
size_t
hstart
,
size_t
plen
)
{
stream_serializer
<
charbuf
>
out
{
&
parent
->
backend
(),
buf
.
data
()
+
hstart
,
sizeof
(
uint32_t
)};
auto
len
=
static_cast
<
uint32_t
>
(
plen
);
out
(
len
);
}
};
template
<
class
Next
>
struct
udp_ordering
{
static
constexpr
size_t
header_size
=
udp_ordering_header_len
;
using
message_type
=
typename
Next
::
message_type
;
using
result_type
=
typename
Next
::
result_type
;
sequence_type
seq_read
=
0
;
sequence_type
seq_write
=
0
;
size_t
max_pending_messages
=
10
;
std
::
chrono
::
milliseconds
pending_to
=
std
::
chrono
::
milliseconds
(
100
);
newb
<
message_type
>*
parent
;
Next
next
;
std
::
unordered_map
<
sequence_type
,
std
::
vector
<
char
>>
pending
;
udp_ordering
(
newb
<
message_type
>*
parent
)
:
parent
(
parent
),
next
(
parent
)
{
// nop
}
error
deliver_pending
()
{
if
(
pending
.
empty
())
return
none
;
while
(
pending
.
count
(
seq_read
)
>
0
)
{
auto
&
buf
=
pending
[
seq_read
];
auto
res
=
next
.
read
(
buf
.
data
(),
buf
.
size
());
pending
.
erase
(
seq_read
);
// TODO: Cancel timeout.
if
(
res
)
return
res
;
}
return
none
;
}
error
add_pending
(
char
*
bytes
,
size_t
count
,
sequence_type
seq
)
{
pending
[
seq
]
=
std
::
vector
<
char
>
(
bytes
+
header_size
,
bytes
+
count
);
parent
->
set_timeout
(
pending_to
,
ordering_atom
::
value
,
seq
);
if
(
pending
.
size
()
>
max_pending_messages
)
{
seq_read
=
pending
.
begin
()
->
first
;
return
deliver_pending
();
}
return
none
;
}
error
read
(
char
*
bytes
,
size_t
count
)
{
if
(
count
<
header_size
)
return
sec
::
unexpected_message
;
udp_ordering_header
hdr
;
binary_deserializer
bd
(
&
parent
->
backend
(),
bytes
,
count
);
bd
(
hdr
);
CAF_MESSAGE
(
"read udp ordering header: "
<<
to_string
(
hdr
));
// TODO: Use the comparison function from BASP instance.
if
(
hdr
.
seq
==
seq_read
)
{
CAF_MESSAGE
(
"it's the expected sequence number"
);
seq_read
+=
1
;
auto
res
=
next
.
read
(
bytes
+
header_size
,
count
-
header_size
);
if
(
res
)
return
res
;
return
deliver_pending
();
}
else
if
(
hdr
.
seq
>
seq_read
)
{
add_pending
(
bytes
,
count
,
hdr
.
seq
);
return
none
;
}
return
none
;
}
error
timeout
(
atom_value
atm
,
sequence_type
id
)
{
if
(
atm
==
ordering_atom
::
value
)
{
error
err
=
none
;
if
(
pending
.
count
(
id
)
>
0
)
{
seq_read
=
static_cast
<
sequence_type
>
(
id
);
err
=
deliver_pending
();
}
return
err
;
}
return
next
.
timeout
(
atm
,
id
);
}
void
write_header
(
byte_buffer
&
buf
,
header_writer
*
hw
)
{
binary_serializer
bs
(
&
parent
->
backend
(),
buf
);
bs
(
udp_ordering_header
{
seq_write
});
seq_write
+=
1
;
next
.
write_header
(
buf
,
hw
);
return
;
}
void
prepare_for_sending
(
byte_buffer
&
buf
,
size_t
hstart
,
size_t
plen
)
{
next
.
prepare_for_sending
(
buf
,
hstart
,
plen
);
}
};
struct
udp_transport_policy
:
public
transport_policy
{
udp_transport_policy
()
:
maximum
{
std
::
numeric_limits
<
uint16_t
>::
max
()},
first_message
{
true
},
writing
{
false
},
written
{
0
},
offline_sum
{
0
}
{
// nop
}
error
read_some
(
network
::
event_handler
*
parent
)
override
{
CAF_MESSAGE
(
"read some from sock "
<<
parent
->
fd
());
CAF_LOG_TRACE
(
CAF_ARG
(
parent
->
fd
()));
memset
(
sender
.
address
(),
0
,
sizeof
(
sockaddr_storage
));
network
::
socket_size_type
len
=
sizeof
(
sockaddr_storage
);
auto
buf_ptr
=
static_cast
<
network
::
socket_recv_ptr
>
(
receive_buffer
.
data
());
auto
buf_len
=
receive_buffer
.
size
();
CAF_MESSAGE
(
"receive buffer size: "
<<
buf_len
);
auto
sres
=
::
recvfrom
(
parent
->
fd
(),
buf_ptr
,
buf_len
,
0
,
sender
.
address
(),
&
len
);
if
(
network
::
is_error
(
sres
,
true
))
{
CAF_LOG_ERROR
(
"recvfrom returned"
<<
CAF_ARG
(
sres
));
return
sec
::
runtime_error
;
}
else
if
(
network
::
would_block_or_temporarily_unavailable
(
network
::
last_socket_error
()))
{
CAF_LOG_DEBUG
(
"try later"
);
return
sec
::
end_of_stream
;
}
CAF_MESSAGE
(
"read "
<<
sres
<<
" bytes"
);
if
(
sres
==
0
)
CAF_LOG_INFO
(
"Received empty datagram"
);
else
if
(
sres
>
static_cast
<
network
::
signed_size_type
>
(
buf_len
))
CAF_LOG_WARNING
(
"recvfrom cut of message, only received "
<<
CAF_ARG
(
buf_len
)
<<
" of "
<<
CAF_ARG
(
sres
)
<<
" bytes"
);
receive_buffer_length
=
(
sres
>
0
)
?
static_cast
<
size_t
>
(
sres
)
:
0
;
*
sender
.
length
()
=
static_cast
<
size_t
>
(
len
);
if
(
first_message
)
{
endpoint
=
sender
;
first_message
=
false
;
}
return
none
;
}
bool
should_deliver
()
override
{
CAF_LOG_TRACE
(
""
);
return
receive_buffer_length
!=
0
&&
sender
==
endpoint
;
}
void
prepare_next_read
(
network
::
event_handler
*
)
override
{
receive_buffer_length
=
0
;
receive_buffer
.
resize
(
maximum
);
}
void
configure_read
(
receive_policy
::
config
)
override
{
// nop
}
error
write_some
(
network
::
event_handler
*
parent
)
override
{
using
namespace
caf
::
io
::
network
;
CAF_LOG_TRACE
(
CAF_ARG
(
parent
->
fd
())
<<
CAF_ARG
(
send_buffer
.
size
()));
socket_size_type
len
=
static_cast
<
socket_size_type
>
(
*
endpoint
.
clength
());
auto
buf_ptr
=
reinterpret_cast
<
socket_send_ptr
>
(
send_buffer
.
data
()
+
written
);
auto
buf_len
=
send_sizes
.
front
();
auto
sres
=
::
sendto
(
parent
->
fd
(),
buf_ptr
,
buf_len
,
0
,
endpoint
.
caddress
(),
len
);
if
(
is_error
(
sres
,
true
))
{
CAF_LOG_ERROR
(
"sendto returned"
<<
CAF_ARG
(
sres
));
return
sec
::
runtime_error
;
}
send_sizes
.
pop_front
();
written
+=
(
sres
>
0
)
?
static_cast
<
size_t
>
(
sres
)
:
0
;
auto
remaining
=
send_buffer
.
size
()
-
written
;
if
(
remaining
==
0
)
prepare_next_write
(
parent
);
return
none
;
}
void
prepare_next_write
(
network
::
event_handler
*
parent
)
override
{
written
=
0
;
send_buffer
.
clear
();
send_sizes
.
clear
();
if
(
offline_buffer
.
empty
())
{
writing
=
false
;
parent
->
backend
().
del
(
network
::
operation
::
write
,
parent
->
fd
(),
parent
);
}
else
{
// Add size of last chunk.
offline_sizes
.
push_back
(
offline_buffer
.
size
()
-
offline_sum
);
// Switch buffers.
send_buffer
.
swap
(
offline_buffer
);
send_sizes
.
swap
(
offline_sizes
);
// Reset sum.
offline_sum
=
0
;
}
}
byte_buffer
&
wr_buf
()
{
if
(
!
offline_buffer
.
empty
())
{
auto
chunk_size
=
offline_buffer
.
size
()
-
offline_sum
;
offline_sizes
.
push_back
(
chunk_size
);
offline_sum
+=
chunk_size
;
}
return
offline_buffer
;
}
void
flush
(
network
::
event_handler
*
parent
)
override
{
CAF_ASSERT
(
parent
!=
nullptr
);
CAF_LOG_TRACE
(
CAF_ARG
(
offline_buffer
.
size
()));
if
(
!
offline_buffer
.
empty
()
&&
!
writing
)
{
parent
->
backend
().
add
(
network
::
operation
::
write
,
parent
->
fd
(),
parent
);
writing
=
true
;
prepare_next_write
(
parent
);
}
}
// State for reading.
size_t
maximum
;
bool
first_message
;
// State for writing.
bool
writing
;
size_t
written
;
size_t
offline_sum
;
std
::
deque
<
size_t
>
send_sizes
;
std
::
deque
<
size_t
>
offline_sizes
;
// UDP endpoints.
network
::
ip_endpoint
endpoint
;
network
::
ip_endpoint
sender
;
};
template
<
class
T
>
struct
udp_protocol_policy
:
protocol_policy
<
typename
T
::
message_type
>
{
T
impl
;
udp_protocol_policy
(
newb
<
typename
T
::
message_type
>*
parent
)
:
impl
(
parent
)
{
// nop
}
error
read
(
char
*
bytes
,
size_t
count
)
override
{
return
impl
.
read
(
bytes
,
count
);
}
error
timeout
(
atom_value
atm
,
uint32_t
id
)
override
{
return
impl
.
timeout
(
atm
,
id
);
}
void
write_header
(
byte_buffer
&
buf
,
header_writer
*
hw
)
override
{
impl
.
write_header
(
buf
,
hw
);
}
void
prepare_for_sending
(
byte_buffer
&
buf
,
size_t
hstart
,
size_t
plen
)
override
{
impl
.
prepare_for_sending
(
buf
,
hstart
,
plen
);
}
};
struct
udp_basp_newb
:
newb
<
new_udp_basp_message
>
{
udp_basp_newb
(
caf
::
actor_config
&
cfg
,
default_multiplexer
&
dm
,
native_socket
sockfd
)
:
newb
<
new_udp_basp_message
>
(
cfg
,
dm
,
sockfd
)
{
// nop
}
void
handle
(
new_udp_basp_message
&
msg
)
override
{
CAF_PUSH_AID_FROM_PTR
(
this
);
CAF_LOG_TRACE
(
""
);
std
::
string
res
;
binary_deserializer
bd
(
&
backend
(),
msg
.
payload
,
msg
.
payload_len
);
bd
(
res
);
send
(
responder
,
res
);
}
behavior
make_behavior
()
override
{
set_default_handler
(
print_and_drop
);
return
{
// Must be implemented at the moment, will be cought by the broker in a
// later implementation.
[
=
](
atom_value
atm
,
uint32_t
id
)
{
protocol
->
timeout
(
atm
,
id
);
},
[
=
](
send_atom
,
actor_id
sender
,
actor_id
receiver
,
std
::
string
payload
)
{
auto
hw
=
caf
::
make_callback
([
&
](
byte_buffer
&
buf
)
->
error
{
binary_serializer
bs
(
&
backend
(),
buf
);
bs
(
udp_basp_header
{
0
,
sender
,
receiver
});
return
none
;
});
//{
// TODO: Need a better idea how to do this ... Maybe pass the write
// handle to flush which then calls `perpare_for_sending`?
auto
whdl
=
wr_buf
(
&
hw
);
CAF_CHECK
(
whdl
.
buf
!=
nullptr
);
CAF_CHECK
(
whdl
.
protocol
!=
nullptr
);
binary_serializer
bs
(
&
backend
(),
*
whdl
.
buf
);
bs
(
payload
);
//}
//flush();
},
[
=
](
quit_atom
)
{
CAF_MESSAGE
(
"newb actor shutting down"
);
// Remove from multiplexer loop.
stop
();
// Quit actor.
quit
();
}
};
}
actor
responder
;
};
struct
udp_accept_policy
:
public
accept_policy
{
virtual
std
::
pair
<
native_socket
,
transport_policy_ptr
>
accept
(
network
::
event_handler
*
)
{
auto
esock
=
new_udp_endpoint_impl
(
0
,
nullptr
,
false
);
if
(
!
esock
)
return
{
invalid_native_socket
,
nullptr
};
auto
result
=
std
::
move
(
*
esock
);
transport_policy_ptr
ptr
{
new
udp_transport_policy
};
return
{
result
,
std
::
move
(
ptr
)};
}
virtual
void
init
(
network
::
event_handler
&
)
{
}
};
template
<
class
ProtocolPolicy
>
struct
udp_basp_acceptor
:
public
newb_acceptor
<
typename
ProtocolPolicy
::
message_type
>
{
using
super
=
newb_acceptor
<
typename
ProtocolPolicy
::
message_type
>
;
udp_basp_acceptor
(
default_multiplexer
&
dm
,
native_socket
sockfd
)
:
super
(
dm
,
sockfd
)
{
// nop
CAF_MESSAGE
(
"Created newb acceptor with socket "
<<
sockfd
);
}
static
expected
<
native_socket
>
create_socket
(
uint16_t
port
,
const
char
*
host
,
bool
reuse
=
false
)
{
return
new_udp_endpoint_impl
(
port
,
host
,
reuse
);
}
expected
<
actor
>
create_newb
(
native_socket
sockfd
,
transport_policy_ptr
pol
)
override
{
CAF_MESSAGE
(
"creating new basp udp newb"
);
auto
n
=
make_newb
<
udp_basp_newb
>
(
this
->
backend
().
system
(),
sockfd
);
auto
ptr
=
caf
::
actor_cast
<
caf
::
abstract_actor
*>
(
n
);
if
(
ptr
==
nullptr
)
return
sec
::
runtime_error
;
auto
&
ref
=
dynamic_cast
<
udp_basp_newb
&>
(
*
ptr
);
ref
.
transport
=
std
::
move
(
pol
);
ref
.
protocol
.
reset
(
new
ProtocolPolicy
(
&
ref
));
ref
.
responder
=
responder
;
// Read first message from this socket.
CAF_MESSAGE
(
"Calling read_some with sock "
<<
this
->
fd
()
<<
" on new udp newb"
);
ref
.
transport
->
prepare_next_read
(
this
);
ref
.
transport
->
read_some
(
this
,
*
ref
.
protocol
.
get
());
// Subsequent messages will be read from `sockfd`.
anon_send
(
responder
,
n
);
return
n
;
}
actor
responder
;
};
// -- test classes -------------------------------------------------------------
// -- test classes -------------------------------------------------------------
struct
dummy_basp_newb
:
newb
<
new_basp_message
>
{
struct
dummy_basp_newb
:
newb
<
new_basp_message
>
{
...
@@ -1359,16 +1849,27 @@ public:
...
@@ -1359,16 +1849,27 @@ public:
}
}
};
};
struct
test_broker_state
{
struct
t
cp_t
est_broker_state
{
tcp_basp_header
hdr
;
tcp_basp_header
hdr
;
bool
expecting_header
=
true
;
bool
expecting_header
=
true
;
};
};
struct
udp_test_broker_state
{
datagram_handle
hdl
;
};
struct
fixture
{
struct
fixture
{
using
protocol_policy_t
=
tcp_protocol_policy
<
tcp_basp
>
;
// TCP testing.
using
accept_policy_t
=
tcp_accept_policy
;
using
tcp_protocol_policy_t
=
tcp_protocol_policy
<
tcp_basp
>
;
using
newb_acceptor_t
=
tcp_basp_acceptor
<
protocol_policy_t
>
;
using
tcp_accept_policy_t
=
tcp_accept_policy
;
using
transport_policy_t
=
tcp_transport_policy
;
using
tcp_newb_acceptor_t
=
tcp_basp_acceptor
<
tcp_protocol_policy_t
>
;
using
tcp_transport_policy_t
=
tcp_transport_policy
;
// UDP testing.
using
udp_protocol_policy_t
=
udp_protocol_policy
<
udp_ordering
<
udp_basp
>>
;
using
udp_accept_policy_t
=
udp_accept_policy
;
using
udp_newb_acceptor_t
=
udp_basp_acceptor
<
udp_protocol_policy_t
>
;
using
udp_transport_policy_t
=
udp_transport_policy
;
io_config
cfg
;
io_config
cfg
;
actor_system
sys
;
actor_system
sys
;
...
@@ -1612,7 +2113,7 @@ CAF_TEST_FIXTURE_SCOPE(tcp_newbs, fixture)
...
@@ -1612,7 +2113,7 @@ CAF_TEST_FIXTURE_SCOPE(tcp_newbs, fixture)
CAF_TEST
(
tcp
basp
newb
)
{
CAF_TEST
(
tcp
basp
newb
)
{
scoped_actor
main_actor
{
sys
};
scoped_actor
main_actor
{
sys
};
actor
newb_actor
;
actor
newb_actor
;
auto
testing
=
[
&
](
stateful_broker
<
test_broker_state
>*
self
,
auto
testing
=
[
&
](
stateful_broker
<
t
cp_t
est_broker_state
>*
self
,
connection_handle
hdl
,
actor
m
)
->
behavior
{
connection_handle
hdl
,
actor
m
)
->
behavior
{
CAF_CHECK
(
hdl
!=
invalid_connection_handle
);
CAF_CHECK
(
hdl
!=
invalid_connection_handle
);
self
->
configure_read
(
hdl
,
receive_policy
::
exactly
(
tcp_basp_header_len
));
self
->
configure_read
(
hdl
,
receive_policy
::
exactly
(
tcp_basp_header_len
));
...
@@ -1676,8 +2177,8 @@ CAF_TEST(tcp basp newb) {
...
@@ -1676,8 +2177,8 @@ CAF_TEST(tcp basp newb) {
},
main_actor
);
},
main_actor
);
CAF_MESSAGE
(
"creating new acceptor"
);
CAF_MESSAGE
(
"creating new acceptor"
);
auto
newb_acceptor_ptr
auto
newb_acceptor_ptr
=
make_newb_acceptor
<
newb_acceptor_t
,
accept_policy_t
>
(
sys
,
port
);
=
make_newb_acceptor
<
tcp_newb_acceptor_t
,
tcp_
accept_policy_t
>
(
sys
,
port
);
dynamic_cast
<
newb_acceptor_t
*>
(
newb_acceptor_ptr
.
get
())
->
responder
dynamic_cast
<
tcp_
newb_acceptor_t
*>
(
newb_acceptor_ptr
.
get
())
->
responder
=
helper_actor
;
=
helper_actor
;
CAF_MESSAGE
(
"connecting from 'old-style' broker"
);
CAF_MESSAGE
(
"connecting from 'old-style' broker"
);
auto
exp
=
sys
.
middleman
().
spawn_client
(
testing
,
host
,
port
,
main_actor
);
auto
exp
=
sys
.
middleman
().
spawn_client
(
testing
,
host
,
port
,
main_actor
);
...
@@ -1713,4 +2214,109 @@ CAF_TEST(tcp basp newb) {
...
@@ -1713,4 +2214,109 @@ CAF_TEST(tcp basp newb) {
CAF_MESSAGE
(
"done"
);
CAF_MESSAGE
(
"done"
);
}
}
CAF_TEST
(
udp
basp
newb
)
{
scoped_actor
main_actor
{
sys
};
actor
newb_actor
;
auto
testing
=
[
&
](
stateful_broker
<
udp_test_broker_state
>*
self
,
std
::
string
host
,
uint16_t
port
,
actor
m
)
->
behavior
{
auto
ehdl
=
self
->
add_udp_datagram_servant
(
host
,
port
);
CAF_REQUIRE
(
ehdl
);
self
->
state
.
hdl
=
std
::
move
(
*
ehdl
);
return
{
[
=
](
send_atom
,
std
::
string
str
)
{
CAF_MESSAGE
(
"sending '"
<<
str
<<
"'"
);
byte_buffer
buf
;
udp_ordering_header
ohdr
{
0
};
udp_basp_header
bhdr
{
0
,
1
,
2
};
binary_serializer
bs
(
sys
,
buf
);
bs
(
ohdr
);
auto
ordering_header_len
=
buf
.
size
();
CAF_REQUIRE
(
ordering_header_len
==
udp_ordering_header_len
);
bs
(
bhdr
);
auto
header_len
=
buf
.
size
();
CAF_REQUIRE
(
header_len
==
udp_ordering_header_len
+
udp_basp_header_len
);
bs
(
str
);
bhdr
.
payload_len
=
buf
.
size
()
-
header_len
;
stream_serializer
<
charbuf
>
out
{
sys
,
buf
.
data
()
+
ordering_header_len
,
sizeof
(
bhdr
.
payload_len
)};
out
(
bhdr
.
payload_len
);
CAF_MESSAGE
(
"header len: "
<<
header_len
<<
", packet_len: "
<<
buf
.
size
()
<<
", ordering header: "
<<
to_string
(
ohdr
)
<<
", basp header: "
<<
to_string
(
bhdr
));
self
->
enqueue_datagram
(
self
->
state
.
hdl
,
std
::
move
(
buf
));
self
->
flush
(
self
->
state
.
hdl
);
},
[
=
](
quit_atom
)
{
CAF_MESSAGE
(
"test broker shutting down"
);
self
->
quit
();
},
[
=
](
new_datagram_msg
&
msg
)
{
binary_deserializer
bd
(
self
->
system
(),
msg
.
buf
);
udp_ordering_header
ohdr
;
udp_basp_header
bhdr
;
std
::
string
str
;
bd
(
ohdr
);
bd
(
bhdr
);
bd
(
str
);
CAF_MESSAGE
(
"received '"
<<
str
<<
"'"
);
self
->
send
(
m
,
quit_atom
::
value
);
self
->
state
.
hdl
=
msg
.
handle
;
}
};
};
auto
helper_actor
=
sys
.
spawn
([
&
](
event_based_actor
*
self
,
actor
m
)
->
behavior
{
return
{
[
=
](
const
std
::
string
&
str
)
{
CAF_MESSAGE
(
"received '"
<<
str
<<
"'"
);
self
->
send
(
m
,
quit_atom
::
value
);
},
[
=
](
actor
a
)
{
CAF_MESSAGE
(
"got new newb handle"
);
self
->
send
(
m
,
a
);
},
[
=
](
quit_atom
)
{
CAF_MESSAGE
(
"helper shutting down"
);
self
->
quit
();
}
};
},
main_actor
);
CAF_MESSAGE
(
"creating new acceptor"
);
auto
newb_acceptor_ptr
=
make_newb_acceptor
<
udp_newb_acceptor_t
,
udp_accept_policy_t
>
(
sys
,
port
);
dynamic_cast
<
udp_newb_acceptor_t
*>
(
newb_acceptor_ptr
.
get
())
->
responder
=
helper_actor
;
CAF_MESSAGE
(
"contacting from 'old-style' broker"
);
auto
test_broker
=
sys
.
middleman
().
spawn_broker
(
testing
,
host
,
port
,
main_actor
);
main_actor
->
send
(
test_broker
,
send_atom
::
value
,
"hello world"
);
std
::
this_thread
::
sleep_for
(
std
::
chrono
::
seconds
(
1
));
main_actor
->
receive
(
[
&
](
actor
a
)
{
newb_actor
=
a
;
}
);
CAF_MESSAGE
(
"new newb was created"
);
main_actor
->
receive
(
[](
quit_atom
)
{
CAF_MESSAGE
(
"check"
);
}
);
CAF_MESSAGE
(
"sending message from newb"
);
main_actor
->
send
(
newb_actor
,
send_atom
::
value
,
actor_id
{
3
},
actor_id
{
4
},
"dlrow olleh"
);
main_actor
->
receive
(
[](
quit_atom
)
{
CAF_MESSAGE
(
"check"
);
}
);
CAF_MESSAGE
(
"shutting everything down"
);
newb_acceptor_ptr
->
stop
();
anon_send
(
newb_actor
,
quit_atom
::
value
);
anon_send
(
helper_actor
,
quit_atom
::
value
);
anon_send
(
test_broker
,
quit_atom
::
value
);
sys
.
await_all_actors_done
();
CAF_MESSAGE
(
"done"
);
}
CAF_TEST_FIXTURE_SCOPE_END
();
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