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
6a47f60d
Commit
6a47f60d
authored
Sep 03, 2019
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Add attach_stream_sink to replace make_sink
parent
f78f6c9c
Changes
8
Hide whitespace changes
Inline
Side-by-side
Showing
8 changed files
with
138 additions
and
74 deletions
+138
-74
examples/streaming/integer_stream.cpp
examples/streaming/integer_stream.cpp
+3
-2
libcaf_core/caf/all.hpp
libcaf_core/caf/all.hpp
+1
-0
libcaf_core/caf/attach_stream_sink.hpp
libcaf_core/caf/attach_stream_sink.hpp
+64
-0
libcaf_core/caf/scheduled_actor.hpp
libcaf_core/caf/scheduled_actor.hpp
+2
-0
libcaf_core/test/continuous_streaming.cpp
libcaf_core/test/continuous_streaming.cpp
+55
-67
libcaf_core/test/fused_downstream_manager.cpp
libcaf_core/test/fused_downstream_manager.cpp
+5
-2
libcaf_core/test/pipeline_streaming.cpp
libcaf_core/test/pipeline_streaming.cpp
+5
-2
libcaf_core/test/selective_streaming.cpp
libcaf_core/test/selective_streaming.cpp
+3
-1
No files found.
examples/streaming/integer_stream.cpp
View file @
6a47f60d
...
...
@@ -2,7 +2,7 @@
* Basic, non-interactive streaming example for processing integers. *
******************************************************************************/
// Manual refs: lines 17-46, 48-78, 80-10
6, 120-124
(Streaming)
// Manual refs: lines 17-46, 48-78, 80-10
7, 121-125
(Streaming)
#include <iostream>
#include <vector>
...
...
@@ -81,7 +81,8 @@ 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
(
return
attach_stream_sink
(
self
,
// Our input source.
in
,
// Initializer. Here, we store all values we receive. Note that streams
...
...
libcaf_core/caf/all.hpp
View file @
6a47f60d
...
...
@@ -35,6 +35,7 @@
#include "caf/after.hpp"
#include "caf/atom.hpp"
#include "caf/attach_continuous_stream_source.hpp"
#include "caf/attach_stream_sink.hpp"
#include "caf/attach_stream_source.hpp"
#include "caf/attach_stream_stage.hpp"
#include "caf/attachable.hpp"
...
...
libcaf_core/caf/attach_stream_sink.hpp
0 → 100644
View file @
6a47f60d
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/default_downstream_manager.hpp"
#include "caf/detail/stream_sink_driver_impl.hpp"
#include "caf/detail/stream_sink_impl.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/fwd.hpp"
#include "caf/make_sink_result.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream.hpp"
#include "caf/stream_source.hpp"
namespace
caf
{
/// Attaches a new stream sink to `self` by creating a default stream sink /
/// manager from given callbacks.
/// @param self Points to the hosting actor.
/// @param xs Additional constructor arguments for `Driver`.
/// @returns The new `stream_manager`, an inbound slot, and an outbound slot.
template
<
class
Driver
,
class
...
Ts
>
make_sink_result
<
typename
Driver
::
input_type
>
attach_stream_sink
(
scheduled_actor
*
self
,
stream
<
typename
Driver
::
input_type
>
in
,
Ts
&&
...
xs
)
{
auto
mgr
=
detail
::
make_stream_sink
<
Driver
>
(
self
,
std
::
forward
<
Ts
>
(
xs
)...);
auto
slot
=
mgr
->
add_inbound_path
(
in
);
return
{
slot
,
std
::
move
(
mgr
)};
}
/// Attaches a new stream sink to `self` by creating a default stream sink
/// manager from given callbacks.
/// @param self Points to the hosting actor.
/// @param in Stream handshake from upstream path.
/// @param init Function object for initializing the state of the sink.
/// @param fun Processing function.
/// @param fin Optional cleanup handler.
/// @returns The new `stream_manager` and the inbound slot.
template
<
class
In
,
class
Init
,
class
Fun
,
class
Finalize
=
unit_t
,
class
Trait
=
stream_sink_trait_t
<
Fun
>
>
make_sink_result
<
In
>
attach_stream_sink
(
scheduled_actor
*
self
,
stream
<
In
>
in
,
Init
init
,
Fun
fun
,
Finalize
fin
=
{})
{
using
driver
=
detail
::
stream_sink_driver_impl
<
In
,
Fun
,
Finalize
>
;
return
attach_stream_sink
<
driver
>
(
self
,
in
,
std
::
move
(
init
),
std
::
move
(
fun
),
std
::
move
(
fin
));
}
}
// namespace caf
libcaf_core/caf/scheduled_actor.hpp
View file @
6a47f60d
...
...
@@ -521,6 +521,7 @@ public:
std
::
move
(
done
),
std
::
move
(
fin
));
}
/// @deprecated Please use `attach_stream_sink` instead.
template
<
class
Driver
,
class
...
Ts
>
make_sink_result
<
typename
Driver
::
input_type
>
make_sink
(
const
stream
<
typename
Driver
::
input_type
>&
src
,
Ts
&&
...
xs
)
{
...
...
@@ -529,6 +530,7 @@ public:
return
{
slot
,
std
::
move
(
mgr
)};
}
/// @deprecated Please use `attach_stream_sink` instead.
template
<
class
In
,
class
Init
,
class
Fun
,
class
Finalize
=
unit_t
,
class
Trait
=
stream_sink_trait_t
<
Fun
>
>
make_sink_result
<
In
>
make_sink
(
const
stream
<
In
>&
in
,
Init
init
,
Fun
fun
,
...
...
libcaf_core/test/continuous_streaming.cpp
View file @
6a47f60d
...
...
@@ -25,6 +25,7 @@
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/attach_stream_sink.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/stateful_actor.hpp"
...
...
@@ -46,39 +47,37 @@ TESTEE_STATE(file_reader) {
};
VARARGS_TESTEE
(
file_reader
,
size_t
buf_size
)
{
return
{
[
=
](
string
&
fname
)
->
output_stream
<
int
,
string
>
{
CAF_CHECK_EQUAL
(
fname
,
"numbers.txt"
);
CAF_CHECK_EQUAL
(
self
->
mailbox
().
empty
(),
true
);
return
attach_stream_source
(
self
,
// forward file name in handshake to next stage
std
::
forward_as_tuple
(
std
::
move
(
fname
)),
// initialize state
[
=
](
unit_t
&
)
{
auto
&
xs
=
self
->
state
.
buf
;
xs
.
resize
(
buf_size
);
std
::
iota
(
xs
.
begin
(),
xs
.
end
(),
1
);
},
// get next element
[
=
](
unit_t
&
,
downstream
<
int
>&
out
,
size_t
num
)
{
auto
&
xs
=
self
->
state
.
buf
;
CAF_MESSAGE
(
"push "
<<
num
<<
" messages downstream"
);
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
unit_t
&
)
{
if
(
self
->
state
.
buf
.
empty
())
{
CAF_MESSAGE
(
self
->
name
()
<<
" is done"
);
return
true
;
}
return
false
;
});
}
};
return
{[
=
](
string
&
fname
)
->
output_stream
<
int
,
string
>
{
CAF_CHECK_EQUAL
(
fname
,
"numbers.txt"
);
CAF_CHECK_EQUAL
(
self
->
mailbox
().
empty
(),
true
);
return
attach_stream_source
(
self
,
// forward file name in handshake to next stage
std
::
forward_as_tuple
(
std
::
move
(
fname
)),
// initialize state
[
=
](
unit_t
&
)
{
auto
&
xs
=
self
->
state
.
buf
;
xs
.
resize
(
buf_size
);
std
::
iota
(
xs
.
begin
(),
xs
.
end
(),
1
);
},
// get next element
[
=
](
unit_t
&
,
downstream
<
int
>&
out
,
size_t
num
)
{
auto
&
xs
=
self
->
state
.
buf
;
CAF_MESSAGE
(
"push "
<<
num
<<
" messages downstream"
);
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
unit_t
&
)
{
if
(
self
->
state
.
buf
.
empty
())
{
CAF_MESSAGE
(
self
->
name
()
<<
" is done"
);
return
true
;
}
return
false
;
});
}};
}
TESTEE_STATE
(
sum_up
)
{
...
...
@@ -86,32 +85,26 @@ TESTEE_STATE(sum_up) {
};
TESTEE
(
sum_up
)
{
return
{
[
=
](
stream
<
int
>&
in
,
const
string
&
fname
)
{
CAF_CHECK_EQUAL
(
fname
,
"numbers.txt"
);
using
int_ptr
=
int
*
;
return
self
->
make_sink
(
// input stream
in
,
// initialize state
[
=
](
int_ptr
&
x
)
{
x
=
&
self
->
state
.
x
;
},
// processing step
[](
int_ptr
&
x
,
int
y
)
{
*
x
+=
y
;
},
// cleanup
[
=
](
int_ptr
&
,
const
error
&
)
{
CAF_MESSAGE
(
self
->
name
()
<<
" is done"
);
}
);
},
[
=
](
join_atom
atm
,
actor
src
)
{
CAF_MESSAGE
(
self
->
name
()
<<
" joins a stream"
);
self
->
send
(
self
*
src
,
atm
);
}
};
return
{[
=
](
stream
<
int
>&
in
,
const
string
&
fname
)
{
CAF_CHECK_EQUAL
(
fname
,
"numbers.txt"
);
using
int_ptr
=
int
*
;
return
attach_stream_sink
(
self
,
// input stream
in
,
// initialize state
[
=
](
int_ptr
&
x
)
{
x
=
&
self
->
state
.
x
;
},
// processing step
[](
int_ptr
&
x
,
int
y
)
{
*
x
+=
y
;
},
// cleanup
[
=
](
int_ptr
&
,
const
error
&
)
{
CAF_MESSAGE
(
self
->
name
()
<<
" is done"
);
});
},
[
=
](
join_atom
atm
,
actor
src
)
{
CAF_MESSAGE
(
self
->
name
()
<<
" joins a stream"
);
self
->
send
(
self
*
src
,
atm
);
}};
}
TESTEE_STATE
(
stream_multiplexer
)
{
...
...
@@ -125,14 +118,9 @@ TESTEE(stream_multiplexer) {
// nop
},
// processing step
[](
unit_t
&
,
downstream
<
int
>&
out
,
int
x
)
{
out
.
push
(
x
);
},
[](
unit_t
&
,
downstream
<
int
>&
out
,
int
x
)
{
out
.
push
(
x
);
},
// cleanup
[
=
](
unit_t
&
,
const
error
&
)
{
CAF_MESSAGE
(
self
->
name
()
<<
" is done"
);
}
);
[
=
](
unit_t
&
,
const
error
&
)
{
CAF_MESSAGE
(
self
->
name
()
<<
" is done"
);
});
return
{
[
=
](
join_atom
)
{
CAF_MESSAGE
(
"received 'join' request"
);
...
...
@@ -152,7 +140,7 @@ TESTEE(stream_multiplexer) {
using
fixture
=
test_coordinator_fixture
<>
;
}
// namespace
<anonymous>
}
// namespace
// -- unit tests ---------------------------------------------------------------
...
...
libcaf_core/test/fused_downstream_manager.cpp
View file @
6a47f60d
...
...
@@ -27,6 +27,7 @@
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/attach_stream_sink.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/stateful_actor.hpp"
...
...
@@ -98,7 +99,8 @@ TESTEE_STATE(sum_up) {
TESTEE
(
sum_up
)
{
using
intptr
=
int
*
;
return
{[
=
](
stream
<
int32_t
>&
in
)
{
return
self
->
make_sink
(
return
attach_stream_sink
(
self
,
// input stream
in
,
// initialize state
...
...
@@ -122,7 +124,8 @@ TESTEE_STATE(collect) {
TESTEE
(
collect
)
{
return
{[
=
](
stream
<
string
>&
in
)
{
return
self
->
make_sink
(
return
attach_stream_sink
(
self
,
// input stream
in
,
// initialize state
...
...
libcaf_core/test/pipeline_streaming.cpp
View file @
6a47f60d
...
...
@@ -25,6 +25,7 @@
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/attach_stream_sink.hpp"
#include "caf/attach_stream_stage.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/stateful_actor.hpp"
...
...
@@ -111,7 +112,8 @@ TESTEE_STATE(sum_up) {
TESTEE
(
sum_up
)
{
using
intptr
=
int
*
;
return
{[
=
](
stream
<
int
>&
in
)
{
return
self
->
make_sink
(
return
attach_stream_sink
(
self
,
// input stream
in
,
// initialize state
...
...
@@ -131,7 +133,8 @@ TESTEE(delayed_sum_up) {
return
{[
=
](
ok_atom
)
{
self
->
become
([
=
](
stream
<
int
>&
in
)
{
self
->
set_default_handler
(
print_and_drop
);
return
self
->
make_sink
(
return
attach_stream_sink
(
self
,
// input stream
in
,
// initialize state
...
...
libcaf_core/test/selective_streaming.cpp
View file @
6a47f60d
...
...
@@ -26,6 +26,7 @@
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/atom.hpp"
#include "caf/attach_stream_sink.hpp"
#include "caf/attach_stream_source.hpp"
#include "caf/broadcast_downstream_manager.hpp"
#include "caf/event_based_actor.hpp"
...
...
@@ -136,7 +137,8 @@ TESTEE_STATE(log_consumer) {
TESTEE
(
log_consumer
)
{
return
{[
=
](
stream
<
value_type
>&
in
)
{
return
self
->
make_sink
(
return
attach_stream_sink
(
self
,
// input stream
in
,
// initialize state
...
...
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