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
2008a863
Unverified
Commit
2008a863
authored
Jan 20, 2020
by
Joseph Noir
Committed by
GitHub
Jan 20, 2020
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #1023
Report errors while connecting two CAF nodes
parents
d2bb88a6
d299dd48
Changes
7
Show whitespace changes
Inline
Side-by-side
Showing
7 changed files
with
98 additions
and
41 deletions
+98
-41
libcaf_core/caf/sec.hpp
libcaf_core/caf/sec.hpp
+11
-0
libcaf_io/CMakeLists.txt
libcaf_io/CMakeLists.txt
+2
-0
libcaf_io/caf/io/basp/connection_state.hpp
libcaf_io/caf/io/basp/connection_state.hpp
+49
-6
libcaf_io/caf/io/basp/instance.hpp
libcaf_io/caf/io/basp/instance.hpp
+2
-2
libcaf_io/caf/io/basp_broker.hpp
libcaf_io/caf/io/basp_broker.hpp
+1
-1
libcaf_io/src/io/basp/instance.cpp
libcaf_io/src/io/basp/instance.cpp
+23
-25
libcaf_io/src/io/basp_broker.cpp
libcaf_io/src/io/basp_broker.cpp
+10
-7
No files found.
libcaf_core/caf/sec.hpp
View file @
2008a863
...
@@ -126,6 +126,17 @@ enum class sec : uint8_t {
...
@@ -126,6 +126,17 @@ enum class sec : uint8_t {
socket_operation_failed
=
45
,
socket_operation_failed
=
45
,
/// A resource is temporarily unavailable or would block.
/// A resource is temporarily unavailable or would block.
unavailable_or_would_block
,
unavailable_or_would_block
,
/// Connection refused because of incompatible CAF versions.
incompatible_versions
,
/// Connection refused because of incompatible application IDs.
incompatible_application_ids
,
/// The middleman received a malformed BASP message from another node.
malformed_basp_message
,
/// The middleman closed a connection because it failed to serialize or
/// deserialize a payload.
serializing_basp_payload_failed
,
/// The middleman closed a connection to itself or an already connected node.
redundant_connection
,
/// Resolving a path on a remote node failed.
/// Resolving a path on a remote node failed.
remote_lookup_failed
,
remote_lookup_failed
,
/// Serialization failed because actor_system::tracing_context is null.
/// Serialization failed because actor_system::tracing_context is null.
...
...
libcaf_io/CMakeLists.txt
View file @
2008a863
...
@@ -6,12 +6,14 @@ file(GLOB_RECURSE CAF_IO_HEADERS "caf/*.hpp")
...
@@ -6,12 +6,14 @@ file(GLOB_RECURSE CAF_IO_HEADERS "caf/*.hpp")
enum_to_string
(
"caf/io/basp/message_type.hpp"
"message_type_to_string.cpp"
)
enum_to_string
(
"caf/io/basp/message_type.hpp"
"message_type_to_string.cpp"
)
enum_to_string
(
"caf/io/network/operation.hpp"
"operation_to_string.cpp"
)
enum_to_string
(
"caf/io/network/operation.hpp"
"operation_to_string.cpp"
)
enum_to_string
(
"caf/io/basp/connection_state.hpp"
"connection_state_to_string.cpp"
)
# -- list cpp files ------------------------------------------------------------
# -- list cpp files ------------------------------------------------------------
set
(
CAF_IO_SOURCES
set
(
CAF_IO_SOURCES
"
${
CMAKE_BINARY_DIR
}
/message_type_to_string.cpp"
"
${
CMAKE_BINARY_DIR
}
/message_type_to_string.cpp"
"
${
CMAKE_BINARY_DIR
}
/operation_to_string.cpp"
"
${
CMAKE_BINARY_DIR
}
/operation_to_string.cpp"
"
${
CMAKE_BINARY_DIR
}
/connection_state_to_string.cpp"
src/detail/socket_guard.cpp
src/detail/socket_guard.cpp
src/io/abstract_broker.cpp
src/io/abstract_broker.cpp
src/io/basp/header.cpp
src/io/basp/header.cpp
...
...
libcaf_io/caf/io/basp/connection_state.hpp
View file @
2008a863
...
@@ -18,11 +18,15 @@
...
@@ -18,11 +18,15 @@
#pragma once
#pragma once
#include "caf/detail/io_export.hpp"
#include "caf/sec.hpp"
namespace
caf
::
io
::
basp
{
namespace
caf
::
io
::
basp
{
/// @addtogroup BASP
/// @addtogroup BASP
/// Denotes the state of a connection between to BASP nodes.
/// Denotes the state of a connection between two BASP nodes. Overlaps with
/// `sec` (these states get converted to an error by the BASP instance).
enum
connection_state
{
enum
connection_state
{
/// Indicates that a connection is established and this node is
/// Indicates that a connection is established and this node is
/// waiting for the next BASP header.
/// waiting for the next BASP header.
...
@@ -31,16 +35,55 @@ enum connection_state {
...
@@ -31,16 +35,55 @@ enum connection_state {
/// and is waiting for the data.
/// and is waiting for the data.
await_payload
,
await_payload
,
/// Indicates that this connection no longer exists.
/// Indicates that this connection no longer exists.
close_connection
close_connection
,
/// See `sec::incompatible_versions`.
incompatible_versions
,
/// See `sec::incompatible_application_ids`.
incompatible_application_ids
,
/// See `sec::malformed_basp_message`.
malformed_basp_message
,
/// See `sec::serializing_basp_payload_failed`.
serializing_basp_payload_failed
,
/// See `sec::redundant_connection`.
redundant_connection
,
/// See `sec::no_route_to_receiving_node`.
no_route_to_receiving_node
,
};
};
/// Returns whether the connection state requries a shutdown of the socket
/// connection.
/// @relates connection_state
inline
bool
requires_shutdown
(
connection_state
x
)
{
// Any enum value other than await_header (0) and await_payload (1) signal the
// BASP broker to shutdown the connection.
return
static_cast
<
int
>
(
x
)
>
1
;
}
/// Converts the connection state to a system error code if it holds one of the
/// overlapping values. Otherwise returns `sec::none`.
/// @relates connection_state
/// @relates connection_state
inline
std
::
string
to_string
(
connection_state
x
)
{
inline
sec
to_sec
(
connection_state
x
)
{
return
x
==
await_header
switch
(
x
)
{
?
"await_header"
default:
:
(
x
==
await_payload
?
"await_payload"
:
"close_connection"
);
return
sec
::
none
;
case
incompatible_versions
:
return
sec
::
incompatible_versions
;
case
incompatible_application_ids
:
return
sec
::
incompatible_application_ids
;
case
malformed_basp_message
:
return
sec
::
malformed_basp_message
;
case
serializing_basp_payload_failed
:
return
sec
::
serializing_basp_payload_failed
;
case
redundant_connection
:
return
sec
::
redundant_connection
;
case
no_route_to_receiving_node
:
return
sec
::
no_route_to_receiving_node
;
}
}
}
/// @relates connection_state
CAF_IO_EXPORT
std
::
string
to_string
(
connection_state
x
);
/// @}
/// @}
}
// namespace caf::io::basp
}
// namespace caf::io::basp
libcaf_io/caf/io/basp/instance.hpp
View file @
2008a863
...
@@ -221,8 +221,8 @@ public:
...
@@ -221,8 +221,8 @@ public:
return
system
().
config
();
return
system
().
config
();
}
}
bool
handle
(
execution_unit
*
ctx
,
connection_handle
hdl
,
header
&
hdr
,
connection_state
handle
(
execution_unit
*
ctx
,
connection_handle
hdl
,
byte_buffer
*
payload
);
header
&
hdr
,
byte_buffer
*
payload
);
private:
private:
void
forward
(
execution_unit
*
ctx
,
const
node_id
&
dest_node
,
const
header
&
hdr
,
void
forward
(
execution_unit
*
ctx
,
const
node_id
&
dest_node
,
const
header
&
hdr
,
...
...
libcaf_io/caf/io/basp_broker.hpp
View file @
2008a863
...
@@ -111,7 +111,7 @@ public:
...
@@ -111,7 +111,7 @@ public:
void
set_context
(
connection_handle
hdl
);
void
set_context
(
connection_handle
hdl
);
/// Cleans up any state for `hdl`.
/// Cleans up any state for `hdl`.
void
connection_cleanup
(
connection_handle
hdl
);
void
connection_cleanup
(
connection_handle
hdl
,
sec
code
);
/// Sends a basp::down_message message to a remote node.
/// Sends a basp::down_message message to a remote node.
void
send_basp_down_message
(
const
node_id
&
nid
,
actor_id
aid
,
error
err
);
void
send_basp_down_message
(
const
node_id
&
nid
,
actor_id
aid
,
error
err
);
...
...
libcaf_io/src/io/basp/instance.cpp
View file @
2008a863
...
@@ -51,10 +51,10 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
...
@@ -51,10 +51,10 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
header
&
hdr
,
bool
is_payload
)
{
header
&
hdr
,
bool
is_payload
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
dm
)
<<
CAF_ARG
(
is_payload
));
CAF_LOG_TRACE
(
CAF_ARG
(
dm
)
<<
CAF_ARG
(
is_payload
));
// function object providing cleanup code on errors
// function object providing cleanup code on errors
auto
err
=
[
&
](
)
->
connection_state
{
auto
err
=
[
&
](
connection_state
code
)
{
if
(
auto
nid
=
tbl_
.
erase_direct
(
dm
.
handle
))
if
(
auto
nid
=
tbl_
.
erase_direct
(
dm
.
handle
))
callee_
.
purge_state
(
nid
);
callee_
.
purge_state
(
nid
);
return
c
lose_connection
;
return
c
ode
;
};
};
byte_buffer
*
payload
=
nullptr
;
byte_buffer
*
payload
=
nullptr
;
if
(
is_payload
)
{
if
(
is_payload
)
{
...
@@ -62,14 +62,14 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
...
@@ -62,14 +62,14 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
if
(
payload
->
size
()
!=
hdr
.
payload_len
)
{
if
(
payload
->
size
()
!=
hdr
.
payload_len
)
{
CAF_LOG_WARNING
(
"received invalid payload, expected"
CAF_LOG_WARNING
(
"received invalid payload, expected"
<<
hdr
.
payload_len
<<
"bytes, got"
<<
payload
->
size
());
<<
hdr
.
payload_len
<<
"bytes, got"
<<
payload
->
size
());
return
err
();
return
err
(
malformed_basp_message
);
}
}
}
else
{
}
else
{
binary_deserializer
bd
{
ctx
,
dm
.
buf
};
binary_deserializer
bd
{
ctx
,
dm
.
buf
};
auto
e
=
bd
(
hdr
);
auto
e
=
bd
(
hdr
);
if
(
e
||
!
valid
(
hdr
))
{
if
(
e
||
!
valid
(
hdr
))
{
CAF_LOG_WARNING
(
"received invalid header:"
<<
CAF_ARG
(
hdr
));
CAF_LOG_WARNING
(
"received invalid header:"
<<
CAF_ARG
(
hdr
));
return
err
();
return
err
(
malformed_basp_message
);
}
}
if
(
hdr
.
payload_len
>
0
)
{
if
(
hdr
.
payload_len
>
0
)
{
CAF_LOG_DEBUG
(
"await payload before processing further"
);
CAF_LOG_DEBUG
(
"await payload before processing further"
);
...
@@ -77,9 +77,7 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
...
@@ -77,9 +77,7 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
}
}
}
}
CAF_LOG_DEBUG
(
CAF_ARG
(
hdr
));
CAF_LOG_DEBUG
(
CAF_ARG
(
hdr
));
if
(
!
handle
(
ctx
,
dm
.
handle
,
hdr
,
payload
))
return
handle
(
ctx
,
dm
.
handle
,
hdr
,
payload
);
return
err
();
return
await_header
;
}
}
void
instance
::
handle_heartbeat
(
execution_unit
*
ctx
)
{
void
instance
::
handle_heartbeat
(
execution_unit
*
ctx
)
{
...
@@ -286,18 +284,18 @@ void instance::write_heartbeat(execution_unit* ctx, byte_buffer& buf) {
...
@@ -286,18 +284,18 @@ void instance::write_heartbeat(execution_unit* ctx, byte_buffer& buf) {
write
(
ctx
,
buf
,
hdr
);
write
(
ctx
,
buf
,
hdr
);
}
}
bool
instance
::
handle
(
execution_unit
*
ctx
,
connection_handle
hdl
,
header
&
hdr
,
connection_state
instance
::
handle
(
execution_unit
*
ctx
,
connection_handle
hdl
,
byte_buffer
*
payload
)
{
header
&
hdr
,
byte_buffer
*
payload
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
hdl
)
<<
CAF_ARG
(
hdr
));
CAF_LOG_TRACE
(
CAF_ARG
(
hdl
)
<<
CAF_ARG
(
hdr
));
// Check payload validity.
// Check payload validity.
if
(
payload
==
nullptr
)
{
if
(
payload
==
nullptr
)
{
if
(
hdr
.
payload_len
!=
0
)
{
if
(
hdr
.
payload_len
!=
0
)
{
CAF_LOG_WARNING
(
"
invalid
payload"
);
CAF_LOG_WARNING
(
"
missing
payload"
);
return
fals
e
;
return
malformed_basp_messag
e
;
}
}
}
else
if
(
hdr
.
payload_len
!=
payload
->
size
())
{
}
else
if
(
hdr
.
payload_len
!=
payload
->
size
())
{
CAF_LOG_WARNING
(
"
invalid payload
"
);
CAF_LOG_WARNING
(
"
actual payload size differs from advertised size
"
);
return
fals
e
;
return
malformed_basp_messag
e
;
}
}
// Dispatch by message type.
// Dispatch by message type.
switch
(
hdr
.
operation
)
{
switch
(
hdr
.
operation
)
{
...
@@ -311,7 +309,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
...
@@ -311,7 +309,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if
(
auto
err
=
bd
(
source_node
,
app_ids
,
aid
,
sigs
))
{
if
(
auto
err
=
bd
(
source_node
,
app_ids
,
aid
,
sigs
))
{
CAF_LOG_WARNING
(
"unable to deserialize payload of server handshake:"
CAF_LOG_WARNING
(
"unable to deserialize payload of server handshake:"
<<
ctx
->
system
().
render
(
err
));
<<
ctx
->
system
().
render
(
err
));
return
false
;
return
serializing_basp_payload_failed
;
}
}
// Check the application ID.
// Check the application ID.
auto
whitelist
=
get_or
(
config
(),
"middleman.app-identifiers"
,
auto
whitelist
=
get_or
(
config
(),
"middleman.app-identifiers"
,
...
@@ -321,20 +319,20 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
...
@@ -321,20 +319,20 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if
(
i
==
app_ids
.
end
())
{
if
(
i
==
app_ids
.
end
())
{
CAF_LOG_WARNING
(
"refuse to connect to server due to app ID mismatch:"
CAF_LOG_WARNING
(
"refuse to connect to server due to app ID mismatch:"
<<
CAF_ARG
(
app_ids
)
<<
CAF_ARG
(
whitelist
));
<<
CAF_ARG
(
app_ids
)
<<
CAF_ARG
(
whitelist
));
return
false
;
return
incompatible_application_ids
;
}
}
// Close connection to ourselves immediately after sending client HS.
// Close connection to ourselves immediately after sending client HS.
if
(
source_node
==
this_node_
)
{
if
(
source_node
==
this_node_
)
{
CAF_LOG_DEBUG
(
"close connection to self immediately"
);
CAF_LOG_DEBUG
(
"close connection to self immediately"
);
callee_
.
finalize_handshake
(
source_node
,
aid
,
sigs
);
callee_
.
finalize_handshake
(
source_node
,
aid
,
sigs
);
return
false
;
return
redundant_connection
;
}
}
// Close this connection if we already have a direct connection.
// Close this connection if we already have a direct connection.
if
(
tbl_
.
lookup_direct
(
source_node
))
{
if
(
tbl_
.
lookup_direct
(
source_node
))
{
CAF_LOG_DEBUG
(
CAF_LOG_DEBUG
(
"close redundant direct connection:"
<<
CAF_ARG
(
source_node
));
"close redundant direct connection:"
<<
CAF_ARG
(
source_node
));
callee_
.
finalize_handshake
(
source_node
,
aid
,
sigs
);
callee_
.
finalize_handshake
(
source_node
,
aid
,
sigs
);
return
false
;
return
redundant_connection
;
}
}
// Add direct route to this node and remove any indirect entry.
// Add direct route to this node and remove any indirect entry.
CAF_LOG_DEBUG
(
"new direct connection:"
<<
CAF_ARG
(
source_node
));
CAF_LOG_DEBUG
(
"new direct connection:"
<<
CAF_ARG
(
source_node
));
...
@@ -344,7 +342,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
...
@@ -344,7 +342,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
auto
path
=
tbl_
.
lookup
(
source_node
);
auto
path
=
tbl_
.
lookup
(
source_node
);
if
(
!
path
)
{
if
(
!
path
)
{
CAF_LOG_ERROR
(
"no route to host after server handshake"
);
CAF_LOG_ERROR
(
"no route to host after server handshake"
);
return
fals
e
;
return
no_route_to_receiving_nod
e
;
}
}
write_client_handshake
(
ctx
,
callee_
.
get_buffer
(
path
->
hdl
));
write_client_handshake
(
ctx
,
callee_
.
get_buffer
(
path
->
hdl
));
callee_
.
learned_new_node_directly
(
source_node
,
was_indirect
);
callee_
.
learned_new_node_directly
(
source_node
,
was_indirect
);
...
@@ -359,7 +357,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
...
@@ -359,7 +357,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if
(
auto
err
=
bd
(
source_node
))
{
if
(
auto
err
=
bd
(
source_node
))
{
CAF_LOG_WARNING
(
"unable to deserialize payload of client handshake:"
CAF_LOG_WARNING
(
"unable to deserialize payload of client handshake:"
<<
ctx
->
system
().
render
(
err
));
<<
ctx
->
system
().
render
(
err
));
return
false
;
return
serializing_basp_payload_failed
;
}
}
// Drop repeated handshakes.
// Drop repeated handshakes.
if
(
tbl_
.
lookup_direct
(
source_node
))
{
if
(
tbl_
.
lookup_direct
(
source_node
))
{
...
@@ -383,11 +381,11 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
...
@@ -383,11 +381,11 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
CAF_LOG_WARNING
(
CAF_LOG_WARNING
(
"unable to deserialize source and destination for routed message:"
"unable to deserialize source and destination for routed message:"
<<
ctx
->
system
().
render
(
err
));
<<
ctx
->
system
().
render
(
err
));
return
false
;
return
serializing_basp_payload_failed
;
}
}
if
(
dest_node
!=
this_node_
)
{
if
(
dest_node
!=
this_node_
)
{
forward
(
ctx
,
dest_node
,
hdr
,
*
payload
);
forward
(
ctx
,
dest_node
,
hdr
,
*
payload
);
return
true
;
return
await_header
;
}
}
auto
last_hop
=
tbl_
.
lookup_direct
(
hdl
);
auto
last_hop
=
tbl_
.
lookup_direct
(
hdl
);
if
(
source_node
!=
none
&&
source_node
!=
this_node_
if
(
source_node
!=
none
&&
source_node
!=
this_node_
...
@@ -441,7 +439,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
...
@@ -441,7 +439,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if
(
auto
err
=
bd
(
source_node
,
dest_node
))
{
if
(
auto
err
=
bd
(
source_node
,
dest_node
))
{
CAF_LOG_WARNING
(
"unable to deserialize payload of monitor message:"
CAF_LOG_WARNING
(
"unable to deserialize payload of monitor message:"
<<
ctx
->
system
().
render
(
err
));
<<
ctx
->
system
().
render
(
err
));
return
false
;
return
serializing_basp_payload_failed
;
}
}
if
(
dest_node
==
this_node_
)
if
(
dest_node
==
this_node_
)
callee_
.
proxy_announced
(
source_node
,
hdr
.
dest_actor
);
callee_
.
proxy_announced
(
source_node
,
hdr
.
dest_actor
);
...
@@ -458,7 +456,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
...
@@ -458,7 +456,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if
(
auto
err
=
bd
(
source_node
,
dest_node
,
fail_state
))
{
if
(
auto
err
=
bd
(
source_node
,
dest_node
,
fail_state
))
{
CAF_LOG_WARNING
(
"unable to deserialize payload of down message:"
CAF_LOG_WARNING
(
"unable to deserialize payload of down message:"
<<
ctx
->
system
().
render
(
err
));
<<
ctx
->
system
().
render
(
err
));
return
false
;
return
serializing_basp_payload_failed
;
}
}
if
(
dest_node
==
this_node_
)
{
if
(
dest_node
==
this_node_
)
{
// Delay this message to make sure we don't skip in-flight messages.
// Delay this message to make sure we don't skip in-flight messages.
...
@@ -481,10 +479,10 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
...
@@ -481,10 +479,10 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
}
}
default:
{
default:
{
CAF_LOG_ERROR
(
"invalid operation"
);
CAF_LOG_ERROR
(
"invalid operation"
);
return
fals
e
;
return
malformed_basp_messag
e
;
}
}
}
}
return
true
;
return
await_header
;
}
}
void
instance
::
forward
(
execution_unit
*
ctx
,
const
node_id
&
dest_node
,
void
instance
::
forward
(
execution_unit
*
ctx
,
const
node_id
&
dest_node
,
...
...
libcaf_io/src/io/basp_broker.cpp
View file @
2008a863
...
@@ -127,8 +127,8 @@ behavior basp_broker::make_behavior() {
...
@@ -127,8 +127,8 @@ behavior basp_broker::make_behavior() {
auto
&
ctx
=
*
this_context
;
auto
&
ctx
=
*
this_context
;
auto
next
=
instance
.
handle
(
context
(),
msg
,
ctx
.
hdr
,
auto
next
=
instance
.
handle
(
context
(),
msg
,
ctx
.
hdr
,
ctx
.
cstate
==
basp
::
await_payload
);
ctx
.
cstate
==
basp
::
await_payload
);
if
(
next
==
basp
::
close_connection
)
{
if
(
requires_shutdown
(
next
)
)
{
connection_cleanup
(
msg
.
handle
);
connection_cleanup
(
msg
.
handle
,
to_sec
(
next
)
);
close
(
msg
.
handle
);
close
(
msg
.
handle
);
return
;
return
;
}
}
...
@@ -225,7 +225,9 @@ behavior basp_broker::make_behavior() {
...
@@ -225,7 +225,9 @@ behavior basp_broker::make_behavior() {
delete_atom
::
value
,
msg
.
handle
));
delete_atom
::
value
,
msg
.
handle
));
},
},
// received from the message handler above for connection_closed_msg
// received from the message handler above for connection_closed_msg
[
=
](
delete_atom
,
connection_handle
hdl
)
{
connection_cleanup
(
hdl
);
},
[
=
](
delete_atom
,
connection_handle
hdl
)
{
connection_cleanup
(
hdl
,
sec
::
none
);
},
// 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
(
""
);
...
@@ -561,8 +563,8 @@ void basp_broker::set_context(connection_handle hdl) {
...
@@ -561,8 +563,8 @@ void basp_broker::set_context(connection_handle hdl) {
t_last_hop
=
&
i
->
second
.
id
;
t_last_hop
=
&
i
->
second
.
id
;
}
}
void
basp_broker
::
connection_cleanup
(
connection_handle
hdl
)
{
void
basp_broker
::
connection_cleanup
(
connection_handle
hdl
,
sec
code
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
hdl
));
CAF_LOG_TRACE
(
CAF_ARG
(
hdl
)
<<
CAF_ARG
(
code
)
);
// Remove handle from the routing table and clean up any node-specific state
// Remove handle from the routing table and clean up any node-specific state
// we might still have.
// we might still have.
if
(
auto
nid
=
instance
.
tbl
().
erase_direct
(
hdl
))
if
(
auto
nid
=
instance
.
tbl
().
erase_direct
(
hdl
))
...
@@ -574,8 +576,9 @@ void basp_broker::connection_cleanup(connection_handle hdl) {
...
@@ -574,8 +576,9 @@ void basp_broker::connection_cleanup(connection_handle hdl) {
auto
&
ref
=
i
->
second
;
auto
&
ref
=
i
->
second
;
CAF_ASSERT
(
i
->
first
==
ref
.
hdl
);
CAF_ASSERT
(
i
->
first
==
ref
.
hdl
);
if
(
ref
.
callback
)
{
if
(
ref
.
callback
)
{
CAF_LOG_DEBUG
(
"connection closed during handshake"
);
CAF_LOG_DEBUG
(
"connection closed during handshake:"
<<
CAF_ARG
(
code
));
ref
.
callback
->
deliver
(
sec
::
disconnect_during_handshake
);
auto
x
=
code
!=
sec
::
none
?
code
:
sec
::
disconnect_during_handshake
;
ref
.
callback
->
deliver
(
x
);
}
}
ctx
.
erase
(
i
);
ctx
.
erase
(
i
);
}
}
...
...
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