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
28a53ad4
Commit
28a53ad4
authored
Sep 05, 2018
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Add streaming example
parent
903f801c
Changes
2
Show whitespace changes
Inline
Side-by-side
Showing
2 changed files
with
146 additions
and
0 deletions
+146
-0
examples/CMakeLists.txt
examples/CMakeLists.txt
+3
-0
examples/streaming/integer_stream.cpp
examples/streaming/integer_stream.cpp
+143
-0
No files found.
examples/CMakeLists.txt
View file @
28a53ad4
...
@@ -30,6 +30,9 @@ add(message_passing promises)
...
@@ -30,6 +30,9 @@ add(message_passing promises)
add
(
message_passing request
)
add
(
message_passing request
)
add
(
message_passing typed_calculator
)
add
(
message_passing typed_calculator
)
# streaming API
add
(
streaming integer_stream
)
# dynamic behavior changes using 'become'
# dynamic behavior changes using 'become'
add
(
dynamic_behavior skip_messages
)
add
(
dynamic_behavior skip_messages
)
add
(
dynamic_behavior dining_philosophers
)
add
(
dynamic_behavior dining_philosophers
)
...
...
examples/streaming/integer_stream.cpp
0 → 100644
View file @
28a53ad4
/******************************************************************************
* Basic, non-interactive streaming example for processing integers. *
******************************************************************************/
// Manual refs: lines 17-52, 54-87, 89-120, 133-139 (Streaming)
#include <iostream>
#include <vector>
#include "caf/all.hpp"
using
std
::
endl
;
using
namespace
caf
;
namespace
{
// Simple source for generating a stream of integers from [0, n).
behavior
int_source
(
event_based_actor
*
self
)
{
return
{
[
=
](
open_atom
,
int
n
)
{
// Produce at least one value.
if
(
n
<=
0
)
n
=
1
;
// Create a stream manager for implementing a stream source. The
// streaming logic requires three functions: initializer, generator, and
// predicate.
return
self
->
make_source
(
// Initializer. The type of the first argument (state) is freely
// chosen. If no state is required, `caf::unit_t` can be used here.
[](
int
&
x
)
{
x
=
0
;
},
// Generator. This function is called by CAF to produce new stream
// elements for downstream actors. The `x` argument is our state again
// (with our freely chosen type). The second argument `out` points to
// the output buffer. The template argument (here: int) determines what
// elements downstream actors receive in this stream. Finally, `num` is
// a hint from CAF how many elements we should ideally insert into
// `out`. We can always insert fewer or more items.
[
n
](
int
&
x
,
downstream
<
int
>&
out
,
size_t
num
)
{
auto
max_x
=
std
::
min
(
x
+
static_cast
<
int
>
(
num
),
n
);
for
(;
x
<
max_x
;
++
x
)
out
.
push
(
x
);
},
// Predicate. This function tells CAF when we reached the end.
[
n
](
const
int
&
x
)
{
return
x
==
n
;
}
);
}
};
}
// Simple stage that only selects even numbers.
behavior
int_selector
(
event_based_actor
*
self
)
{
return
{
[
=
](
stream
<
int
>
in
)
{
// Create a stream manager for implementing a stream stage. Similar to
// `make_source`, we need three functions: initialzer, processor, and
// finalizer.
return
self
->
make_stage
(
// Our input source.
in
,
// Initializer. Here, we don't need any state and simply use unit_t.
[](
unit_t
&
)
{
// nop
},
// Processor. This function takes individual input elements as `val`
// and forwards even integers to `out`.
[](
unit_t
&
,
downstream
<
int
>&
out
,
int
val
)
{
if
(
val
%
2
==
0
)
out
.
push
(
val
);
},
// Finalizer. Allows us to run cleanup code once the stream terminates.
[
=
](
unit_t
&
,
const
error
&
err
)
{
if
(
err
)
{
aout
(
self
)
<<
"int_selector aborted with error: "
<<
err
<<
std
::
endl
;
}
else
{
aout
(
self
)
<<
"int_selector finalized"
<<
std
::
endl
;
}
// else: regular stream shutdown
}
);
}
};
}
behavior
int_sink
(
event_based_actor
*
self
)
{
return
{
[
=
](
stream
<
int
>
in
)
{
// Create a stream manager for implementing a stream sink. Once more, we
// have to provide three functions: Initializer, Consumer, Finalizer.
return
self
->
make_sink
(
// Our input source.
in
,
// Initializer. Here, we store all values we receive. Note that streams
// are potentially unbound, so this is usually a bad idea outside small
// examples like this one.
[](
std
::
vector
<
int
>&
)
{
// nop
},
// Consumer. Takes individual input elements as `val` and stores them
// in our history.
[](
std
::
vector
<
int
>&
xs
,
int
val
)
{
xs
.
emplace_back
(
val
);
},
// Finalizer. Allows us to run cleanup code once the stream terminates.
[
=
](
std
::
vector
<
int
>&
xs
,
const
error
&
err
)
{
if
(
err
)
{
aout
(
self
)
<<
"int_sink aborted with error: "
<<
err
<<
std
::
endl
;
}
else
{
aout
(
self
)
<<
"int_sink finalized after receiving: "
<<
xs
<<
std
::
endl
;
}
}
);
}
};
}
struct
config
:
actor_system_config
{
config
()
{
opt_group
{
custom_options_
,
"global"
}
.
add
(
with_stage
,
"with-stage,s"
,
"use a stage for filtering odd numbers"
)
.
add
(
n
,
"num-values,n"
,
"number of values produced by the source"
);
}
bool
with_stage
=
false
;
int
n
=
100
;
};
void
caf_main
(
actor_system
&
sys
,
const
config
&
cfg
)
{
auto
src
=
sys
.
spawn
(
int_source
);
auto
snk
=
sys
.
spawn
(
int_sink
);
auto
pipeline
=
cfg
.
with_stage
?
snk
*
sys
.
spawn
(
int_selector
)
*
src
:
snk
*
src
;
anon_send
(
pipeline
,
open_atom
::
value
,
cfg
.
n
);
}
}
// namespace <anonymous>
CAF_MAIN
()
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