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
9b548fbb
Commit
9b548fbb
authored
Nov 17, 2016
by
Joseph Noir
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Add client side handshake message
parent
f435999b
Changes
7
Show whitespace changes
Inline
Side-by-side
Showing
7 changed files
with
93 additions
and
34 deletions
+93
-34
libcaf_io/caf/io/basp/instance.hpp
libcaf_io/caf/io/basp/instance.hpp
+3
-0
libcaf_io/caf/io/network/default_multiplexer.hpp
libcaf_io/caf/io/network/default_multiplexer.hpp
+1
-2
libcaf_io/src/basp_broker.cpp
libcaf_io/src/basp_broker.cpp
+12
-4
libcaf_io/src/default_multiplexer.cpp
libcaf_io/src/default_multiplexer.cpp
+33
-17
libcaf_io/src/instance.cpp
libcaf_io/src/instance.cpp
+12
-0
libcaf_io/test/basp.cpp
libcaf_io/test/basp.cpp
+29
-8
libcaf_io/test/datagram.cpp
libcaf_io/test/datagram.cpp
+3
-3
No files found.
libcaf_io/caf/io/basp/instance.hpp
View file @
9b548fbb
...
@@ -198,6 +198,9 @@ public:
...
@@ -198,6 +198,9 @@ public:
void
write_client_handshake
(
execution_unit
*
ctx
,
void
write_client_handshake
(
execution_unit
*
ctx
,
buffer_type
&
buf
,
const
node_id
&
remote_side
);
buffer_type
&
buf
,
const
node_id
&
remote_side
);
/// Start handshake ...
void
write_udp_client_handshake
(
execution_unit
*
ctx
,
buffer_type
&
buf
);
/// Writes an `announce_proxy` to `buf`.
/// Writes an `announce_proxy` to `buf`.
void
write_announce_proxy
(
execution_unit
*
ctx
,
buffer_type
&
buf
,
void
write_announce_proxy
(
execution_unit
*
ctx
,
buffer_type
&
buf
,
const
node_id
&
dest_node
,
actor_id
aid
);
const
node_id
&
dest_node
,
actor_id
aid
);
...
...
libcaf_io/caf/io/network/default_multiplexer.hpp
View file @
9b548fbb
...
@@ -638,8 +638,6 @@ private:
...
@@ -638,8 +638,6 @@ private:
// state for writing
// state for writing
manager_ptr
writer_
;
manager_ptr
writer_
;
bool
ack_writes_
;
bool
ack_writes_
;
bool
writing_
;
size_t
written_
;
buffer_type
wr_buf_
;
buffer_type
wr_buf_
;
buffer_type
wr_offline_buf_
;
buffer_type
wr_offline_buf_
;
...
@@ -716,6 +714,7 @@ private:
...
@@ -716,6 +714,7 @@ private:
manager_ptr
mgr_
;
manager_ptr
mgr_
;
size_t
dgram_size_
;
size_t
dgram_size_
;
buffer_type
rd_buf_
;
buffer_type
rd_buf_
;
size_t
bytes_read_
;
// general state
// general state
struct
sockaddr_storage
sockaddr_
;
struct
sockaddr_storage
sockaddr_
;
...
...
libcaf_io/src/basp_broker.cpp
View file @
9b548fbb
...
@@ -678,13 +678,14 @@ behavior basp_broker::make_behavior() {
...
@@ -678,13 +678,14 @@ behavior basp_broker::make_behavior() {
// received from underlying broker implementation
// received from underlying broker implementation
[
=
](
const
acceptor_closed_msg
&
msg
)
{
[
=
](
const
acceptor_closed_msg
&
msg
)
{
CAF_LOG_TRACE
(
""
);
CAF_LOG_TRACE
(
""
);
std
::
cerr
<<
"[BB] Received acceptor closed msg"
<<
std
::
endl
;
auto
port
=
local_port
(
msg
.
handle
);
auto
port
=
local_port
(
msg
.
handle
);
state
.
instance
.
remove_published_actor
(
port
);
state
.
instance
.
remove_published_actor
(
port
);
},
},
// received from underlying broker implementation
// received from underlying broker implementation
[
=
](
const
new_endpoint_msg
&
msg
)
{
[
=
](
const
new_endpoint_msg
&
msg
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
msg
.
handle
));
CAF_LOG_TRACE
(
CAF_ARG
(
msg
.
handle
));
std
::
cerr
<<
"new endpoint msg received"
<<
std
::
endl
;
std
::
cerr
<<
"
[BB]
new endpoint msg received"
<<
std
::
endl
;
//auto& bi = state.instance;
//auto& bi = state.instance;
//bi.write_server_handshake(context(), wr_buf(msg.handle),
//bi.write_server_handshake(context(), wr_buf(msg.handle),
// local_port(msg.source));
// local_port(msg.source));
...
@@ -693,7 +694,7 @@ behavior basp_broker::make_behavior() {
...
@@ -693,7 +694,7 @@ behavior basp_broker::make_behavior() {
// TODO: is this right?
// TODO: is this right?
},
},
[
=
](
const
dgram_delegate_msg
&
)
{
[
=
](
const
dgram_delegate_msg
&
)
{
std
::
cerr
<<
"Received new delegate message"
<<
std
::
endl
;
std
::
cerr
<<
"
[BB]
Received new delegate message"
<<
std
::
endl
;
},
},
// received from underlying broker implementation
// received from underlying broker implementation
[
=
](
const
dgram_scribe_closed_msg
&
msg
)
{
[
=
](
const
dgram_scribe_closed_msg
&
msg
)
{
...
@@ -714,6 +715,12 @@ behavior basp_broker::make_behavior() {
...
@@ -714,6 +715,12 @@ behavior basp_broker::make_behavior() {
auto
port
=
local_port
(
msg
.
handle
);
auto
port
=
local_port
(
msg
.
handle
);
state
.
instance
.
remove_published_actor
(
port
);
state
.
instance
.
remove_published_actor
(
port
);
},
},
[
=
](
const
dgram_doorman_closed_msg
&
msg
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
msg
.
handle
));
std
::
cerr
<<
"[BB] Received dgram_doorman_closed_msg"
<<
std
::
endl
;
auto
port
=
local_port
(
msg
.
handle
);
state
.
instance
.
remove_published_actor
(
port
);
},
// received from middleman actor
// received from middleman actor
[
=
](
publish_atom
,
accept_handle
hdl
,
uint16_t
port
,
[
=
](
publish_atom
,
accept_handle
hdl
,
uint16_t
port
,
const
strong_actor_ptr
&
whom
,
std
::
set
<
std
::
string
>&
sigs
)
{
const
strong_actor_ptr
&
whom
,
std
::
set
<
std
::
string
>&
sigs
)
{
...
@@ -770,7 +777,7 @@ behavior basp_broker::make_behavior() {
...
@@ -770,7 +777,7 @@ behavior basp_broker::make_behavior() {
[
=
](
connect_atom
,
dgram_scribe_handle
hdl
,
[
=
](
connect_atom
,
dgram_scribe_handle
hdl
,
const
std
::
string
&
host
,
uint16_t
port
)
{
const
std
::
string
&
host
,
uint16_t
port
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
hdl
.
id
()));
CAF_LOG_TRACE
(
CAF_ARG
(
hdl
.
id
()));
std
::
cerr
<<
"connect for dgram to "
<<
host
<<
":"
<<
port
<<
std
::
endl
;
std
::
cerr
<<
"
[BB]
connect for dgram to "
<<
host
<<
":"
<<
port
<<
std
::
endl
;
auto
rp
=
make_response_promise
();
auto
rp
=
make_response_promise
();
auto
res
=
assign_dgram_scribe
(
hdl
,
host
,
port
);
auto
res
=
assign_dgram_scribe
(
hdl
,
host
,
port
);
if
(
res
)
{
if
(
res
)
{
...
@@ -779,8 +786,9 @@ behavior basp_broker::make_behavior() {
...
@@ -779,8 +786,9 @@ behavior basp_broker::make_behavior() {
ctx
.
remote_port
=
port
;
ctx
.
remote_port
=
port
;
ctx
.
callback
=
rp
;
ctx
.
callback
=
rp
;
auto
&
bi
=
state
.
instance
;
auto
&
bi
=
state
.
instance
;
bi
.
write_server_handshake
(
context
(),
wr_buf
(
hdl
),
local_port
(
hdl
)
);
//bi.write_server_handshake(context(), wr_buf(hdl), port
);
//bi.write_client_handshake(context(), wr_buf(hdl), local_port(hdl));
//bi.write_client_handshake(context(), wr_buf(hdl), local_port(hdl));
bi
.
write_udp_client_handshake
(
context
(),
wr_buf
(
hdl
));
flush
(
hdl
);
flush
(
hdl
);
configure_datagram_size
(
hdl
,
1500
);
configure_datagram_size
(
hdl
,
1500
);
}
else
{
}
else
{
...
...
libcaf_io/src/default_multiplexer.cpp
View file @
9b548fbb
...
@@ -599,6 +599,7 @@ int del_flag(operation op, int bf) {
...
@@ -599,6 +599,7 @@ int del_flag(operation op, int bf) {
CAF_LOG_ERROR
(
"unexpected operation"
);
CAF_LOG_ERROR
(
"unexpected operation"
);
break
;
break
;
}
}
std
::
cerr
<<
"WEIRD STUFF!"
<<
std
::
endl
;
// weird stuff going on
// weird stuff going on
return
0
;
return
0
;
}
}
...
@@ -908,6 +909,7 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd,
...
@@ -908,6 +909,7 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd,
public:
public:
impl
(
abstract_broker
*
ptr
,
default_multiplexer
&
mx
,
native_socket
sockfd
)
impl
(
abstract_broker
*
ptr
,
default_multiplexer
&
mx
,
native_socket
sockfd
)
:
dgram_scribe
(
ptr
,
network
::
dg_sink_hdl_from_socket
(
sockfd
)),
:
dgram_scribe
(
ptr
,
network
::
dg_sink_hdl_from_socket
(
sockfd
)),
launched_
(
false
),
communicator_
(
mx
,
sockfd
)
{
communicator_
(
mx
,
sockfd
)
{
// nop
// nop
}
}
...
@@ -1249,30 +1251,34 @@ bool try_accept(native_socket& result, native_socket fd) {
...
@@ -1249,30 +1251,34 @@ bool try_accept(native_socket& result, native_socket fd) {
bool
send_datagram
(
size_t
&
result
,
native_socket
fd
,
void
*
buf
,
size_t
buf_len
,
bool
send_datagram
(
size_t
&
result
,
native_socket
fd
,
void
*
buf
,
size_t
buf_len
,
sockaddr_storage
&
sa
,
size_t
sa_len
)
{
sockaddr_storage
&
sa
,
size_t
sa_len
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
fd
)
<<
CAF_ARG
(
len
));
CAF_LOG_TRACE
(
CAF_ARG
(
fd
)
<<
CAF_ARG
(
buf_
len
));
std
::
cerr
<<
"
Sending datagram
"
<<
std
::
endl
;
std
::
cerr
<<
"
[SD] Sending datagram ...
"
<<
std
::
endl
;
auto
sres
=
::
sendto
(
fd
,
reinterpret_cast
<
socket_send_ptr
>
(
buf
),
buf_len
,
auto
sres
=
::
sendto
(
fd
,
reinterpret_cast
<
socket_send_ptr
>
(
buf
),
buf_len
,
no_sigpipe_flag
,
reinterpret_cast
<
sockaddr
*>
(
&
sa
),
no_sigpipe_flag
,
reinterpret_cast
<
sockaddr
*>
(
&
sa
),
sa_len
);
sa_len
);
/*
if
(
is_error
(
sres
,
true
))
{
auto sres = ::send(fd, reinterpret_cast<socket_send_ptr>(buf), buf_len,
std
::
cerr
<<
"[SD] Send error? ("
<<
sres
<<
")"
<<
std
::
endl
;
no_sigpipe_flag);
*/
if
(
is_error
(
sres
,
true
))
return
false
;
return
false
;
}
result
=
(
sres
>
0
)
?
static_cast
<
size_t
>
(
sres
)
:
0
;
result
=
(
sres
>
0
)
?
static_cast
<
size_t
>
(
sres
)
:
0
;
std
::
cerr
<<
"[SD] Sent "
<<
result
<<
" bytes"
<<
std
::
endl
;
return
true
;
return
true
;
}
}
bool
receive_datagram
(
size_t
&
result
,
native_socket
fd
,
void
*
buf
,
size_t
len
,
bool
receive_datagram
(
size_t
&
result
,
native_socket
fd
,
void
*
buf
,
size_t
len
,
sockaddr_storage
&
sender_addr
,
socklen_t
&
sender_len
)
{
sockaddr_storage
&
sender_addr
,
socklen_t
&
sender_len
)
{
std
::
cerr
<<
"Receiving datagram"
<<
std
::
endl
;
sender_len
=
sizeof
(
sender_addr
);
CAF_LOG_TRACE
(
CAF_ARG
(
fd
));
CAF_LOG_TRACE
(
CAF_ARG
(
fd
));
//sender_len = sizeof(sender_addr);
sender_len
=
sizeof
(
sockaddr
);
std
::
cerr
<<
"[RD] Receiving datagram ..."
<<
std
::
endl
;
std
::
cerr
<<
"[RD] Settings: buffsize "
<<
len
<<
", "
<<
"addrsize "
<<
sender_len
<<
", "
<<
"socket "
<<
fd
<<
std
::
endl
;
auto
sres
=
::
recvfrom
(
fd
,
buf
,
len
,
no_sigpipe_flag
,
auto
sres
=
::
recvfrom
(
fd
,
buf
,
len
,
no_sigpipe_flag
,
reinterpret_cast
<
struct
sockaddr
*>
(
&
sender_addr
),
reinterpret_cast
<
struct
sockaddr
*>
(
&
sender_addr
),
&
sender_len
);
&
sender_len
);
if
(
is_error
(
sres
,
true
)
||
sres
==
0
)
{
if
(
is_error
(
sres
,
true
)
||
sres
==
0
)
{
std
::
cerr
<<
"[RD] Nothing to receive (res: "
<<
sres
<<
")"
<<
std
::
endl
;
// Nothing to receive
// Nothing to receive
return
false
;
return
false
;
}
}
...
@@ -1580,7 +1586,6 @@ dgram_communicator::dgram_communicator(default_multiplexer& backend_ref,
...
@@ -1580,7 +1586,6 @@ dgram_communicator::dgram_communicator(default_multiplexer& backend_ref,
:
event_handler
(
backend_ref
,
sockfd
),
:
event_handler
(
backend_ref
,
sockfd
),
dgram_size_
(
0
),
dgram_size_
(
0
),
ack_writes_
(
false
),
ack_writes_
(
false
),
writing_
(
false
),
waiting_for_remote_endpoint
(
true
)
{
waiting_for_remote_endpoint
(
true
)
{
// TODO: Set some reasonable default.
// TODO: Set some reasonable default.
configure_datagram_size
(
1500
);
configure_datagram_size
(
1500
);
...
@@ -1595,6 +1600,7 @@ void dgram_communicator::activate(manager_type* mgr) {
...
@@ -1595,6 +1600,7 @@ void dgram_communicator::activate(manager_type* mgr) {
if
(
!
writer_
)
{
if
(
!
writer_
)
{
writer_
.
reset
(
mgr
);
writer_
.
reset
(
mgr
);
event_handler
::
activate
();
event_handler
::
activate
();
prepare_next_read
();
}
}
}
}
...
@@ -1607,6 +1613,8 @@ void dgram_communicator::ack_writes(bool x) {
...
@@ -1607,6 +1613,8 @@ void dgram_communicator::ack_writes(bool x) {
}
}
void
dgram_communicator
::
write
(
const
void
*
buf
,
size_t
num_bytes
)
{
void
dgram_communicator
::
write
(
const
void
*
buf
,
size_t
num_bytes
)
{
std
::
cerr
<<
"[DC] write called with "
<<
num_bytes
<<
" bytes"
<<
std
::
endl
;
CAF_LOG_TRACE
(
CAF_ARG
(
num_bytes
));
CAF_LOG_TRACE
(
CAF_ARG
(
num_bytes
));
auto
first
=
reinterpret_cast
<
const
char
*>
(
buf
);
auto
first
=
reinterpret_cast
<
const
char
*>
(
buf
);
auto
last
=
first
+
num_bytes
;
auto
last
=
first
+
num_bytes
;
...
@@ -1616,7 +1624,7 @@ void dgram_communicator::write(const void* buf, size_t num_bytes) {
...
@@ -1616,7 +1624,7 @@ void dgram_communicator::write(const void* buf, size_t num_bytes) {
void
dgram_communicator
::
flush
(
const
manager_ptr
&
mgr
)
{
void
dgram_communicator
::
flush
(
const
manager_ptr
&
mgr
)
{
CAF_ASSERT
(
mgr
!=
nullptr
);
CAF_ASSERT
(
mgr
!=
nullptr
);
CAF_LOG_TRACE
(
CAF_ARG
(
wr_offline_buf_
.
size
()));
CAF_LOG_TRACE
(
CAF_ARG
(
wr_offline_buf_
.
size
()));
if
(
wr_offline_buf_
.
empty
())
{
if
(
!
wr_offline_buf_
.
empty
())
{
backend
().
add
(
operation
::
write
,
fd
(),
this
);
backend
().
add
(
operation
::
write
,
fd
(),
this
);
writer_
=
mgr
;
writer_
=
mgr
;
prepare_next_write
();
prepare_next_write
();
...
@@ -1639,6 +1647,8 @@ void dgram_communicator::removed_from_loop(operation op) {
...
@@ -1639,6 +1647,8 @@ void dgram_communicator::removed_from_loop(operation op) {
void
dgram_communicator
::
handle_event
(
operation
op
)
{
void
dgram_communicator
::
handle_event
(
operation
op
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
op
));
CAF_LOG_TRACE
(
CAF_ARG
(
op
));
std
::
cerr
<<
"[DC] handle_event: "
<<
to_string
(
op
)
<<
std
::
endl
;
switch
(
op
)
{
switch
(
op
)
{
case
operation
:
:
read
:
{
case
operation
:
:
read
:
{
// incoming message should signify a new remote enpoint
// incoming message should signify a new remote enpoint
...
@@ -1681,7 +1691,7 @@ void dgram_communicator::handle_event(operation op) {
...
@@ -1681,7 +1691,7 @@ void dgram_communicator::handle_event(operation op) {
}
else
{
}
else
{
// TODO: remove this if sure that datagrams are either written
// TODO: remove this if sure that datagrams are either written
// as a whole or not at all
// as a whole or not at all
std
::
cerr
<<
"Partial datagram wrtten: "
<<
wb
std
::
cerr
<<
"
[DC]
Partial datagram wrtten: "
<<
wb
<<
" of "
<<
wr_buf_
.
size
()
<<
std
::
endl
;
<<
" of "
<<
wr_buf_
.
size
()
<<
std
::
endl
;
if
(
writer_
)
if
(
writer_
)
writer_
->
io_failure
(
&
backend
(),
operation
::
write
);
writer_
->
io_failure
(
&
backend
(),
operation
::
write
);
...
@@ -1769,6 +1779,7 @@ void dgram_acceptor::activate(manager_type* mgr) {
...
@@ -1769,6 +1779,7 @@ void dgram_acceptor::activate(manager_type* mgr) {
if
(
!
mgr_
)
{
if
(
!
mgr_
)
{
mgr_
.
reset
(
mgr
);
mgr_
.
reset
(
mgr
);
event_handler
::
activate
();
event_handler
::
activate
();
prepare_next_read
();
}
}
}
}
...
@@ -1799,11 +1810,16 @@ void dgram_acceptor::handle_event(operation op) {
...
@@ -1799,11 +1810,16 @@ void dgram_acceptor::handle_event(operation op) {
sender_from_sockaddr
(
sockaddr_
,
sockaddr_len_
);
sender_from_sockaddr
(
sockaddr_
,
sockaddr_len_
);
if
(
rb
==
0
)
if
(
rb
==
0
)
return
;
return
;
std
::
cerr
<<
"[DA] received "
<<
rb
<<
" bytes data from "
<<
host_
<<
":"
<<
port_
<<
std
::
endl
;
auto
res
=
mgr_
->
new_endpoint
(
rd_buf_
.
data
(),
rb
);
auto
res
=
mgr_
->
new_endpoint
(
rd_buf_
.
data
(),
rb
);
if
(
res
)
{
if
(
!
res
)
{
// What is the right way to propagate this?
// What is the right way to propagate this?
CAF_LOG_DEBUG
(
"Failure during creation of new udp endpoint"
);
CAF_LOG_DEBUG
(
"Failure during creation of new udp endpoint"
);
std
::
cerr
<<
"[DA] Failure during creation of new udp endpoint"
<<
std
::
endl
;;
}
}
bytes_read_
=
rb
;
break
;
break
;
}
}
case
operation
:
:
write
:
{
case
operation
:
:
write
:
{
...
@@ -1827,16 +1843,14 @@ void dgram_acceptor::sender_from_sockaddr(const sockaddr_storage& sa, size_t) {
...
@@ -1827,16 +1843,14 @@ void dgram_acceptor::sender_from_sockaddr(const sockaddr_storage& sa, size_t) {
inet_ntop
(
AF_INET
,
inet_ntop
(
AF_INET
,
&
reinterpret_cast
<
const
sockaddr_in
*>
(
&
sa
)
->
sin_addr
,
&
reinterpret_cast
<
const
sockaddr_in
*>
(
&
sa
)
->
sin_addr
,
addr
,
INET_ADDRSTRLEN
);
addr
,
INET_ADDRSTRLEN
);
host_
.
insert
(
std
::
begin
(
host_
),
std
::
begin
(
addr
),
host_
=
addr
;
std
::
begin
(
addr
)
+
INET_ADDRSTRLEN
);
break
;
break
;
case
AF_INET6
:
case
AF_INET6
:
port_
=
ntohs
(
reinterpret_cast
<
const
sockaddr_in6
*>
(
&
sa
)
->
sin6_port
);
port_
=
ntohs
(
reinterpret_cast
<
const
sockaddr_in6
*>
(
&
sa
)
->
sin6_port
);
inet_ntop
(
AF_INET6
,
inet_ntop
(
AF_INET6
,
&
reinterpret_cast
<
const
sockaddr_in
*>
(
&
sa
)
->
sin_addr
,
&
reinterpret_cast
<
const
sockaddr_in
*>
(
&
sa
)
->
sin_addr
,
addr
,
INET6_ADDRSTRLEN
);
addr
,
INET6_ADDRSTRLEN
);
host_
.
insert
(
std
::
begin
(
host_
),
std
::
begin
(
addr
),
host_
=
addr
;
std
::
begin
(
addr
)
+
INET6_ADDRSTRLEN
);
break
;
break
;
default:
default:
// nop
// nop
...
@@ -2121,6 +2135,8 @@ new_dgram_doorman_impl(uint16_t port, const char* addr, bool reuse_addr) {
...
@@ -2121,6 +2135,8 @@ new_dgram_doorman_impl(uint16_t port, const char* addr, bool reuse_addr) {
return
std
::
move
(
p
.
error
());
return
std
::
move
(
p
.
error
());
// ok, no errors so far
// ok, no errors so far
CAF_LOG_DEBUG
(
CAF_ARG
(
fd
)
<<
CAF_ARG
(
p
));
CAF_LOG_DEBUG
(
CAF_ARG
(
fd
)
<<
CAF_ARG
(
p
));
std
::
cerr
<<
"Created new dgram doorman on socket "
<<
fd
<<
" using port "
<<
*
p
<<
std
::
endl
;
return
std
::
make_pair
(
sguard
.
release
(),
*
p
);
return
std
::
make_pair
(
sguard
.
release
(),
*
p
);
}
}
...
...
libcaf_io/src/instance.cpp
View file @
9b548fbb
...
@@ -419,6 +419,11 @@ void instance::write_server_handshake(execution_unit* ctx,
...
@@ -419,6 +419,11 @@ void instance::write_server_handshake(execution_unit* ctx,
pa
=
&
i
->
second
;
pa
=
&
i
->
second
;
}
}
CAF_LOG_DEBUG_IF
(
!
pa
&&
port
,
"no actor published"
);
CAF_LOG_DEBUG_IF
(
!
pa
&&
port
,
"no actor published"
);
if
(
!
pa
&&
port
)
{
std
::
cerr
<<
"No actor published."
<<
std
::
endl
;
}
else
{
std
::
cerr
<<
"Found locally published actor."
<<
std
::
endl
;
}
auto
writer
=
make_callback
([
&
](
serializer
&
sink
)
->
error
{
auto
writer
=
make_callback
([
&
](
serializer
&
sink
)
->
error
{
auto
&
ref
=
callee_
.
system
().
config
().
middleman_app_identifier
;
auto
&
ref
=
callee_
.
system
().
config
().
middleman_app_identifier
;
auto
e
=
sink
(
const_cast
<
std
::
string
&>
(
ref
));
auto
e
=
sink
(
const_cast
<
std
::
string
&>
(
ref
));
...
@@ -453,6 +458,13 @@ void instance::write_client_handshake(execution_unit* ctx,
...
@@ -453,6 +458,13 @@ void instance::write_client_handshake(execution_unit* ctx,
write
(
ctx
,
buf
,
hdr
,
&
writer
);
write
(
ctx
,
buf
,
hdr
,
&
writer
);
}
}
void
instance
::
write_udp_client_handshake
(
execution_unit
*
ctx
,
buffer_type
&
buf
)
{
header
hdr
{
message_type
::
client_handshake
,
0
,
0
,
0
,
this_node_
,
none
,
invalid_actor_id
,
invalid_actor_id
};
write
(
ctx
,
buf
,
hdr
);
}
void
instance
::
write_announce_proxy
(
execution_unit
*
ctx
,
buffer_type
&
buf
,
void
instance
::
write_announce_proxy
(
execution_unit
*
ctx
,
buffer_type
&
buf
,
const
node_id
&
dest_node
,
actor_id
aid
)
{
const
node_id
&
dest_node
,
actor_id
aid
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
dest_node
)
<<
CAF_ARG
(
aid
));
CAF_LOG_TRACE
(
CAF_ARG
(
dest_node
)
<<
CAF_ARG
(
aid
));
...
...
libcaf_io/test/basp.cpp
View file @
9b548fbb
...
@@ -702,7 +702,13 @@ CAF_TEST(actor_serialize_and_deserialize) {
...
@@ -702,7 +702,13 @@ CAF_TEST(actor_serialize_and_deserialize) {
std
::
vector
<
actor_id
>
{},
msg
);
std
::
vector
<
actor_id
>
{},
msg
);
}
}
// TODO: Reimplement this once reactive
// addressing is available.
CAF_TEST
(
indirect_connections
)
{
CAF_TEST
(
indirect_connections
)
{
CAF_MESSAGE
(
"Currently not implemented."
);
CAF_CHECK_EQUAL
(
true
,
false
);
/*
// this node receives a message from jupiter via mars and responds via mars
// this node receives a message from jupiter via mars and responds via mars
// and any ad-hoc automatic connection requests are ignored
// and any ad-hoc automatic connection requests are ignored
CAF_MESSAGE("self: " << to_string(self()->address()));
CAF_MESSAGE("self: " << to_string(self()->address()));
...
@@ -750,13 +756,17 @@ CAF_TEST(indirect_connections) {
...
@@ -750,13 +756,17 @@ CAF_TEST(indirect_connections) {
self()->id(), jupiter().dummy_actor->id(),
self()->id(), jupiter().dummy_actor->id(),
std::vector<actor_id>{},
std::vector<actor_id>{},
make_message("hello from earth!"));
make_message("hello from earth!"));
*/
}
}
CAF_TEST_FIXTURE_SCOPE_END
()
CAF_TEST_FIXTURE_SCOPE_END
()
CAF_TEST_FIXTURE_SCOPE
(
basp_tests_with_autoconn
,
autoconn_enabled_fixture
)
CAF_TEST_FIXTURE_SCOPE
(
basp_tests_with_autoconn
,
autoconn_enabled_fixture
)
CAF_TEST
(
automatic_connection
)
{
CAF_TEST
(
automatic_connection
)
{
CAF_MESSAGE
(
"Currently not implemented."
);
CAF_CHECK_EQUAL
(
true
,
false
);
// this tells our BASP broker to enable the automatic connection feature
// this tells our BASP broker to enable the automatic connection feature
//anon_send(aut(), ok_atom::value,
//anon_send(aut(), ok_atom::value,
// "middleman.enable-automatic-connections", make_message(true));
// "middleman.enable-automatic-connections", make_message(true));
...
@@ -764,6 +774,14 @@ CAF_TEST(automatic_connection) {
...
@@ -764,6 +774,14 @@ CAF_TEST(automatic_connection) {
// jupiter [remote hdl 0] -> mars [remote hdl 1] -> earth [this_node]
// jupiter [remote hdl 0] -> mars [remote hdl 1] -> earth [this_node]
// (this node receives a message from jupiter via mars and responds via mars,
// (this node receives a message from jupiter via mars and responds via mars,
// but then also establishes a connection to jupiter directly)
// but then also establishes a connection to jupiter directly)
/*
auto check_node_in_tbl = [&](node& n) {
io::id_visitor id_vis;
auto opt_hdl = tbl().lookup_hdl(n.id);
CAF_REQUIRE(opt_hdl);
CAF_CHECK_EQUAL(apply_visitor(id_vis, *opt_hdl),
n.connection.id());
};
mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
CAF_MESSAGE("self: " << to_string(self()->address()));
CAF_MESSAGE("self: " << to_string(self()->address()));
...
@@ -773,7 +791,7 @@ CAF_TEST(automatic_connection) {
...
@@ -773,7 +791,7 @@ CAF_TEST(automatic_connection) {
mpx()->flush_runnables(); // process publish message in basp_broker
mpx()->flush_runnables(); // process publish message in basp_broker
CAF_MESSAGE("connect to mars");
CAF_MESSAGE("connect to mars");
connect_node(mars(), ax, self()->id());
connect_node(mars(), ax, self()->id());
CAF_CHECK_EQUAL
(
tbl
().
lookup_direct
(
mars
().
id
).
id
(),
mars
().
connection
.
id
());
check_node_in_tbl(mars
());
CAF_MESSAGE("simulate that an actor from jupiter "
CAF_MESSAGE("simulate that an actor from jupiter "
"sends a message to us via mars");
"sends a message to us via mars");
mock(mars().connection,
mock(mars().connection,
...
@@ -803,8 +821,9 @@ CAF_TEST(automatic_connection) {
...
@@ -803,8 +821,9 @@ CAF_TEST(automatic_connection) {
no_operation_data, this_node(), jupiter().id,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id());
invalid_actor_id, jupiter().dummy_actor->id());
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
CAF_CHECK_EQUAL
(
tbl
().
lookup_indirect
(
jupiter
().
id
),
mars
().
id
);
// TODO: remove this? (Indirection removed ...)
CAF_CHECK_EQUAL
(
tbl
().
lookup_indirect
(
mars
().
id
),
none
);
//CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), mars().id);
//CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
auto connection_helper = system.latest_actor_id();
auto connection_helper = system.latest_actor_id();
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
// create a dummy config server and respond to the name lookup
// create a dummy config server and respond to the name lookup
...
@@ -835,10 +854,11 @@ CAF_TEST(automatic_connection) {
...
@@ -835,10 +854,11 @@ CAF_TEST(automatic_connection) {
basp::message_type::client_handshake, no_flags, 1u,
basp::message_type::client_handshake, no_flags, 1u,
no_operation_data, this_node(), jupiter().id,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, invalid_actor_id, std::string{});
invalid_actor_id, invalid_actor_id, std::string{});
CAF_CHECK_EQUAL
(
tbl
().
lookup_indirect
(
jupiter
().
id
),
none
);
// TODO: remove this? (Indirection removed ...)
CAF_CHECK_EQUAL
(
tbl
().
lookup_indirect
(
mars
().
id
),
none
);
//CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), none);
CAF_CHECK_EQUAL
(
tbl
().
lookup_direct
(
jupiter
().
id
).
id
(),
jupiter
().
connection
.
id
());
//CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
CAF_CHECK_EQUAL
(
tbl
().
lookup_direct
(
mars
().
id
).
id
(),
mars
().
connection
.
id
());
check_node_in_tbl(mars());
check_node_in_tbl(jupiter());
CAF_MESSAGE("receive message from jupiter");
CAF_MESSAGE("receive message from jupiter");
self()->receive(
self()->receive(
[](const std::string& str) -> std::string {
[](const std::string& str) -> std::string {
...
@@ -856,6 +876,7 @@ CAF_TEST(automatic_connection) {
...
@@ -856,6 +876,7 @@ CAF_TEST(automatic_connection) {
std::vector<actor_id>{},
std::vector<actor_id>{},
make_message("hello from earth!"));
make_message("hello from earth!"));
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
*/
}
}
CAF_TEST_FIXTURE_SCOPE_END
()
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_io/test/datagram.cpp
View file @
9b548fbb
...
@@ -124,9 +124,9 @@ CAF_TEST(test_datagram_remote_actor) {
...
@@ -124,9 +124,9 @@ CAF_TEST(test_datagram_remote_actor) {
CAF_MESSAGE
(
"Published pong on: "
+
to_string
(
*
res3
)
+
"."
);
CAF_MESSAGE
(
"Published pong on: "
+
to_string
(
*
res3
)
+
"."
);
CAF_SET_LOGGER_SYS
(
&
server_side
);
CAF_SET_LOGGER_SYS
(
&
server_side
);
CAF_MESSAGE
(
"Local call to remote actor should acquire the actor."
);
//
CAF_MESSAGE("Local call to remote actor should acquire the actor.");
auto
res4
=
server_side_mm
.
remote_actor
(
*
res3
);
//
auto res4 = server_side_mm.remote_actor(*res3);
CAF_CHECK
(
res4
);
//
CAF_CHECK(res4);
CAF_MESSAGE
(
"Checking from different actor system next."
);
CAF_MESSAGE
(
"Checking from different actor system next."
);
auto
res5
=
client_side_mm
.
remote_actor
(
*
res3
);
auto
res5
=
client_side_mm
.
remote_actor
(
*
res3
);
...
...
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