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
06f7f1fb
Commit
06f7f1fb
authored
Jan 18, 2017
by
Dominik Charousset
Committed by
Marian Triebe
Apr 23, 2017
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Implement self->new_stream, relates #531
parent
d4492a03
Changes
6
Show whitespace changes
Inline
Side-by-side
Showing
6 changed files
with
226 additions
and
12 deletions
+226
-12
libcaf_core/caf/named_actor_config.hpp
libcaf_core/caf/named_actor_config.hpp
+6
-10
libcaf_core/caf/scheduled_actor.hpp
libcaf_core/caf/scheduled_actor.hpp
+52
-0
libcaf_core/caf/stream_result_trait.hpp
libcaf_core/caf/stream_result_trait.hpp
+41
-0
libcaf_core/caf/stream_sink_impl.hpp
libcaf_core/caf/stream_sink_impl.hpp
+1
-1
libcaf_core/src/stream_msg_visitor.cpp
libcaf_core/src/stream_msg_visitor.cpp
+1
-1
libcaf_core/test/streaming.cpp
libcaf_core/test/streaming.cpp
+125
-0
No files found.
libcaf_core/caf/named_actor_config.hpp
View file @
06f7f1fb
...
@@ -25,6 +25,8 @@
...
@@ -25,6 +25,8 @@
#include "caf/atom.hpp"
#include "caf/atom.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/meta/type_name.hpp"
namespace
caf
{
namespace
caf
{
/// Stores a flow-control configuration.
/// Stores a flow-control configuration.
...
@@ -34,16 +36,10 @@ struct named_actor_config {
...
@@ -34,16 +36,10 @@ struct named_actor_config {
size_t
max_pending
;
size_t
max_pending
;
};
};
template
<
class
Processor
>
template
<
class
Inspector
>
void
serialize
(
Processor
&
proc
,
named_actor_config
&
x
,
unsigned
int
)
{
typename
Inspector
::
result_type
inspect
(
Inspector
&
f
,
named_actor_config
&
x
)
{
proc
&
x
.
strategy
;
return
f
(
meta
::
type_name
(
"named_actor_config"
),
x
.
strategy
,
x
.
low_watermark
,
proc
&
x
.
low_watermark
;
x
.
max_pending
);
proc
&
x
.
max_pending
;
}
inline
std
::
string
to_string
(
const
named_actor_config
&
x
)
{
return
"named_actor_config"
+
deep_to_string_as_tuple
(
x
.
strategy
,
x
.
low_watermark
,
x
.
max_pending
);
}
}
}
// namespace caf
}
// namespace caf
...
...
libcaf_core/caf/scheduled_actor.hpp
View file @
06f7f1fb
...
@@ -38,6 +38,7 @@
...
@@ -38,6 +38,7 @@
#include "caf/stream_sink_impl.hpp"
#include "caf/stream_sink_impl.hpp"
#include "caf/stream_stage_impl.hpp"
#include "caf/stream_stage_impl.hpp"
#include "caf/stream_source_impl.hpp"
#include "caf/stream_source_impl.hpp"
#include "caf/stream_result_trait.hpp"
#include "caf/policy/greedy.hpp"
#include "caf/policy/greedy.hpp"
#include "caf/policy/anycast.hpp"
#include "caf/policy/anycast.hpp"
...
@@ -294,6 +295,57 @@ public:
...
@@ -294,6 +295,57 @@ public:
// -- stream management ------------------------------------------------------
// -- stream management ------------------------------------------------------
// Starts a new stream.
template
<
class
Handle
,
class
Init
,
class
Getter
,
class
ClosedPredicate
,
class
ResHandler
>
stream
<
typename
stream_source_trait_t
<
Getter
>::
output
>
new_stream
(
const
Handle
&
dest
,
Init
init
,
Getter
getter
,
ClosedPredicate
pred
,
ResHandler
res_handler
)
{
using
type
=
typename
stream_source_trait_t
<
Getter
>::
output
;
using
state_type
=
typename
stream_source_trait_t
<
Getter
>::
state
;
using
result_type
=
typename
stream_result_trait_t
<
ResHandler
>::
type
;
static_assert
(
std
::
is_same
<
void
(
state_type
&
),
typename
detail
::
get_callable_trait
<
Init
>::
fun_sig
>::
value
,
"Expected signature `void (State&)` for init function"
);
static_assert
(
std
::
is_same
<
bool
(
const
state_type
&
),
typename
detail
::
get_callable_trait
<
ClosedPredicate
>::
fun_sig
>::
value
,
"Expected signature `bool (const State&)` for "
"closed_predicate function"
);
if
(
!
dest
)
{
CAF_LOG_ERROR
(
"cannot stream to an invalid actor handle"
);
return
stream_id
{
nullptr
,
0
};
}
// generate new stream ID
stream_id
sid
{
ctrl
(),
new_request_id
(
message_priority
::
normal
).
integer_value
()};
stream
<
type
>
token
{
sid
};
// generate new ID for the final response message and send handshake
auto
res_id
=
new_request_id
(
message_priority
::
normal
);
dest
->
enqueue
(
make_mailbox_element
(
ctrl
(),
res_id
,
{},
make
<
stream_msg
::
open
>
(
sid
,
make_message
(
std
::
move
(
token
)),
ctrl
(),
stream_priority
::
normal
,
std
::
vector
<
atom_value
>
{},
false
)),
context
());
// install response handler
this
->
add_multiplexed_response_handler
(
res_id
.
response_id
(),
behavior
{[
=
](
result_type
&
res
)
{
res_handler
(
std
::
move
(
res
));
},
[
=
](
error
&
err
)
{
res_handler
(
std
::
move
(
err
));
}});
// install stream handler
using
impl
=
stream_source_impl
<
Getter
,
ClosedPredicate
>
;
std
::
unique_ptr
<
downstream_policy
>
p
{
new
policy
::
anycast
};
auto
ptr
=
make_counted
<
impl
>
(
this
,
sid
,
std
::
move
(
p
),
std
::
move
(
getter
),
std
::
move
(
pred
));
init
(
ptr
->
state
());
streams_
.
emplace
(
std
::
move
(
sid
),
std
::
move
(
ptr
));
return
sid
;
}
/// Adds a stream source to this actor.
/// Adds a stream source to this actor.
template
<
class
Init
,
class
Getter
,
class
ClosedPredicate
>
template
<
class
Init
,
class
Getter
,
class
ClosedPredicate
>
stream
<
typename
stream_source_trait_t
<
Getter
>::
output
>
stream
<
typename
stream_source_trait_t
<
Getter
>::
output
>
...
...
libcaf_core/caf/stream_result_trait.hpp
0 → 100644
View file @
06f7f1fb
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://openresult.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_RESULT_TRAIT_HPP
#define CAF_STREAM_RESULT_TRAIT_HPP
#include "caf/detail/type_traits.hpp"
namespace
caf
{
template
<
class
F
>
struct
stream_result_trait
;
template
<
class
T
>
struct
stream_result_trait
<
void
(
expected
<
T
>
)
>
{
using
type
=
T
;
};
template
<
class
F
>
using
stream_result_trait_t
=
stream_result_trait
<
typename
detail
::
get_callable_trait
<
F
>::
fun_sig
>
;
}
// namespace caf
#endif // CAF_STREAM_RESULT_TRAIT_HPP
libcaf_core/caf/stream_sink_impl.hpp
View file @
06f7f1fb
...
@@ -49,7 +49,7 @@ public:
...
@@ -49,7 +49,7 @@ public:
}
}
error
consume
(
message
&
msg
)
final
{
error
consume
(
message
&
msg
)
final
{
using
vec_type
=
std
::
vector
<
out
put_type
>
;
using
vec_type
=
std
::
vector
<
in
put_type
>
;
if
(
msg
.
match_elements
<
vec_type
>
())
{
if
(
msg
.
match_elements
<
vec_type
>
())
{
auto
&
xs
=
msg
.
get_as
<
vec_type
>
(
0
);
auto
&
xs
=
msg
.
get_as
<
vec_type
>
(
0
);
for
(
auto
&
x
:
xs
)
for
(
auto
&
x
:
xs
)
...
...
libcaf_core/src/stream_msg_visitor.cpp
View file @
06f7f1fb
...
@@ -41,9 +41,9 @@ auto stream_msg_visitor::operator()(stream_msg::open& x) -> result_type {
...
@@ -41,9 +41,9 @@ auto stream_msg_visitor::operator()(stream_msg::open& x) -> result_type {
return
{
sec
::
downstream_already_exists
,
e_
};
return
{
sec
::
downstream_already_exists
,
e_
};
auto
&
predecessor
=
x
.
prev_stage
;
auto
&
predecessor
=
x
.
prev_stage
;
auto
fail
=
[
&
](
error
reason
)
->
result_type
{
auto
fail
=
[
&
](
error
reason
)
->
result_type
{
unsafe_send_as
(
self_
,
predecessor
,
make
<
stream_msg
::
abort
>
(
sid_
,
reason
));
auto
rp
=
self_
->
make_response_promise
();
auto
rp
=
self_
->
make_response_promise
();
rp
.
deliver
(
reason
);
rp
.
deliver
(
reason
);
unsafe_send_as
(
self_
,
predecessor
,
make
<
stream_msg
::
abort
>
(
sid_
,
reason
));
return
{
std
::
move
(
reason
),
e_
};
return
{
std
::
move
(
reason
),
e_
};
};
};
if
(
!
predecessor
)
{
if
(
!
predecessor
)
{
...
...
libcaf_core/test/streaming.cpp
View file @
06f7f1fb
...
@@ -59,6 +59,33 @@ behavior file_reader(event_based_actor* self) {
...
@@ -59,6 +59,33 @@ behavior file_reader(event_based_actor* self) {
};
};
}
}
void
streamer
(
event_based_actor
*
self
,
const
actor
&
dest
)
{
using
buf
=
std
::
deque
<
int
>
;
self
->
new_stream
(
// destination of the stream
dest
,
// initialize state
[
&
](
buf
&
xs
)
{
xs
=
buf
{
1
,
2
,
3
,
4
,
5
,
6
,
7
,
8
,
9
};
},
// get next element
[
=
](
buf
&
xs
,
downstream
<
int
>&
out
,
size_t
num
)
{
auto
n
=
std
::
min
(
num
,
xs
.
size
());
for
(
size_t
i
=
0
;
i
<
n
;
++
i
)
out
.
push
(
xs
[
i
]);
xs
.
erase
(
xs
.
begin
(),
xs
.
begin
()
+
static_cast
<
ptrdiff_t
>
(
n
));
},
// check whether we reached the end
[
=
](
const
buf
&
xs
)
{
return
xs
.
empty
();
},
// handle result of the stream
[
=
](
expected
<
int
>
)
{
// nop
}
);
}
behavior
filter
(
event_based_actor
*
self
)
{
behavior
filter
(
event_based_actor
*
self
)
{
return
{
return
{
[
=
](
stream
<
int
>&
in
)
->
stream
<
int
>
{
[
=
](
stream
<
int
>&
in
)
->
stream
<
int
>
{
...
@@ -114,6 +141,56 @@ behavior sum_up(event_based_actor* self) {
...
@@ -114,6 +141,56 @@ behavior sum_up(event_based_actor* self) {
};
};
}
}
behavior
drop_all
(
event_based_actor
*
self
)
{
return
{
[
=
](
stream
<
int
>&
in
)
{
return
self
->
add_sink
(
// input stream
in
,
// initialize state
[](
unit_t
&
)
{
// nop
},
// processing step
[](
unit_t
&
,
int
)
{
// nop
},
// cleanup and produce void "result"
[](
unit_t
&
)
->
unit_t
{
return
unit
;
}
);
}
};
}
void
streamer_without_result
(
event_based_actor
*
self
,
const
actor
&
dest
)
{
using
buf
=
std
::
deque
<
int
>
;
self
->
new_stream
(
// destination of the stream
dest
,
// initialize state
[
&
](
buf
&
xs
)
{
xs
=
buf
{
1
,
2
,
3
,
4
,
5
,
6
,
7
,
8
,
9
};
},
// get next element
[
=
](
buf
&
xs
,
downstream
<
int
>&
out
,
size_t
num
)
{
auto
n
=
std
::
min
(
num
,
xs
.
size
());
for
(
size_t
i
=
0
;
i
<
n
;
++
i
)
out
.
push
(
xs
[
i
]);
xs
.
erase
(
xs
.
begin
(),
xs
.
begin
()
+
static_cast
<
ptrdiff_t
>
(
n
));
},
// check whether we reached the end
[
=
](
const
buf
&
xs
)
{
return
xs
.
empty
();
},
// handle result of the stream
[
=
](
expected
<
void
>
)
{
// nop
}
);
}
using
fixture
=
test_coordinator_fixture
<>
;
using
fixture
=
test_coordinator_fixture
<>
;
}
// namespace <anonymous>
}
// namespace <anonymous>
...
@@ -263,4 +340,52 @@ CAF_TEST(depth3_pipeline) {
...
@@ -263,4 +340,52 @@ CAF_TEST(depth3_pipeline) {
//CAF_CHECK_EQUAL(res, 25);
//CAF_CHECK_EQUAL(res, 25);
}
}
CAF_TEST
(
broken_pipeline_stramer
)
{
CAF_MESSAGE
(
"streams must abort if a stage fails to initialize its state"
);
auto
stage
=
sys
.
spawn
(
broken_filter
);
// run initialization code
sched
.
run
();
auto
source
=
sys
.
spawn
(
streamer
,
stage
);
// run initialization code
sched
.
run_once
();
// source --(stream_msg::open)--> stage
expect
((
stream_msg
::
open
),
from
(
source
).
to
(
stage
).
with
(
_
,
source
,
_
,
_
,
false
));
CAF_CHECK
(
!
deref
(
source
).
streams
().
empty
());
CAF_CHECK
(
deref
(
stage
).
streams
().
empty
());
// stage --(stream_msg::abort)--> source
expect
((
stream_msg
::
abort
),
from
(
stage
).
to
(
source
).
with
(
sec
::
stream_init_failed
));
CAF_CHECK
(
deref
(
source
).
streams
().
empty
());
CAF_CHECK
(
deref
(
stage
).
streams
().
empty
());
// sink ----(error)---> source
expect
((
error
),
from
(
stage
).
to
(
source
).
with
(
_
));
}
CAF_TEST
(
depth2_pipeline_streamer
)
{
auto
sink
=
sys
.
spawn
(
sum_up
);
// run initialization code
sched
.
run
();
auto
source
=
sys
.
spawn
(
streamer
,
sink
);
// run initialization code
sched
.
run_once
();
// source ----(stream_msgreturn ::open)----> sink
expect
((
stream_msg
::
open
),
from
(
source
).
to
(
sink
).
with
(
_
,
source
,
_
,
_
,
false
));
// source <----(stream_msg::ack_open)------ sink
expect
((
stream_msg
::
ack_open
),
from
(
sink
).
to
(
source
).
with
(
5
,
_
,
false
));
// source ----(stream_msg::batch)---> sink
expect
((
stream_msg
::
batch
),
from
(
source
).
to
(
sink
).
with
(
5
,
std
::
vector
<
int
>
{
1
,
2
,
3
,
4
,
5
},
0
));
// source <--(stream_msg::ack_batch)---- sink
expect
((
stream_msg
::
ack_batch
),
from
(
sink
).
to
(
source
).
with
(
5
,
0
));
// source ----(stream_msg::batch)---> sink
expect
((
stream_msg
::
batch
),
from
(
source
).
to
(
sink
).
with
(
4
,
std
::
vector
<
int
>
{
6
,
7
,
8
,
9
},
1
));
// source <--(stream_msg::ack_batch)---- sink
expect
((
stream_msg
::
ack_batch
),
from
(
sink
).
to
(
source
).
with
(
4
,
1
));
// source ----(stream_msg::close)---> sink
expect
((
stream_msg
::
close
),
from
(
source
).
to
(
sink
).
with
());
// sink ----(result: 25)---> source
expect
((
int
),
from
(
sink
).
to
(
source
).
with
(
45
));
}
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