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
ba5cd462
Commit
ba5cd462
authored
Jun 06, 2016
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Fix shadowing warnings
parent
69f75141
Changes
22
Hide whitespace changes
Inline
Side-by-side
Showing
22 changed files
with
124 additions
and
132 deletions
+124
-132
libcaf_core/caf/monitorable_actor.hpp
libcaf_core/caf/monitorable_actor.hpp
+3
-3
libcaf_core/caf/optional.hpp
libcaf_core/caf/optional.hpp
+5
-5
libcaf_core/caf/scheduler/coordinator.hpp
libcaf_core/caf/scheduler/coordinator.hpp
+2
-2
libcaf_core/caf/scheduler/profiled_coordinator.hpp
libcaf_core/caf/scheduler/profiled_coordinator.hpp
+4
-4
libcaf_core/src/actor_companion.cpp
libcaf_core/src/actor_companion.cpp
+2
-2
libcaf_core/src/actor_system_config.cpp
libcaf_core/src/actor_system_config.cpp
+7
-8
libcaf_core/src/blocking_actor.cpp
libcaf_core/src/blocking_actor.cpp
+2
-2
libcaf_core/src/config_option.cpp
libcaf_core/src/config_option.cpp
+4
-5
libcaf_core/src/group_manager.cpp
libcaf_core/src/group_manager.cpp
+3
-3
libcaf_core/src/local_actor.cpp
libcaf_core/src/local_actor.cpp
+7
-7
libcaf_core/src/logger.cpp
libcaf_core/src/logger.cpp
+5
-5
libcaf_core/src/monitorable_actor.cpp
libcaf_core/src/monitorable_actor.cpp
+10
-10
libcaf_core/src/response_promise.cpp
libcaf_core/src/response_promise.cpp
+2
-2
libcaf_core/test/actor_termination.cpp
libcaf_core/test/actor_termination.cpp
+5
-5
libcaf_core/test/request_response.cpp
libcaf_core/test/request_response.cpp
+10
-11
libcaf_core/test/typed_spawn.cpp
libcaf_core/test/typed_spawn.cpp
+6
-9
libcaf_io/caf/io/middleman.hpp
libcaf_io/caf/io/middleman.hpp
+6
-6
libcaf_io/src/abstract_broker.cpp
libcaf_io/src/abstract_broker.cpp
+4
-4
libcaf_io/src/basp_broker.cpp
libcaf_io/src/basp_broker.cpp
+22
-22
libcaf_io/src/default_multiplexer.cpp
libcaf_io/src/default_multiplexer.cpp
+6
-6
libcaf_io/src/test_multiplexer.cpp
libcaf_io/src/test_multiplexer.cpp
+2
-2
libcaf_test/caf/test/unit_test_impl.hpp
libcaf_test/caf/test/unit_test_impl.hpp
+7
-9
No files found.
libcaf_core/caf/monitorable_actor.hpp
View file @
ba5cd462
...
...
@@ -152,14 +152,14 @@ protected:
// handles `exit_msg`, `sys_atom` messages, and additionally `down_msg`
// with `down_msg_handler`; returns true if the message is handled
template
<
class
F
>
bool
handle_system_message
(
mailbox_element
&
node
,
execution_unit
*
context
,
bool
handle_system_message
(
mailbox_element
&
x
,
execution_unit
*
context
,
bool
trap_exit
,
F
&
down_msg_handler
)
{
auto
&
msg
=
node
.
msg
;
auto
&
msg
=
x
.
msg
;
if
(
msg
.
size
()
==
1
&&
msg
.
match_element
<
down_msg
>
(
0
))
{
down_msg_handler
(
msg
.
get_as_mutable
<
down_msg
>
(
0
));
return
true
;
}
return
handle_system_message
(
node
,
context
,
trap_exit
);
return
handle_system_message
(
x
,
context
,
trap_exit
);
}
// Calls `fun` with exclusive access to an actor's state.
...
...
libcaf_core/caf/optional.hpp
View file @
ba5cd462
...
...
@@ -49,8 +49,8 @@ class optional {
class
E
=
typename
std
::
enable_if
<
std
::
is_convertible
<
U
,
T
>
::
value
>::
type
>
optional
(
U
value
)
:
m_valid
(
false
)
{
cr
(
std
::
move
(
value
));
optional
(
U
x
)
:
m_valid
(
false
)
{
cr
(
std
::
move
(
x
));
}
optional
(
const
optional
&
other
)
:
m_valid
(
false
)
{
...
...
@@ -151,10 +151,10 @@ class optional {
}
template
<
class
V
>
void
cr
(
V
&&
value
)
{
void
cr
(
V
&&
x
)
{
CAF_ASSERT
(
!
m_valid
);
m_valid
=
true
;
new
(
&
m_value
)
T
(
std
::
forward
<
V
>
(
value
));
new
(
&
m_value
)
T
(
std
::
forward
<
V
>
(
x
));
}
bool
m_valid
;
...
...
@@ -172,7 +172,7 @@ class optional<T&> {
// nop
}
optional
(
T
&
value
)
:
m_value
(
&
value
)
{
optional
(
T
&
x
)
:
m_value
(
&
x
)
{
// nop
}
...
...
libcaf_core/caf/scheduler/coordinator.hpp
View file @
ba5cd462
...
...
@@ -47,8 +47,8 @@ public:
using
worker_type
=
worker
<
Policy
>
;
worker_type
*
worker_by_id
(
size_t
id
)
{
//override
{
return
workers_
[
id
].
get
();
worker_type
*
worker_by_id
(
size_t
x
)
{
return
workers_
[
x
].
get
();
}
policy_data
&
data
()
{
...
...
libcaf_core/caf/scheduler/profiled_coordinator.hpp
View file @
ba5cd462
...
...
@@ -27,7 +27,7 @@
#elif defined(CAF_WINDOWS)
#include <Windows.h>
#include <Psapi.h>
#else
#else
#include <sys/resource.h>
#endif
...
...
@@ -104,7 +104,7 @@ public:
GetProcessTimes
(
GetCurrentProcess
(),
&
creation_time
,
&
exit_time
,
&
kernel_time
,
&
user_time
);
GetProcessMemoryInfo
(
GetCurrentProcess
(),
&
pmc
,
sizeof
(
pmc
));
m
.
mem
=
pmc
.
PeakWorkingSetSize
/
1024
;
...
...
@@ -256,11 +256,11 @@ public:
}
template
<
class
Time
,
class
Label
>
void
record
(
Time
t
,
Label
label
,
size_t
id
,
const
measurement
&
m
)
{
void
record
(
Time
t
,
Label
label
,
size_t
rec_
id
,
const
measurement
&
m
)
{
using
std
::
setw
;
file_
<<
setw
(
21
)
<<
t
.
time_since_epoch
().
count
()
<<
setw
(
10
)
<<
label
<<
setw
(
10
)
<<
id
<<
setw
(
10
)
<<
rec_
id
<<
m
<<
'\n'
;
}
...
...
libcaf_core/src/actor_companion.cpp
View file @
ba5cd462
...
...
@@ -43,10 +43,10 @@ void actor_companion::enqueue(mailbox_element_ptr ptr, execution_unit*) {
on_enqueue_
(
std
::
move
(
ptr
));
}
void
actor_companion
::
enqueue
(
strong_actor_ptr
s
ender
,
message_id
mid
,
void
actor_companion
::
enqueue
(
strong_actor_ptr
s
rc
,
message_id
mid
,
message
content
,
execution_unit
*
eu
)
{
using
detail
::
memory
;
auto
ptr
=
mailbox_element
::
make
(
s
ender
,
mid
,
{},
std
::
move
(
content
));
auto
ptr
=
mailbox_element
::
make
(
s
td
::
move
(
src
)
,
mid
,
{},
std
::
move
(
content
));
enqueue
(
std
::
move
(
ptr
),
eu
);
}
...
...
libcaf_core/src/actor_system_config.cpp
View file @
ba5cd462
...
...
@@ -272,17 +272,16 @@ actor_system_config::add_error_category(atom_value x, error_renderer y) {
return
*
this
;
}
actor_system_config
&
actor_system_config
::
set
(
const
char
*
config_name
,
config_value
config_value
)
{
std
::
string
cn
;
actor_system_config
&
actor_system_config
::
set
(
const
char
*
cn
,
config_value
cv
)
{
std
::
string
full_name
;
for
(
auto
&
x
:
options_
)
{
// config_name has format "$category.$name"
cn
=
x
->
category
();
cn
+=
'.'
;
cn
+=
x
->
name
();
if
(
cn
==
config_name
)
{
full_name
=
x
->
category
();
full_name
+=
'.'
;
full_name
+=
x
->
name
();
if
(
full_name
==
cn
)
{
auto
f
=
x
->
to_sink
();
f
(
0
,
c
onfig_value
);
f
(
0
,
c
v
);
}
}
return
*
this
;
...
...
libcaf_core/src/blocking_actor.cpp
View file @
ba5cd462
...
...
@@ -39,12 +39,12 @@ blocking_actor::~blocking_actor() {
void
blocking_actor
::
enqueue
(
mailbox_element_ptr
ptr
,
execution_unit
*
)
{
auto
mid
=
ptr
->
mid
;
auto
s
ender
=
ptr
->
sender
;
auto
s
rc
=
ptr
->
sender
;
// returns false if mailbox has been closed
if
(
!
mailbox
().
synchronized_enqueue
(
mtx_
,
cv_
,
ptr
.
release
()))
{
if
(
mid
.
is_request
())
{
detail
::
sync_request_bouncer
srb
{
exit_reason
()};
srb
(
s
ender
,
mid
);
srb
(
s
rc
,
mid
);
}
}
}
...
...
libcaf_core/src/config_option.cpp
View file @
ba5cd462
...
...
@@ -23,11 +23,10 @@
namespace
caf
{
config_option
::
config_option
(
const
char
*
category
,
const
char
*
name
,
const
char
*
explanation
)
:
category_
(
category
),
name_
(
name
),
explanation_
(
explanation
),
config_option
::
config_option
(
const
char
*
cat
,
const
char
*
nm
,
const
char
*
expl
)
:
category_
(
cat
),
name_
(
nm
),
explanation_
(
expl
),
short_name_
(
'\0'
)
{
auto
last
=
name_
.
end
();
auto
comma
=
std
::
find
(
name_
.
begin
(),
last
,
','
);
...
...
libcaf_core/src/group_manager.cpp
View file @
ba5cd462
...
...
@@ -198,11 +198,11 @@ public:
private:
void
send_to_acquaintances
(
const
message
&
what
)
{
// send to all remote subscribers
auto
s
ender
=
current_element_
->
sender
;
auto
s
rc
=
current_element_
->
sender
;
CAF_LOG_DEBUG
(
CAF_ARG
(
acquaintances_
.
size
())
<<
CAF_ARG
(
s
ender
)
<<
CAF_ARG
(
what
));
<<
CAF_ARG
(
s
rc
)
<<
CAF_ARG
(
what
));
for
(
auto
&
acquaintance
:
acquaintances_
)
acquaintance
->
enqueue
(
s
ender
,
invalid_message_id
,
what
,
context
());
acquaintance
->
enqueue
(
s
rc
,
invalid_message_id
,
what
,
context
());
}
local_group_ptr
group_
;
...
...
libcaf_core/src/local_actor.cpp
View file @
ba5cd462
...
...
@@ -319,9 +319,9 @@ uint32_t local_actor::active_timeout_id() const {
return
timeout_id_
;
}
local_actor
::
msg_type
local_actor
::
filter_msg
(
mailbox_element
&
node
)
{
message
&
msg
=
node
.
msg
;
auto
mid
=
node
.
mid
;
local_actor
::
msg_type
local_actor
::
filter_msg
(
mailbox_element
&
x
)
{
message
&
msg
=
x
.
msg
;
auto
mid
=
x
.
mid
;
if
(
mid
.
is_response
())
return
msg_type
::
response
;
switch
(
msg
.
type_token
())
{
...
...
@@ -331,14 +331,14 @@ local_actor::msg_type local_actor::filter_msg(mailbox_element& node) {
auto
&
what
=
msg
.
get_as
<
std
::
string
>
(
2
);
if
(
what
==
"info"
)
{
CAF_LOG_DEBUG
(
"reply to 'info' message"
);
node
.
sender
->
enqueue
(
mailbox_element
::
make
(
ctrl
(),
node
.
mid
.
response_id
(),
x
.
sender
->
enqueue
(
mailbox_element
::
make
(
ctrl
(),
x
.
mid
.
response_id
(),
{},
ok_atom
::
value
,
std
::
move
(
what
),
address
(),
name
()),
context
());
}
else
{
node
.
sender
->
enqueue
(
mailbox_element
::
make
(
ctrl
(),
node
.
mid
.
response_id
(),
x
.
sender
->
enqueue
(
mailbox_element
::
make
(
ctrl
(),
x
.
mid
.
response_id
(),
{},
sec
::
unsupported_sys_key
),
context
());
}
...
...
libcaf_core/src/logger.cpp
View file @
ba5cd462
...
...
@@ -285,18 +285,18 @@ void logger::run() {
<<
"_"
<<
to_string
(
system_
.
node
())
<<
".log"
;
std
::
fstream
out
(
fname
.
str
().
c_str
(),
std
::
ios
::
out
|
std
::
ios
::
app
);
std
::
unique_ptr
<
event
>
event
;
std
::
unique_ptr
<
event
>
ptr
;
for
(;;)
{
// make sure we have data to read
queue_
.
synchronized_await
(
queue_mtx_
,
queue_cv_
);
// read & process event
event
.
reset
(
queue_
.
try_pop
());
CAF_ASSERT
(
event
!=
nullptr
);
if
(
event
->
msg
.
empty
())
{
ptr
.
reset
(
queue_
.
try_pop
());
CAF_ASSERT
(
ptr
!=
nullptr
);
if
(
ptr
->
msg
.
empty
())
{
out
.
close
();
return
;
}
out
<<
event
->
msg
<<
std
::
flush
;
out
<<
ptr
->
msg
<<
std
::
flush
;
}
}
...
...
libcaf_core/src/monitorable_actor.cpp
View file @
ba5cd462
...
...
@@ -237,18 +237,18 @@ size_t monitorable_actor::detach_impl(const attachable::token& what,
return
detach_impl
(
what
,
ptr
->
next
,
stop_on_hit
,
dry_run
);
}
bool
monitorable_actor
::
handle_system_message
(
mailbox_element
&
node
,
execution_unit
*
c
ontext
,
bool
monitorable_actor
::
handle_system_message
(
mailbox_element
&
x
,
execution_unit
*
c
tx
,
bool
trap_exit
)
{
auto
&
msg
=
node
.
msg
;
auto
&
msg
=
x
.
msg
;
if
(
!
trap_exit
&&
msg
.
size
()
==
1
&&
msg
.
match_element
<
exit_msg
>
(
0
))
{
// exits for non-normal exit reasons
auto
&
em
=
msg
.
get_as_mutable
<
exit_msg
>
(
0
);
if
(
em
.
reason
)
cleanup
(
std
::
move
(
em
.
reason
),
c
ontext
);
cleanup
(
std
::
move
(
em
.
reason
),
c
tx
);
return
true
;
}
else
if
(
msg
.
size
()
>
1
&&
msg
.
match_element
<
sys_atom
>
(
0
))
{
if
(
!
node
.
sender
)
if
(
!
x
.
sender
)
return
true
;
error
err
;
mailbox_element_ptr
res
;
...
...
@@ -259,20 +259,20 @@ bool monitorable_actor::handle_system_message(mailbox_element& node,
err
=
sec
::
unsupported_sys_key
;
return
;
}
res
=
mailbox_element
::
make
(
ctrl
(),
node
.
mid
.
response_id
(),
{},
res
=
mailbox_element
::
make
(
ctrl
(),
x
.
mid
.
response_id
(),
{},
ok_atom
::
value
,
std
::
move
(
what
),
address
(),
name
());
}
});
if
(
!
res
&&
!
err
)
err
=
sec
::
unsupported_sys_message
;
if
(
err
&&
node
.
mid
.
is_request
())
res
=
mailbox_element
::
make
(
ctrl
(),
node
.
mid
.
response_id
(),
if
(
err
&&
x
.
mid
.
is_request
())
res
=
mailbox_element
::
make
(
ctrl
(),
x
.
mid
.
response_id
(),
{},
std
::
move
(
err
));
if
(
res
)
{
auto
s
=
actor_cast
<
strong_actor_ptr
>
(
node
.
sender
);
auto
s
=
actor_cast
<
strong_actor_ptr
>
(
x
.
sender
);
if
(
s
)
s
->
enqueue
(
std
::
move
(
res
),
c
ontext
);
s
->
enqueue
(
std
::
move
(
res
),
c
tx
);
}
return
true
;
}
...
...
libcaf_core/src/response_promise.cpp
View file @
ba5cd462
...
...
@@ -30,8 +30,8 @@ response_promise::response_promise()
// nop
}
response_promise
::
response_promise
(
local_actor
*
self
,
mailbox_element
&
src
)
:
self_
(
self
),
response_promise
::
response_promise
(
local_actor
*
ptr
,
mailbox_element
&
src
)
:
self_
(
ptr
),
id_
(
src
.
mid
)
{
// form an invalid request promise when initialized from a
// response ID, since CAF always drops messages in this case
...
...
libcaf_core/test/actor_termination.cpp
View file @
ba5cd462
...
...
@@ -38,26 +38,26 @@ behavior mirror_impl(event_based_actor* self) {
struct
fixture
{
actor_system
system
;
scoped_actor
self
;
scoped_actor
s
coped_s
elf
;
actor
mirror
;
actor
testee
;
fixture
()
:
self
(
system
),
:
s
coped_s
elf
(
system
),
mirror
(
system
.
spawn
(
mirror_impl
)),
testee
(
unsafe_actor_handle_init
)
{
self
->
set_down_handler
([](
local_actor
*
,
down_msg
&
dm
)
{
s
coped_s
elf
->
set_down_handler
([](
local_actor
*
,
down_msg
&
dm
)
{
CAF_CHECK_EQUAL
(
dm
.
reason
,
exit_reason
::
normal
);
});
}
template
<
class
...
Ts
>
void
spawn
(
Ts
&&
...
xs
)
{
testee
=
self
->
spawn
<
monitored
>
(
std
::
forward
<
Ts
>
(
xs
)...);
testee
=
s
coped_s
elf
->
spawn
<
monitored
>
(
std
::
forward
<
Ts
>
(
xs
)...);
}
~
fixture
()
{
self
->
wait_for
(
testee
);
s
coped_s
elf
->
wait_for
(
testee
);
}
};
...
...
libcaf_core/test/request_response.cpp
View file @
ba5cd462
...
...
@@ -259,16 +259,16 @@ CAF_TEST(test_void_res) {
}
CAF_TEST
(
pending_quit
)
{
auto
mirror
=
system
.
spawn
([](
event_based_actor
*
self
)
->
behavior
{
self
->
set_default_handler
(
reflect
);
auto
mirror
=
system
.
spawn
([](
event_based_actor
*
ptr
)
->
behavior
{
ptr
->
set_default_handler
(
reflect
);
return
{
[]
{
// nop
}
};
});
system
.
spawn
([
mirror
](
event_based_actor
*
self
)
{
self
->
request
(
mirror
,
infinite
,
42
).
then
(
system
.
spawn
([
mirror
](
event_based_actor
*
ptr
)
{
ptr
->
request
(
mirror
,
infinite
,
42
).
then
(
[](
int
)
{
CAF_ERROR
(
"received result, should've been terminated already"
);
},
...
...
@@ -276,7 +276,7 @@ CAF_TEST(pending_quit) {
CAF_CHECK_EQUAL
(
err
,
sec
::
request_receiver_down
);
}
);
self
->
quit
();
ptr
->
quit
();
});
}
...
...
@@ -334,7 +334,6 @@ CAF_TEST(request_to_mirror) {
}
CAF_TEST
(
request_to_a_fwd2_b_fwd2_c
)
{
scoped_actor
self
{
system
};
self
->
request
(
self
->
spawn
<
A
,
monitored
>
(
self
),
infinite
,
go_atom
::
value
,
self
->
spawn
<
B
>
(
self
->
spawn
<
C
>
())).
receive
(
[](
ok_atom
)
{
...
...
@@ -428,21 +427,21 @@ CAF_TEST(request_no_then) {
}
CAF_TEST
(
async_request
)
{
auto
foo
=
system
.
spawn
([](
event_based_actor
*
self
)
->
behavior
{
auto
receiver
=
self
->
spawn
<
linked
>
([](
event_based_actor
*
ptr
)
->
behavior
{
auto
foo
=
system
.
spawn
([](
event_based_actor
*
ptr
)
->
behavior
{
auto
receiver
=
ptr
->
spawn
<
linked
>
([](
event_based_actor
*
ptr2
)
->
behavior
{
return
{
[
=
](
int
)
{
return
ptr
->
make_response_promise
();
return
ptr
2
->
make_response_promise
();
}
};
});
self
->
request
(
receiver
,
infinite
,
1
).
then
(
ptr
->
request
(
receiver
,
infinite
,
1
).
then
(
[
=
](
int
)
{}
);
return
{
[
=
](
int
)
{
CAF_MESSAGE
(
"int received"
);
self
->
quit
(
exit_reason
::
user_shutdown
);
ptr
->
quit
(
exit_reason
::
user_shutdown
);
}
};
});
...
...
libcaf_core/test/typed_spawn.cpp
View file @
ba5cd462
...
...
@@ -413,7 +413,6 @@ CAF_TEST(string_delegator_chain) {
}
CAF_TEST
(
maybe_string_delegator_chain
)
{
scoped_actor
self
{
system
};
CAF_LOG_TRACE
(
CAF_ARG
(
self
));
auto
aut
=
system
.
spawn
(
maybe_string_delegator
,
system
.
spawn
(
maybe_string_reverter
));
...
...
@@ -437,7 +436,6 @@ CAF_TEST(maybe_string_delegator_chain) {
}
CAF_TEST
(
sending_typed_actors
)
{
scoped_actor
self
{
system
};
auto
aut
=
system
.
spawn
(
int_fun
);
self
->
send
(
self
->
spawn
(
foo
),
10
,
aut
);
self
->
receive
(
...
...
@@ -448,7 +446,6 @@ CAF_TEST(sending_typed_actors) {
}
CAF_TEST
(
sending_typed_actors_and_down_msg
)
{
scoped_actor
self
{
system
};
auto
aut
=
system
.
spawn
(
int_fun2
);
self
->
send
(
self
->
spawn
(
foo2
),
10
,
aut
);
self
->
receive
([](
int
i
)
{
...
...
@@ -460,20 +457,20 @@ CAF_TEST(check_signature) {
using
foo_type
=
typed_actor
<
replies_to
<
put_atom
>::
with
<
ok_atom
>>
;
using
foo_result_type
=
optional
<
ok_atom
>
;
using
bar_type
=
typed_actor
<
reacts_to
<
ok_atom
>>
;
auto
foo_action
=
[](
foo_type
::
pointer
self
)
->
foo_type
::
behavior_type
{
auto
foo_action
=
[](
foo_type
::
pointer
ptr
)
->
foo_type
::
behavior_type
{
return
{
[
=
]
(
put_atom
)
->
foo_result_type
{
self
->
quit
();
ptr
->
quit
();
return
{
ok_atom
::
value
};
}
};
};
auto
bar_action
=
[
=
](
bar_type
::
pointer
self
)
->
bar_type
::
behavior_type
{
auto
foo
=
self
->
spawn
<
linked
>
(
foo_action
);
self
->
send
(
foo
,
put_atom
::
value
);
auto
bar_action
=
[
=
](
bar_type
::
pointer
ptr
)
->
bar_type
::
behavior_type
{
auto
foo
=
ptr
->
spawn
<
linked
>
(
foo_action
);
ptr
->
send
(
foo
,
put_atom
::
value
);
return
{
[
=
](
ok_atom
)
{
self
->
quit
();
ptr
->
quit
();
}
};
};
...
...
libcaf_io/caf/io/middleman.hpp
View file @
ba5cd462
...
...
@@ -279,10 +279,10 @@ private:
auto
hdl
=
backend
().
new_tcp_scribe
(
host
,
port
);
detail
::
init_fun_factory
<
Impl
,
F
>
fac
;
actor_config
cfg
{
&
backend
()};
auto
init
=
fac
(
std
::
move
(
fun
),
hdl
,
std
::
forward
<
Ts
>
(
xs
)...);
cfg
.
init_fun
=
[
hdl
,
init
](
local_actor
*
ptr
)
->
behavior
{
auto
init
_fun
=
fac
(
std
::
move
(
fun
),
hdl
,
std
::
forward
<
Ts
>
(
xs
)...);
cfg
.
init_fun
=
[
hdl
,
init
_fun
](
local_actor
*
ptr
)
->
behavior
{
static_cast
<
abstract_broker
*>
(
ptr
)
->
assign_tcp_scribe
(
hdl
);
return
init
(
ptr
);
return
init
_fun
(
ptr
);
};
return
system
().
spawn_class
<
Impl
,
Os
>
(
cfg
);
}
...
...
@@ -291,12 +291,12 @@ private:
typename
infer_handle_from_class
<
Impl
>::
type
spawn_server_impl
(
F
fun
,
uint16_t
port
,
Ts
&&
...
xs
)
{
detail
::
init_fun_factory
<
Impl
,
F
>
fac
;
auto
init
=
fac
(
std
::
move
(
fun
),
std
::
forward
<
Ts
>
(
xs
)...);
auto
init
_fun
=
fac
(
std
::
move
(
fun
),
std
::
forward
<
Ts
>
(
xs
)...);
auto
hdl
=
backend
().
new_tcp_doorman
(
port
).
first
;
actor_config
cfg
{
&
backend
()};
cfg
.
init_fun
=
[
hdl
,
init
](
local_actor
*
ptr
)
->
behavior
{
cfg
.
init_fun
=
[
hdl
,
init
_fun
](
local_actor
*
ptr
)
->
behavior
{
static_cast
<
abstract_broker
*>
(
ptr
)
->
assign_tcp_doorman
(
hdl
);
return
init
(
ptr
);
return
init
_fun
(
ptr
);
};
return
system
().
spawn_class
<
Impl
,
Os
>
(
cfg
);
}
...
...
libcaf_io/src/abstract_broker.cpp
View file @
ba5cd462
...
...
@@ -194,10 +194,10 @@ resumable::subtype_t abstract_broker::subtype() const {
}
resumable
::
resume_result
abstract_broker
::
resume
(
execution_unit
*
c
ontext
,
size_t
mt
)
{
CAF_ASSERT
(
c
ontext
!=
nullptr
);
CAF_ASSERT
(
c
ontext
==
&
backend
());
return
local_actor
::
resume
(
c
ontext
,
mt
);
abstract_broker
::
resume
(
execution_unit
*
c
tx
,
size_t
mt
)
{
CAF_ASSERT
(
c
tx
!=
nullptr
);
CAF_ASSERT
(
c
tx
==
&
backend
());
return
local_actor
::
resume
(
c
tx
,
mt
);
}
const
char
*
abstract_broker
::
name
()
const
{
...
...
libcaf_io/src/basp_broker.cpp
View file @
ba5cd462
...
...
@@ -486,50 +486,50 @@ behavior basp_broker::make_behavior() {
}
},
// received from proxy instances
[
=
](
forward_atom
,
strong_actor_ptr
&
s
ender
,
[
=
](
forward_atom
,
strong_actor_ptr
&
s
rc
,
const
std
::
vector
<
strong_actor_ptr
>&
fwd_stack
,
strong_actor_ptr
&
receiver
,
message_id
mid
,
const
message
&
msg
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
s
ender
)
<<
CAF_ARG
(
receiver
)
strong_actor_ptr
&
dest
,
message_id
mid
,
const
message
&
msg
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
s
rc
)
<<
CAF_ARG
(
dest
)
<<
CAF_ARG
(
mid
)
<<
CAF_ARG
(
msg
));
if
(
!
receiver
||
system
().
node
()
==
receiver
->
node
())
{
if
(
!
dest
||
system
().
node
()
==
dest
->
node
())
{
CAF_LOG_WARNING
(
"cannot forward to invalid or local actor:"
<<
CAF_ARG
(
receiver
));
<<
CAF_ARG
(
dest
));
return
;
}
if
(
s
ender
&&
system
().
node
()
==
sender
->
node
())
system
().
registry
().
put
(
s
ender
->
id
(),
sender
);
if
(
!
state
.
instance
.
dispatch
(
context
(),
s
ender
,
fwd_stack
,
receiver
,
mid
,
msg
)
if
(
s
rc
&&
system
().
node
()
==
src
->
node
())
system
().
registry
().
put
(
s
rc
->
id
(),
src
);
if
(
!
state
.
instance
.
dispatch
(
context
(),
s
rc
,
fwd_stack
,
dest
,
mid
,
msg
)
&&
mid
.
is_request
())
{
detail
::
sync_request_bouncer
srb
{
exit_reason
::
remote_link_unreachable
};
srb
(
s
ender
,
mid
);
srb
(
s
rc
,
mid
);
}
},
// received from some system calls like whereis
[
=
](
forward_atom
,
strong_actor_ptr
&
s
ender
,
const
node_id
&
receiving_node
,
atom_value
receiver
_name
,
[
=
](
forward_atom
,
strong_actor_ptr
&
s
rc
,
const
node_id
&
dest_node
,
atom_value
dest
_name
,
const
message
&
msg
)
->
result
<
void
>
{
CAF_LOG_TRACE
(
CAF_ARG
(
s
ender
)
<<
", "
<<
CAF_ARG
(
receiving
_node
)
<<
", "
<<
CAF_ARG
(
receiver
_name
)
CAF_LOG_TRACE
(
CAF_ARG
(
s
rc
)
<<
", "
<<
CAF_ARG
(
dest
_node
)
<<
", "
<<
CAF_ARG
(
dest
_name
)
<<
", "
<<
CAF_ARG
(
msg
));
if
(
!
s
ender
)
if
(
!
s
rc
)
return
sec
::
cannot_forward_to_invalid_actor
;
if
(
system
().
node
()
==
s
ender
->
node
())
system
().
registry
().
put
(
s
ender
->
id
(),
sender
);
if
(
system
().
node
()
==
s
rc
->
node
())
system
().
registry
().
put
(
s
rc
->
id
(),
src
);
auto
writer
=
make_callback
([
&
](
serializer
&
sink
)
{
std
::
vector
<
actor_addr
>
stages
;
sink
<<
receiver
_name
<<
stages
<<
msg
;
sink
<<
dest
_name
<<
stages
<<
msg
;
});
auto
path
=
this
->
state
.
instance
.
tbl
().
lookup
(
receiving
_node
);
auto
path
=
this
->
state
.
instance
.
tbl
().
lookup
(
dest
_node
);
if
(
!
path
)
{
CAF_LOG_ERROR
(
"no route to receiving node"
);
return
sec
::
no_route_to_receiving_node
;
}
basp
::
header
hdr
{
basp
::
message_type
::
dispatch_message
,
basp
::
header
::
named_receiver_flag
,
0
,
0
,
state
.
this_node
(),
receiving
_node
,
s
ender
->
id
(),
invalid_actor_id
};
0
,
0
,
state
.
this_node
(),
dest
_node
,
s
rc
->
id
(),
invalid_actor_id
};
state
.
instance
.
write
(
context
(),
path
->
wr_buf
,
hdr
,
&
writer
);
state
.
instance
.
flush
(
*
path
);
return
unit
;
...
...
libcaf_io/src/default_multiplexer.cpp
View file @
ba5cd462
...
...
@@ -724,10 +724,10 @@ connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
CAF_LOG_TRACE
(
""
);
class
impl
:
public
scribe
{
public:
impl
(
abstract_broker
*
ptr
,
default_multiplexer
&
ref
,
native_socket
sockfd
)
impl
(
abstract_broker
*
ptr
,
default_multiplexer
&
mx
,
native_socket
sockfd
)
:
scribe
(
ptr
,
network
::
conn_hdl_from_socket
(
sockfd
)),
launched_
(
false
),
stream_
(
ref
,
sockfd
)
{
stream_
(
mx
,
sockfd
)
{
// nop
}
void
configure_read
(
receive_policy
::
config
config
)
override
{
...
...
@@ -782,9 +782,9 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
CAF_ASSERT
(
fd
!=
network
::
invalid_native_socket
);
class
impl
:
public
doorman
{
public:
impl
(
abstract_broker
*
ptr
,
default_multiplexer
&
ref
,
native_socket
sockfd
)
impl
(
abstract_broker
*
ptr
,
default_multiplexer
&
mx
,
native_socket
sockfd
)
:
doorman
(
ptr
,
network
::
accept_hdl_from_socket
(
sockfd
)),
acceptor_
(
ref
,
sockfd
)
{
acceptor_
(
mx
,
sockfd
)
{
// nop
}
void
new_connection
()
override
{
...
...
@@ -998,8 +998,8 @@ void pipe_reader::handle_event(operation op) {
}
}
void
pipe_reader
::
init
(
native_socket
fd
)
{
fd_
=
fd
;
void
pipe_reader
::
init
(
native_socket
sock_
fd
)
{
fd_
=
sock_
fd
;
}
stream
::
stream
(
default_multiplexer
&
backend_ref
,
native_socket
sockfd
)
...
...
libcaf_io/src/test_multiplexer.cpp
View file @
ba5cd462
...
...
@@ -39,9 +39,9 @@ test_multiplexer::~test_multiplexer() {
}
connection_handle
test_multiplexer
::
new_tcp_scribe
(
const
std
::
string
&
host
,
uint16_t
port
)
{
test_multiplexer
::
new_tcp_scribe
(
const
std
::
string
&
host
,
uint16_t
port
_hint
)
{
connection_handle
result
;
auto
i
=
scribes_
.
find
(
std
::
make_pair
(
host
,
port
));
auto
i
=
scribes_
.
find
(
std
::
make_pair
(
host
,
port
_hint
));
if
(
i
!=
scribes_
.
end
())
{
result
=
i
->
second
;
scribes_
.
erase
(
i
);
...
...
libcaf_test/caf/test/unit_test_impl.hpp
View file @
ba5cd462
...
...
@@ -196,14 +196,12 @@ logger::stream& logger::stream::operator<<(const char* cstr) {
return
*
this
;
}
logger
::
stream
&
logger
::
stream
::
operator
<<
(
const
std
::
string
&
str
)
{
if
(
str
.
empty
())
{
logger
::
stream
&
logger
::
stream
::
operator
<<
(
const
std
::
string
&
x
)
{
if
(
x
.
empty
())
return
*
this
;
}
buf_
<<
str
;
if
(
str
.
back
()
==
'\n'
)
{
buf_
<<
x
;
if
(
x
.
back
()
==
'\n'
)
flush
();
}
return
*
this
;
}
...
...
@@ -212,10 +210,10 @@ std::string logger::stream::str() const {
}
void
logger
::
stream
::
flush
()
{
auto
str
=
buf_
.
str
();
auto
x
=
buf_
.
str
();
buf_
.
str
(
""
);
logger_
.
log
(
level_
,
str
);
str_
+=
str
;
logger_
.
log
(
level_
,
x
);
str_
+=
x
;
}
bool
logger
::
init
(
int
lvl_cons
,
int
lvl_file
,
const
std
::
string
&
logfile
)
{
...
...
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