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
24571158
Commit
24571158
authored
Jun 22, 2016
by
ufownl
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Add app identifier config parameter
parent
0f810ffb
Changes
7
Hide whitespace changes
Inline
Side-by-side
Showing
7 changed files
with
63 additions
and
34 deletions
+63
-34
examples/caf-application.ini
examples/caf-application.ini
+2
-0
libcaf_core/caf/actor_system_config.hpp
libcaf_core/caf/actor_system_config.hpp
+1
-0
libcaf_core/src/actor_system_config.cpp
libcaf_core/src/actor_system_config.cpp
+2
-0
libcaf_io/src/header.cpp
libcaf_io/src/header.cpp
+0
-1
libcaf_io/src/instance.cpp
libcaf_io/src/instance.cpp
+29
-1
libcaf_io/src/middleman_actor.cpp
libcaf_io/src/middleman_actor.cpp
+3
-2
libcaf_io/test/basp.cpp
libcaf_io/test/basp.cpp
+26
-30
No files found.
examples/caf-application.ini
View file @
24571158
...
@@ -39,6 +39,8 @@ relaxed-sleep-duration=10000
...
@@ -39,6 +39,8 @@ relaxed-sleep-duration=10000
enable-automatic-connections
=
false
enable-automatic-connections
=
false
; accepted alternative: 'asio' (only when compiling CAF with ASIO)
; accepted alternative: 'asio' (only when compiling CAF with ASIO)
network-backend
=
'default'
network-backend
=
'default'
; application identifier of this node
app-identifier
=
''
; maximum number of consecutive I/O reads per broker
; maximum number of consecutive I/O reads per broker
max-consecutive-reads
=
50
max-consecutive-reads
=
50
; heartbeat message interval in ms (0 disables heartbeating)
; heartbeat message interval in ms (0 disables heartbeating)
...
...
libcaf_core/caf/actor_system_config.hpp
View file @
24571158
...
@@ -242,6 +242,7 @@ public:
...
@@ -242,6 +242,7 @@ public:
// Config parameters of middleman.
// Config parameters of middleman.
atom_value
middleman_network_backend
;
atom_value
middleman_network_backend
;
std
::
string
middleman_app_identifier
;
bool
middleman_enable_automatic_connections
;
bool
middleman_enable_automatic_connections
;
size_t
middleman_max_consecutive_reads
;
size_t
middleman_max_consecutive_reads
;
size_t
middleman_heartbeat_interval
;
size_t
middleman_heartbeat_interval
;
...
...
libcaf_core/src/actor_system_config.cpp
View file @
24571158
...
@@ -131,6 +131,8 @@ actor_system_config::actor_system_config()
...
@@ -131,6 +131,8 @@ actor_system_config::actor_system_config()
opt_group
{
options_
,
"middleman"
}
opt_group
{
options_
,
"middleman"
}
.
add
(
middleman_network_backend
,
"network-backend"
,
.
add
(
middleman_network_backend
,
"network-backend"
,
"sets the network backend to either 'default' or 'asio' (if available)"
)
"sets the network backend to either 'default' or 'asio' (if available)"
)
.
add
(
middleman_app_identifier
,
"app-identifier"
,
"sets the application identifier of this node"
)
.
add
(
middleman_enable_automatic_connections
,
"enable-automatic-connections"
,
.
add
(
middleman_enable_automatic_connections
,
"enable-automatic-connections"
,
"enables or disables automatic connection management (off per default)"
)
"enables or disables automatic connection management (off per default)"
)
.
add
(
middleman_max_consecutive_reads
,
"max-consecutive-reads"
,
.
add
(
middleman_max_consecutive_reads
,
"max-consecutive-reads"
,
...
...
libcaf_io/src/header.cpp
View file @
24571158
...
@@ -84,7 +84,6 @@ bool client_handshake_valid(const header& hdr) {
...
@@ -84,7 +84,6 @@ bool client_handshake_valid(const header& hdr) {
&&
hdr
.
source_node
!=
hdr
.
dest_node
&&
hdr
.
source_node
!=
hdr
.
dest_node
&&
zero
(
hdr
.
source_actor
)
&&
zero
(
hdr
.
source_actor
)
&&
zero
(
hdr
.
dest_actor
)
&&
zero
(
hdr
.
dest_actor
)
&&
zero
(
hdr
.
payload_len
)
&&
zero
(
hdr
.
operation_data
);
&&
zero
(
hdr
.
operation_data
);
}
}
...
...
libcaf_io/src/instance.cpp
View file @
24571158
...
@@ -22,6 +22,7 @@
...
@@ -22,6 +22,7 @@
#include "caf/streambuf.hpp"
#include "caf/streambuf.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/io/basp/version.hpp"
#include "caf/io/basp/version.hpp"
...
@@ -116,7 +117,16 @@ connection_state instance::handle(execution_unit* ctx,
...
@@ -116,7 +117,16 @@ connection_state instance::handle(execution_unit* ctx,
std
::
set
<
std
::
string
>
sigs
;
std
::
set
<
std
::
string
>
sigs
;
if
(
payload_valid
())
{
if
(
payload_valid
())
{
binary_deserializer
bd
{
ctx
,
*
payload
};
binary_deserializer
bd
{
ctx
,
*
payload
};
std
::
string
remote_appid
;
bd
>>
remote_appid
;
if
(
remote_appid
!=
callee_
.
system
().
config
().
middleman_app_identifier
)
{
CAF_LOG_ERROR
(
"app identifier mismatch"
);
return
err
();
}
bd
>>
aid
>>
sigs
;
bd
>>
aid
>>
sigs
;
}
else
{
CAF_LOG_ERROR
(
"fail to receive the app identifier"
);
return
err
();
}
}
// close self connection after handshake is done
// close self connection after handshake is done
if
(
hdr
.
source_node
==
this_node_
)
{
if
(
hdr
.
source_node
==
this_node_
)
{
...
@@ -153,6 +163,18 @@ connection_state instance::handle(execution_unit* ctx,
...
@@ -153,6 +163,18 @@ connection_state instance::handle(execution_unit* ctx,
<<
CAF_ARG
(
hdr
.
source_node
));
<<
CAF_ARG
(
hdr
.
source_node
));
break
;
break
;
}
}
if
(
payload_valid
())
{
binary_deserializer
bd
{
ctx
,
*
payload
};
std
::
string
remote_appid
;
bd
>>
remote_appid
;
if
(
remote_appid
!=
callee_
.
system
().
config
().
middleman_app_identifier
)
{
CAF_LOG_ERROR
(
"app identifier mismatch"
);
return
err
();
}
}
else
{
CAF_LOG_ERROR
(
"fail to receive the app identifier"
);
return
err
();
}
// add direct route to this node and remove any indirect entry
// add direct route to this node and remove any indirect entry
CAF_LOG_INFO
(
"new direct connection:"
<<
CAF_ARG
(
hdr
.
source_node
));
CAF_LOG_INFO
(
"new direct connection:"
<<
CAF_ARG
(
hdr
.
source_node
));
tbl_
.
add_direct
(
dm
.
handle
,
hdr
.
source_node
);
tbl_
.
add_direct
(
dm
.
handle
,
hdr
.
source_node
);
...
@@ -366,9 +388,12 @@ void instance::write_server_handshake(execution_unit* ctx,
...
@@ -366,9 +388,12 @@ void instance::write_server_handshake(execution_unit* ctx,
}
}
CAF_LOG_DEBUG_IF
(
!
pa
&&
port
,
"no actor published"
);
CAF_LOG_DEBUG_IF
(
!
pa
&&
port
,
"no actor published"
);
auto
writer
=
make_callback
([
&
](
serializer
&
sink
)
{
auto
writer
=
make_callback
([
&
](
serializer
&
sink
)
{
sink
<<
callee_
.
system
().
config
().
middleman_app_identifier
;
if
(
pa
)
{
if
(
pa
)
{
auto
i
=
pa
->
first
?
pa
->
first
->
id
()
:
invalid_actor_id
;
auto
i
=
pa
->
first
?
pa
->
first
->
id
()
:
invalid_actor_id
;
sink
<<
i
<<
pa
->
second
;
sink
<<
i
<<
pa
->
second
;
}
else
{
sink
<<
invalid_actor_id
<<
std
::
set
<
std
::
string
>
{};
}
}
});
});
header
hdr
{
message_type
::
server_handshake
,
0
,
0
,
version
,
header
hdr
{
message_type
::
server_handshake
,
0
,
0
,
version
,
...
@@ -382,9 +407,12 @@ void instance::write_client_handshake(execution_unit* ctx,
...
@@ -382,9 +407,12 @@ void instance::write_client_handshake(execution_unit* ctx,
buffer_type
&
buf
,
buffer_type
&
buf
,
const
node_id
&
remote_side
)
{
const
node_id
&
remote_side
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
remote_side
));
CAF_LOG_TRACE
(
CAF_ARG
(
remote_side
));
auto
writer
=
make_callback
([
&
](
serializer
&
sink
)
{
sink
<<
callee_
.
system
().
config
().
middleman_app_identifier
;
});
header
hdr
{
message_type
::
client_handshake
,
0
,
0
,
0
,
header
hdr
{
message_type
::
client_handshake
,
0
,
0
,
0
,
this_node_
,
remote_side
,
invalid_actor_id
,
invalid_actor_id
};
this_node_
,
remote_side
,
invalid_actor_id
,
invalid_actor_id
};
write
(
ctx
,
buf
,
hdr
);
write
(
ctx
,
buf
,
hdr
,
&
writer
);
}
}
void
instance
::
write_announce_proxy
(
execution_unit
*
ctx
,
buffer_type
&
buf
,
void
instance
::
write_announce_proxy
(
execution_unit
*
ctx
,
buffer_type
&
buf
,
...
...
libcaf_io/src/middleman_actor.cpp
View file @
24571158
...
@@ -131,9 +131,10 @@ public:
...
@@ -131,9 +131,10 @@ public:
auto
i
=
pending_
.
find
(
key
);
auto
i
=
pending_
.
find
(
key
);
if
(
i
==
pending_
.
end
())
if
(
i
==
pending_
.
end
())
return
;
return
;
monitor
(
addr
);
if
(
nid
&&
addr
)
{
if
(
nid
&&
addr
)
monitor
(
addr
);
cached_
.
emplace
(
key
,
std
::
make_tuple
(
nid
,
addr
,
sigs
));
cached_
.
emplace
(
key
,
std
::
make_tuple
(
nid
,
addr
,
sigs
));
}
auto
res
=
make_message
(
ok_atom
::
value
,
std
::
move
(
nid
),
auto
res
=
make_message
(
ok_atom
::
value
,
std
::
move
(
nid
),
std
::
move
(
addr
),
std
::
move
(
sigs
));
std
::
move
(
addr
),
std
::
move
(
sigs
));
for
(
auto
&
promise
:
i
->
second
)
for
(
auto
&
promise
:
i
->
second
)
...
...
libcaf_io/test/basp.cpp
View file @
24571158
...
@@ -287,33 +287,27 @@ public:
...
@@ -287,33 +287,27 @@ public:
mpx_
->
accept_connection
(
src
);
mpx_
->
accept_connection
(
src
);
// technically, the server handshake arrives
// technically, the server handshake arrives
// before we send the client handshake
// before we send the client handshake
auto
m
=
mock
(
hdl
,
mock
(
hdl
,
{
basp
::
message_type
::
client_handshake
,
0
,
0
,
0
,
{
basp
::
message_type
::
client_handshake
,
0
,
0
,
0
,
remote_node
(
i
),
this_node
(),
remote_node
(
i
),
this_node
(),
invalid_actor_id
,
invalid_actor_id
});
invalid_actor_id
,
invalid_actor_id
},
std
::
string
{})
if
(
published_actor_id
!=
invalid_actor_id
)
.
expect
(
hdl
,
m
.
expect
(
hdl
,
basp
::
message_type
::
server_handshake
,
no_flags
,
basp
::
message_type
::
server_handshake
,
no_flags
,
any_vals
,
basp
::
version
,
this_node
(),
node_id
{
invalid_node_id
},
any_vals
,
basp
::
version
,
this_node
(),
node_id
{
invalid_node_id
},
published_actor_id
,
invalid_actor_id
,
std
::
string
{},
published_actor_id
,
invalid_actor_id
,
published_actor_id
,
published_actor_id
,
published_actor_ifs
)
published_actor_ifs
);
else
m
.
expect
(
hdl
,
basp
::
message_type
::
server_handshake
,
no_flags
,
any_vals
,
basp
::
version
,
this_node
(),
node_id
{
invalid_node_id
},
invalid_actor_id
,
invalid_actor_id
);
// upon receiving our client handshake, BASP will check
// upon receiving our client handshake, BASP will check
// whether there is a SpawnServ actor on this node
// whether there is a SpawnServ actor on this node
m
.
expect
(
hdl
,
.
expect
(
hdl
,
basp
::
message_type
::
dispatch_message
,
basp
::
message_type
::
dispatch_message
,
basp
::
header
::
named_receiver_flag
,
any_vals
,
basp
::
header
::
named_receiver_flag
,
any_vals
,
no_operation_data
,
no_operation_data
,
this_node
(),
remote_node
(
i
),
this_node
(),
remote_node
(
i
),
any_vals
,
invalid_actor_id
,
any_vals
,
invalid_actor_id
,
spawn_serv_atom
,
spawn_serv_atom
,
std
::
vector
<
actor_addr
>
{},
std
::
vector
<
actor_addr
>
{},
make_message
(
sys_atom
::
value
,
get_atom
::
value
,
"info"
));
make_message
(
sys_atom
::
value
,
get_atom
::
value
,
"info"
));
// test whether basp instance correctly updates the
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
// routing table upon receiving client handshakes
auto
path
=
tbl
().
lookup
(
remote_node
(
i
));
auto
path
=
tbl
().
lookup
(
remote_node
(
i
));
...
@@ -489,7 +483,7 @@ CAF_TEST(non_empty_server_handshake) {
...
@@ -489,7 +483,7 @@ CAF_TEST(non_empty_server_handshake) {
basp
::
header
expected
{
basp
::
message_type
::
server_handshake
,
0
,
0
,
basp
::
header
expected
{
basp
::
message_type
::
server_handshake
,
0
,
0
,
basp
::
version
,
this_node
(),
invalid_node_id
,
basp
::
version
,
this_node
(),
invalid_node_id
,
self
()
->
id
(),
invalid_actor_id
};
self
()
->
id
(),
invalid_actor_id
};
to_buf
(
expected_buf
,
expected
,
nullptr
,
to_buf
(
expected_buf
,
expected
,
nullptr
,
std
::
string
{},
self
()
->
id
(),
set
<
string
>
{
"caf::replies_to<@u16>::with<@u16>"
});
self
()
->
id
(),
set
<
string
>
{
"caf::replies_to<@u16>::with<@u16>"
});
CAF_CHECK_EQUAL
(
hexstr
(
buf
),
hexstr
(
expected_buf
));
CAF_CHECK_EQUAL
(
hexstr
(
buf
),
hexstr
(
expected_buf
));
}
}
...
@@ -595,12 +589,13 @@ CAF_TEST(remote_actor_and_send) {
...
@@ -595,12 +589,13 @@ CAF_TEST(remote_actor_and_send) {
{
basp
::
message_type
::
server_handshake
,
0
,
0
,
basp
::
version
,
{
basp
::
message_type
::
server_handshake
,
0
,
0
,
basp
::
version
,
remote_node
(
0
),
invalid_node_id
,
remote_node
(
0
),
invalid_node_id
,
pseudo_remote
(
0
)
->
id
(),
invalid_actor_id
},
pseudo_remote
(
0
)
->
id
(),
invalid_actor_id
},
std
::
string
{},
pseudo_remote
(
0
)
->
id
(),
pseudo_remote
(
0
)
->
id
(),
uint32_t
{
0
})
uint32_t
{
0
})
.
expect
(
remote_hdl
(
0
),
.
expect
(
remote_hdl
(
0
),
basp
::
message_type
::
client_handshake
,
no_flags
,
no_payload
,
basp
::
message_type
::
client_handshake
,
no_flags
,
1u
,
no_operation_data
,
this_node
(),
remote_node
(
0
),
no_operation_data
,
this_node
(),
remote_node
(
0
),
invalid_actor_id
,
invalid_actor_id
)
invalid_actor_id
,
invalid_actor_id
,
std
::
string
{}
)
.
expect
(
remote_hdl
(
0
),
.
expect
(
remote_hdl
(
0
),
basp
::
message_type
::
dispatch_message
,
basp
::
message_type
::
dispatch_message
,
basp
::
header
::
named_receiver_flag
,
any_vals
,
basp
::
header
::
named_receiver_flag
,
any_vals
,
...
@@ -823,12 +818,13 @@ CAF_TEST(automatic_connection) {
...
@@ -823,12 +818,13 @@ CAF_TEST(automatic_connection) {
{
basp
::
message_type
::
server_handshake
,
0
,
0
,
basp
::
version
,
{
basp
::
message_type
::
server_handshake
,
0
,
0
,
basp
::
version
,
remote_node
(
0
),
invalid_node_id
,
remote_node
(
0
),
invalid_node_id
,
pseudo_remote
(
0
)
->
id
(),
invalid_actor_id
},
pseudo_remote
(
0
)
->
id
(),
invalid_actor_id
},
std
::
string
{},
pseudo_remote
(
0
)
->
id
(),
pseudo_remote
(
0
)
->
id
(),
uint32_t
{
0
})
uint32_t
{
0
})
.
expect
(
remote_hdl
(
0
),
.
expect
(
remote_hdl
(
0
),
basp
::
message_type
::
client_handshake
,
no_flags
,
no_payload
,
basp
::
message_type
::
client_handshake
,
no_flags
,
1u
,
no_operation_data
,
this_node
(),
remote_node
(
0
),
no_operation_data
,
this_node
(),
remote_node
(
0
),
invalid_actor_id
,
invalid_actor_id
);
invalid_actor_id
,
invalid_actor_id
,
std
::
string
{}
);
CAF_CHECK_EQUAL
(
tbl
().
lookup_indirect
(
remote_node
(
0
)),
invalid_node_id
);
CAF_CHECK_EQUAL
(
tbl
().
lookup_indirect
(
remote_node
(
0
)),
invalid_node_id
);
CAF_CHECK_EQUAL
(
tbl
().
lookup_indirect
(
remote_node
(
1
)),
invalid_node_id
);
CAF_CHECK_EQUAL
(
tbl
().
lookup_indirect
(
remote_node
(
1
)),
invalid_node_id
);
CAF_CHECK_EQUAL
(
tbl
().
lookup_direct
(
remote_node
(
0
)).
id
(),
remote_hdl
(
0
).
id
());
CAF_CHECK_EQUAL
(
tbl
().
lookup_direct
(
remote_node
(
0
)).
id
(),
remote_hdl
(
0
).
id
());
...
...
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