Commit 155ba7b4 authored by Dominik Charousset's avatar Dominik Charousset

Add attach_stream_source to replace make_source

parent 5d6cda55
......@@ -56,18 +56,18 @@ data to the next available worker.
\subsection{Defining Sources}
\cppexample[17-52]{streaming/integer_stream}
\cppexample[17-46]{streaming/integer_stream}
The simplest way to defining a source is to use the \lstinline^make_source^
function and pass it three arguments: \emph{initializer} for the state,
\emph{generator} for producing values, and \emph{predicate} for signaling the
end of the stream.
The simplest way to defining a source is to use the
\lstinline^attach_stream_source^ function and pass it four arguments: a pointer
to \emph{self}, \emph{initializer} for the state, \emph{generator} for
producing values, and \emph{predicate} for signaling the end of the stream.
\clearpage
\subsection{Defining Stages}
\cppexample[54-87]{streaming/integer_stream}
\cppexample[43-72]{streaming/integer_stream}
The function \lstinline^make_stage^ also takes three lambdas but additionally
the received input stream handshake as first argument. Instead of a predicate,
......@@ -78,7 +78,7 @@ data on its own and a stream terminates if no more sources exist.
\subsection{Defining Sinks}
\cppexample[89-120]{streaming/integer_stream}
\cppexample[73-100]{streaming/integer_stream}
The function \lstinline^make_sink^ is similar to \lstinline^make_stage^, except
that is does not produce outputs.
......@@ -87,7 +87,7 @@ that is does not produce outputs.
\subsection{Initiating Streams}
\cppexample[133-139]{streaming/integer_stream}
\cppexample[119-123]{streaming/integer_stream}
In our example, we always have a source \lstinline^int_source^ and a sink
\lstinline^int_sink^ with an optional stage \lstinline^int_selector^. Sending
......
......@@ -2,7 +2,7 @@
* Basic, non-interactive streaming example for processing integers. *
******************************************************************************/
// Manual refs: lines 17-52, 54-87, 89-120, 133-139 (Streaming)
// Manual refs: lines 17-46, 43-72, 73-100, 119-123 (Streaming)
#include <iostream>
#include <vector>
......@@ -16,114 +16,99 @@ 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;
}
);
}
};
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 attach_stream_source(
self,
// 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
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;
}
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");
.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;
......@@ -138,6 +123,6 @@ void caf_main(actor_system& sys, const config& cfg) {
anon_send(pipeline, open_atom::value, cfg.n);
}
} // namespace <anonymous>
} // namespace
CAF_MAIN()
......@@ -20,93 +20,94 @@
#include "caf/config.hpp"
#include "caf/sec.hpp"
#include "caf/atom.hpp"
#include "caf/send.hpp"
#include "caf/skip.hpp"
#include "caf/unit.hpp"
#include "caf/term.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/abstract_channel.hpp"
#include "caf/abstract_composable_behavior.hpp"
#include "caf/abstract_group.hpp"
#include "caf/actor.hpp"
#include "caf/after.hpp"
#include "caf/error.hpp"
#include "caf/group.hpp"
#include "caf/extend.hpp"
#include "caf/logger.hpp"
#include "caf/others.hpp"
#include "caf/result.hpp"
#include "caf/stream.hpp"
#include "caf/message.hpp"
#include "caf/node_id.hpp"
#include "caf/behavior.hpp"
#include "caf/defaults.hpp"
#include "caf/duration.hpp"
#include "caf/expected.hpp"
#include "caf/exec_main.hpp"
#include "caf/resumable.hpp"
#include "caf/streambuf.hpp"
#include "caf/to_string.hpp"
#include "caf/actor_addr.hpp"
#include "caf/actor_pool.hpp"
#include "caf/attachable.hpp"
#include "caf/message_id.hpp"
#include "caf/replies_to.hpp"
#include "caf/serializer.hpp"
#include "caf/actor_clock.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/actor_pool.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/exit_reason.hpp"
#include "caf/local_actor.hpp"
#include "caf/raise_error.hpp"
#include "caf/ref_counted.hpp"
#include "caf/stream_slot.hpp"
#include "caf/thread_hook.hpp"
#include "caf/typed_actor.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/after.hpp"
#include "caf/atom.hpp"
#include "caf/attach_stream_source.hpp"
#include "caf/attachable.hpp"
#include "caf/behavior.hpp"
#include "caf/behavior_policy.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/blocking_actor.hpp"
#include "caf/composable_behavior.hpp"
#include "caf/composed_behavior.hpp"
#include "caf/config_option.hpp"
#include "caf/config_option_adder.hpp"
#include "caf/config_value.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/defaults.hpp"
#include "caf/deserializer.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/upstream_msg.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/config_option.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/duration.hpp"
#include "caf/error.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/exec_main.hpp"
#include "caf/execution_unit.hpp"
#include "caf/exit_reason.hpp"
#include "caf/expected.hpp"
#include "caf/extend.hpp"
#include "caf/function_view.hpp"
#include "caf/fused_downstream_manager.hpp"
#include "caf/group.hpp"
#include "caf/index_mapping.hpp"
#include "caf/spawn_options.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/abstract_group.hpp"
#include "caf/blocking_actor.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/execution_unit.hpp"
#include "caf/local_actor.hpp"
#include "caf/logger.hpp"
#include "caf/make_config_option.hpp"
#include "caf/may_have_timeout.hpp"
#include "caf/memory_managed.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/behavior_policy.hpp"
#include "caf/message.hpp"
#include "caf/message_builder.hpp"
#include "caf/message_handler.hpp"
#include "caf/response_handle.hpp"
#include "caf/system_messages.hpp"
#include "caf/abstract_channel.hpp"
#include "caf/may_have_timeout.hpp"
#include "caf/message_id.hpp"
#include "caf/message_priority.hpp"
#include "caf/typed_actor_view.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/composed_behavior.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/node_id.hpp"
#include "caf/others.hpp"
#include "caf/primitive_variant.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/raise_error.hpp"
#include "caf/ref_counted.hpp"
#include "caf/replies_to.hpp"
#include "caf/response_handle.hpp"
#include "caf/result.hpp"
#include "caf/resumable.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/scoped_execution_unit.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
#include "caf/serializer.hpp"
#include "caf/skip.hpp"
#include "caf/spawn_options.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/stream.hpp"
#include "caf/stream_deserializer.hpp"
#include "caf/stream_serializer.hpp"
#include "caf/make_config_option.hpp"
#include "caf/stream_slot.hpp"
#include "caf/streambuf.hpp"
#include "caf/system_messages.hpp"
#include "caf/term.hpp"
#include "caf/thread_hook.hpp"
#include "caf/timeout_definition.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/composable_behavior.hpp"
#include "caf/config_option_adder.hpp"
#include "caf/stream_deserializer.hpp"
#include "caf/to_string.hpp"
#include "caf/typed_actor.hpp"
#include "caf/typed_actor_pointer.hpp"
#include "caf/scoped_execution_unit.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/typed_actor_view.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/fused_downstream_manager.hpp"
#include "caf/abstract_composable_behavior.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/unit.hpp"
#include "caf/upstream_msg.hpp"
#include "caf/decorator/sequencer.hpp"
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <tuple>
#include "caf/broadcast_downstream_manager.hpp"
#include "caf/default_downstream_manager.hpp"
#include "caf/detail/stream_source_driver_impl.hpp"
#include "caf/detail/stream_source_impl.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/fwd.hpp"
#include "caf/is_actor_handle.hpp"
#include "caf/make_source_result.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream_source.hpp"
namespace caf {
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with `Driver`.
/// @param self Points to the hosting actor.
/// @param xs User-defined arguments for the stream handshake.
/// @param init Function object for initializing the state of the source.
/// @param pull Function object for generating downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Optional cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
template <class Driver, class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t>
make_source_result_t<typename Driver::downstream_manager_type, Ts...>
attach_stream_source(scheduled_actor* self, std::tuple<Ts...> xs, Init init,
Pull pull, Done done, Finalize fin = {}) {
using namespace detail;
auto mgr = make_stream_source<Driver>(self, std::move(init), std::move(pull),
std::move(done), std::move(fin));
auto slot = mgr->add_outbound_path(std::move(xs));
return {slot, std::move(mgr)};
}
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with the default driver.
/// @param self Points to the hosting actor.
/// @param xs User-defined arguments for the stream handshake.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
template <class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t,
class DownstreamManager = broadcast_downstream_manager<
typename stream_source_trait_t<Pull>::output>>
make_source_result_t<DownstreamManager, Ts...>
attach_stream_source(scheduled_actor* self, std::tuple<Ts...> xs, Init init,
Pull pull, Done done, Finalize fin = {},
policy::arg<DownstreamManager> = {}) {
using driver = detail::stream_source_driver_impl<DownstreamManager, Pull,
Done, Finalize>;
return attach_stream_source<driver>(self, std::move(xs), std::move(init),
std::move(pull), std::move(done),
std::move(fin));
}
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with the default driver.
/// @param self Points to the hosting actor.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
template <class Init, class Pull, class Done, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<!is_actor_handle<Init>::value && Trait::valid,
make_source_result_t<DownstreamManager>>
attach_stream_source(scheduled_actor* self, Init init, Pull pull, Done done,
Finalize finalize = {},
policy::arg<DownstreamManager> token = {}) {
return attach_stream_source(self, std::make_tuple(), init, pull, done,
finalize, token);
}
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with the default driver and starts sending to `dest` immediately.
/// @param self Points to the hosting actor.
/// @param dest Handle to the next stage in the pipeline.
/// @param xs User-defined arguments for the stream handshake.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
template <class ActorHandle, class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<is_actor_handle<ActorHandle>::value,
make_source_result_t<DownstreamManager>>
attach_stream_source(scheduled_actor* self, const ActorHandle& dest,
std::tuple<Ts...> xs, Init init, Pull pull, Done done,
Finalize fin = {}, policy::arg<DownstreamManager> = {}) {
// TODO: type checking of dest
using driver = detail::stream_source_driver_impl<DownstreamManager, Pull,
Done, Finalize>;
auto mgr = detail::make_stream_source<driver>(self, std::move(init),
std::move(pull),
std::move(done),
std::move(fin));
auto slot = mgr->add_outbound_path(dest, std::move(xs));
return {slot, std::move(mgr)};
}
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with the default driver and starts sending to `dest` immediately.
/// @param self Points to the hosting actor.
/// @param dest Handle to the next stage in the pipeline.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
template <class ActorHandle, class Init, class Pull, class Done,
class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<is_actor_handle<ActorHandle>::value && Trait::valid,
make_source_result_t<DownstreamManager>>
attach_stream_source(scheduled_actor* self, const ActorHandle& dest, Init init,
Pull pull, Done done, Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
return attach_stream_source(self, dest, std::make_tuple(), std::move(init),
std::move(pull), std::move(done), std::move(fin),
token);
}
} // namespace caf
......@@ -785,6 +785,38 @@ struct is_list_like {
&& !has_mapped_type_alias<T>::value;
};
template <class F, class... Ts>
struct is_invocable {
private:
template <class U>
static auto sfinae(U* f)
-> decltype((*f)(std::declval<Ts>()...), std::true_type());
template <class U>
static auto sfinae(...) -> std::false_type;
using sfinae_type = decltype(sfinae<F>(nullptr));
public:
static constexpr bool value = sfinae_type::value;
};
template <class R, class F, class... Ts>
struct is_invocable_r {
private:
template <class U>
static auto sfinae(U* f)
-> std::is_same<R, decltype((*f)(std::declval<Ts>()...))>;
template <class U>
static auto sfinae(...) -> std::false_type;
using sfinae_type = decltype(sfinae<F>(nullptr));
public:
static constexpr bool value = sfinae_type::value;
};
} // namespace detail
} // namespace caf
......
......@@ -420,14 +420,7 @@ public:
// -- stream management ------------------------------------------------------
/// Creates a new stream source by instantiating the default source
/// implementation with `Driver`.
/// @param xs User-defined handshake payload.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
/// @deprecated Please use `attach_stream_source` instead.
template <class Driver, class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t>
make_source_result_t<typename Driver::downstream_manager_type, Ts...>
......@@ -441,13 +434,7 @@ public:
return {slot, std::move(mgr)};
}
/// Creates a new stream source from given arguments.
/// @param xs User-defined handshake payload.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
/// @deprecated Please use `attach_stream_source` instead.
template <class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t,
class DownstreamManager = broadcast_downstream_manager<
......@@ -461,6 +448,7 @@ public:
std::move(done), std::move(fin));
}
/// @deprecated Please use `attach_stream_source` instead.
template <class Init, class Pull, class Done, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
class Trait = stream_source_trait_t<Pull>>
......@@ -472,7 +460,7 @@ public:
token);
}
/// Creates a new stream source and adds `dest` as first outbound path to it.
/// @deprecated Please use `attach_stream_source` instead.
template <class ActorHandle, class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
......@@ -493,7 +481,7 @@ public:
return {slot, std::move(mgr)};
}
/// Creates a new stream source and adds `dest` as first outbound path to it.
/// @deprecated Please use `attach_stream_source` instead.
template <class ActorHandle, class Init, class Pull, class Done,
class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
......
......@@ -50,7 +50,8 @@ VARARGS_TESTEE(file_reader, size_t buf_size) {
[=](string& fname) -> output_stream<int, string> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return self->make_source(
return attach_stream_source(
self,
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
// initialize state
......
......@@ -16,7 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE fused_streaming
#define CAF_SUITE fused_downstream_manager
#include "caf/fused_downstream_manager.hpp"
#include "caf/test/dsl.hpp"
......@@ -26,7 +28,6 @@
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/fused_downstream_manager.hpp"
#include "caf/stateful_actor.hpp"
using std::string;
......@@ -56,48 +57,38 @@ void push(std::deque<T>& xs, downstream<T>& out, size_t num) {
VARARGS_TESTEE(int_file_reader, size_t buf_size) {
using buf = std::deque<int32_t>;
return {
[=](string& fname) -> output_stream<int32_t> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->make_source(
// initialize state
[=](buf& xs) {
xs.resize(buf_size);
std::iota(xs.begin(), xs.end(), 1);
},
// get next element
[](buf& xs, downstream<int32_t>& out, size_t num) {
push(xs, out, num);
},
// check whether we reached the end
[=](const buf& xs) {
return xs.empty();
});
}
};
return {[=](string& fname) -> output_stream<int32_t> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return attach_stream_source(
self,
// initialize state
[=](buf& xs) {
xs.resize(buf_size);
std::iota(xs.begin(), xs.end(), 1);
},
// get next element
[](buf& xs, downstream<int32_t>& out, size_t num) { push(xs, out, num); },
// check whether we reached the end
[=](const buf& xs) { return xs.empty(); });
}};
}
VARARGS_TESTEE(string_file_reader, size_t buf_size) {
using buf = std::deque<string>;
return {
[=](string& fname) -> output_stream<string> {
CAF_CHECK_EQUAL(fname, "strings.txt");
return self->make_source(
// initialize state
[=](buf& xs) {
for (size_t i = 0; i < buf_size; ++i)
xs.emplace_back("some string data");
},
// get next element
[](buf& xs, downstream<string>& out, size_t num) {
push(xs, out, num);
},
// check whether we reached the end
[=](const buf& xs) {
return xs.empty();
});
}
};
return {[=](string& fname) -> output_stream<string> {
CAF_CHECK_EQUAL(fname, "strings.txt");
return attach_stream_source(
self,
// initialize state
[=](buf& xs) {
for (size_t i = 0; i < buf_size; ++i)
xs.emplace_back("some string data");
},
// get next element
[](buf& xs, downstream<string>& out, size_t num) { push(xs, out, num); },
// check whether we reached the end
[=](const buf& xs) { return xs.empty(); });
}};
}
TESTEE_STATE(sum_up) {
......@@ -106,30 +97,23 @@ TESTEE_STATE(sum_up) {
TESTEE(sum_up) {
using intptr = int*;
return {
[=](stream<int32_t>& in) {
return self->make_sink(
// input stream
in,
// initialize state
[=](intptr& x) {
x = &self->state.x;
},
// processing step
[](intptr& x, int32_t y) {
*x += y;
},
// cleanup and produce result message
[=](intptr&, const error&) {
CAF_MESSAGE(self->name() << " is done");
}
);
},
[=](join_atom, actor src) {
CAF_MESSAGE(self->name() << " joins a stream");
self->send(self * src, join_atom::value, ints_atom::value);
}
};
return {[=](stream<int32_t>& in) {
return self->make_sink(
// input stream
in,
// initialize state
[=](intptr& x) { x = &self->state.x; },
// processing step
[](intptr& x, int32_t y) { *x += y; },
// cleanup and produce result message
[=](intptr&, const error&) {
CAF_MESSAGE(self->name() << " is done");
});
},
[=](join_atom, actor src) {
CAF_MESSAGE(self->name() << " joins a stream");
self->send(self * src, join_atom::value, ints_atom::value);
}};
}
TESTEE_STATE(collect) {
......@@ -137,38 +121,35 @@ TESTEE_STATE(collect) {
};
TESTEE(collect) {
return {
[=](stream<string>& in) {
return self->make_sink(
// input stream
in,
// initialize state
[](unit_t&) {
// nop
},
// processing step
[=](unit_t&, string y) {
self->state.strings.emplace_back(std::move(y));
},
// cleanup and produce result message
[=](unit_t&, const error&) {
CAF_MESSAGE(self->name() << " is done");
}
);
},
[=](join_atom, actor src) {
CAF_MESSAGE(self->name() << " joins a stream");
self->send(self * src, join_atom::value, strings_atom::value);
}
};
return {[=](stream<string>& in) {
return self->make_sink(
// input stream
in,
// initialize state
[](unit_t&) {
// nop
},
// processing step
[=](unit_t&, string y) {
self->state.strings.emplace_back(std::move(y));
},
// cleanup and produce result message
[=](unit_t&, const error&) {
CAF_MESSAGE(self->name() << " is done");
});
},
[=](join_atom, actor src) {
CAF_MESSAGE(self->name() << " joins a stream");
self->send(self * src, join_atom::value, strings_atom::value);
}};
}
using int_downstream_manager = broadcast_downstream_manager<int>;
using string_downstream_manager = broadcast_downstream_manager<string>;
using fused_manager =
fused_downstream_manager<int_downstream_manager, string_downstream_manager>;
using fused_manager = fused_downstream_manager<int_downstream_manager,
string_downstream_manager>;
class fused_stage : public stream_manager {
public:
......@@ -226,30 +207,28 @@ TESTEE_STATE(stream_multiplexer) {
TESTEE(stream_multiplexer) {
self->state.stage = make_counted<fused_stage>(self);
return {
[=](join_atom, ints_atom) {
auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request for integers");
auto result = stg->add_unchecked_outbound_path<int>();
stg->out().assign<int_downstream_manager>(result);
return result;
},
[=](join_atom, strings_atom) {
auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request for strings");
auto result = stg->add_unchecked_outbound_path<string>();
stg->out().assign<string_downstream_manager>(result);
return result;
},
[=](const stream<int32_t>& in) {
CAF_MESSAGE("received handshake for integers");
return self->state.stage->add_unchecked_inbound_path(in);
},
[=](const stream<string>& in) {
CAF_MESSAGE("received handshake for strings");
return self->state.stage->add_unchecked_inbound_path(in);
}
};
return {[=](join_atom, ints_atom) {
auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request for integers");
auto result = stg->add_unchecked_outbound_path<int>();
stg->out().assign<int_downstream_manager>(result);
return result;
},
[=](join_atom, strings_atom) {
auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request for strings");
auto result = stg->add_unchecked_outbound_path<string>();
stg->out().assign<string_downstream_manager>(result);
return result;
},
[=](const stream<int32_t>& in) {
CAF_MESSAGE("received handshake for integers");
return self->state.stage->add_unchecked_inbound_path(in);
},
[=](const stream<string>& in) {
CAF_MESSAGE("received handshake for strings");
return self->state.stage->add_unchecked_inbound_path(in);
}};
}
struct config : actor_system_config {
......@@ -260,11 +239,11 @@ struct config : actor_system_config {
using fixture = test_coordinator_fixture<>;
} // namespace <anonymous>
} // namespace
// -- unit tests ---------------------------------------------------------------
CAF_TEST_FIXTURE_SCOPE(fused_streaming_tests, fixture)
CAF_TEST_FIXTURE_SCOPE(fused_downstream_manager_tests, fixture)
CAF_TEST(depth_3_pipeline_with_fork) {
auto src1 = sys.spawn(int_file_reader, 50u);
......
......@@ -64,7 +64,7 @@ std::function<bool(const buf&)> is_done(scheduled_actor* self) {
}
template <class T>
std::function<void (T&, const error&)> fin(scheduled_actor* self) {
std::function<void(T&, const error&)> fin(scheduled_actor* self) {
return [=](T&, const error& err) {
if (err == none) {
CAF_MESSAGE(self->name() << " is done");
......@@ -75,51 +75,32 @@ std::function<void (T&, const error&)> fin(scheduled_actor* self) {
}
TESTEE(infinite_source) {
return {
[=](string& fname) -> output_stream<int> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return self->make_source(
[](int& x) {
x = 0;
},
[](int& x, downstream<int>& out, size_t num) {
for (size_t i = 0; i < num; ++i)
out.push(x++);
},
[](const int&) {
return false;
},
fin<int>(self)
);
}
};
return {[=](string& fname) -> output_stream<int> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return attach_stream_source(
self, [](int& x) { x = 0; },
[](int& x, downstream<int>& out, size_t num) {
for (size_t i = 0; i < num; ++i)
out.push(x++);
},
[](const int&) { return false; }, fin<int>(self));
}};
}
VARARGS_TESTEE(file_reader, size_t buf_size) {
return {
[=](string& fname) -> output_stream<int> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return self->make_source(
init(buf_size),
push_from_buf,
is_done(self),
fin<buf>(self)
);
},
[=](string& fname, actor next) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
self->make_source(
next,
init(buf_size),
push_from_buf,
is_done(self),
fin<buf>(self)
);
}
};
return {[=](string& fname) -> output_stream<int> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return attach_stream_source(self, init(buf_size), push_from_buf,
is_done(self), fin<buf>(self));
},
[=](string& fname, actor next) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
attach_stream_source(self, next, init(buf_size), push_from_buf,
is_done(self), fin<buf>(self));
}};
}
TESTEE_STATE(sum_up) {
......@@ -128,23 +109,15 @@ TESTEE_STATE(sum_up) {
TESTEE(sum_up) {
using intptr = int*;
return {
[=](stream<int>& in) {
return self->make_sink(
// input stream
in,
// initialize state
[=](intptr& x) {
x = &self->state.x;
},
// processing step
[](intptr& x, int y) {
*x += y;
},
fin<intptr>(self)
);
}
};
return {[=](stream<int>& in) {
return self->make_sink(
// input stream
in,
// initialize state
[=](intptr& x) { x = &self->state.x; },
// processing step
[](intptr& x, int y) { *x += y; }, fin<intptr>(self));
}};
}
TESTEE_STATE(delayed_sum_up) {
......@@ -154,83 +127,64 @@ TESTEE_STATE(delayed_sum_up) {
TESTEE(delayed_sum_up) {
using intptr = int*;
self->set_default_handler(skip);
return {
[=](ok_atom) {
self->become(
[=](stream<int>& in) {
self->set_default_handler(print_and_drop);
return self->make_sink(
// input stream
in,
// initialize state
[=](intptr& x) {
x = &self->state.x;
},
// processing step
[](intptr& x, int y) {
*x += y;
},
// cleanup
fin<intptr>(self)
);
}
);
}
};
return {[=](ok_atom) {
self->become([=](stream<int>& in) {
self->set_default_handler(print_and_drop);
return self->make_sink(
// input stream
in,
// initialize state
[=](intptr& x) { x = &self->state.x; },
// processing step
[](intptr& x, int y) { *x += y; },
// cleanup
fin<intptr>(self));
});
}};
}
TESTEE(broken_sink) {
CAF_IGNORE_UNUSED(self);
return {
[=](stream<int>&, const actor&) {
// nop
}
};
return {[=](stream<int>&, const actor&) {
// nop
}};
}
TESTEE(filter) {
CAF_IGNORE_UNUSED(self);
return {
[=](stream<int>& in) {
return self->make_stage(
// input stream
in,
// initialize state
[](unit_t&) {
// nop
},
// processing step
[](unit_t&, downstream<int>& out, int x) {
if ((x & 0x01) != 0)
out.push(x);
},
// cleanup
fin<unit_t>(self)
);
}
};
return {[=](stream<int>& in) {
return self->make_stage(
// input stream
in,
// initialize state
[](unit_t&) {
// nop
},
// processing step
[](unit_t&, downstream<int>& out, int x) {
if ((x & 0x01) != 0)
out.push(x);
},
// cleanup
fin<unit_t>(self));
}};
}
TESTEE(doubler) {
CAF_IGNORE_UNUSED(self);
return {
[=](stream<int>& in) {
return self->make_stage(
// input stream
in,
// initialize state
[](unit_t&) {
// nop
},
// processing step
[](unit_t&, downstream<int>& out, int x) {
out.push(x * 2);
},
// cleanup
fin<unit_t>(self)
);
}
};
return {[=](stream<int>& in) {
return self->make_stage(
// input stream
in,
// initialize state
[](unit_t&) {
// nop
},
// processing step
[](unit_t&, downstream<int>& out, int x) { out.push(x * 2); },
// cleanup
fin<unit_t>(self));
}};
}
struct fixture : test_coordinator_fixture<> {
......@@ -245,7 +199,7 @@ struct fixture : test_coordinator_fixture<> {
}
};
} // namespace <anonymous>
} // namespace
// -- unit tests ---------------------------------------------------------------
......@@ -426,8 +380,8 @@ CAF_TEST(depth_4_pipeline_500_items) {
auto stg1 = sys.spawn(filter);
auto stg2 = sys.spawn(doubler);
auto snk = sys.spawn(sum_up);
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(stg1)
<< CAF_ARG(stg2) << CAF_ARG(snk));
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(stg1) << CAF_ARG(stg2)
<< CAF_ARG(snk));
CAF_MESSAGE("initiate stream handshake");
self->send(snk * stg2 * stg1 * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
......
......@@ -26,6 +26,7 @@
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/atom.hpp"
#include "caf/attach_stream_source.hpp"
#include "caf/broadcast_downstream_manager.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/stateful_actor.hpp"
......@@ -36,13 +37,7 @@ using namespace caf;
namespace {
enum class level {
all,
trace,
debug,
warning,
error
};
enum class level { all, trace, debug, warning, error };
using value_type = std::pair<level, string>;
......@@ -55,7 +50,7 @@ struct select {
}
};
using downstream_manager = broadcast_downstream_manager<value_type, level, select>;
using manager_type = broadcast_downstream_manager<value_type, level, select>;
using buf = std::vector<value_type>;
......@@ -76,43 +71,38 @@ buf make_log(level lvl) {
TESTEE_SETUP();
TESTEE(log_producer) {
return {
[=](level lvl) -> output_stream<value_type> {
auto res = self->make_source(
// initialize state
[=](buf& xs) {
xs = make_log(lvl);
},
// get next element
[](buf& xs, downstream<value_type>& out, size_t num) {
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 buf& xs) {
if (xs.empty()) {
CAF_MESSAGE(self->name() << " is done");
return true;
}
return false;
},
unit,
policy::arg<downstream_manager>::value
);
auto& out = res.ptr()->out();
static_assert(std::is_same<decltype(out), downstream_manager&>::value,
"source has wrong downstream_manager type");
out.set_filter(res.outbound_slot(), lvl);
return res;
}
};
return {[=](level lvl) -> output_stream<value_type> {
auto res = attach_stream_source(
self,
// initialize state
[=](buf& xs) { xs = make_log(lvl); },
// get next element
[](buf& xs, downstream<value_type>& out, size_t num) {
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 buf& xs) {
if (xs.empty()) {
CAF_MESSAGE(self->name() << " is done");
return true;
}
return false;
},
unit, policy::arg<manager_type>::value);
auto& out = res.ptr()->out();
static_assert(std::is_same<decltype(out), manager_type&>::value,
"source has wrong manager_type type");
out.set_filter(res.outbound_slot(), lvl);
return res;
}};
}
TESTEE_STATE(log_dispatcher) {
stream_stage_ptr<value_type, downstream_manager> stage;
stream_stage_ptr<value_type, manager_type> stage;
};
TESTEE(log_dispatcher) {
......@@ -126,23 +116,18 @@ TESTEE(log_dispatcher) {
out.push(std::move(x));
},
// cleanup
[=](unit_t&, const error&) {
CAF_MESSAGE(self->name() << " is done");
},
policy::arg<downstream_manager>::value
);
return {
[=](join_atom, level lvl) {
auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request");
auto result = stg->add_outbound_path();
stg->out().set_filter(result, lvl);
return result;
},
[=](const stream<value_type>& in) {
self->state.stage->add_inbound_path(in);
}
};
[=](unit_t&, const error&) { CAF_MESSAGE(self->name() << " is done"); },
policy::arg<manager_type>::value);
return {[=](join_atom, level lvl) {
auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request");
auto result = stg->add_outbound_path();
stg->out().set_filter(result, lvl);
return result;
},
[=](const stream<value_type>& in) {
self->state.stage->add_inbound_path(in);
}};
}
TESTEE_STATE(log_consumer) {
......@@ -150,26 +135,21 @@ TESTEE_STATE(log_consumer) {
};
TESTEE(log_consumer) {
return {
[=](stream<value_type>& in) {
return self->make_sink(
// input stream
in,
// initialize state
[=](unit_t&) {
// nop
},
// processing step
[=](unit_t&, value_type x) {
self->state.log.emplace_back(std::move(x));
},
// cleanup and produce result message
[=](unit_t&, const error&) {
CAF_MESSAGE(self->name() << " is done");
}
);
}
};
return {[=](stream<value_type>& in) {
return self->make_sink(
// input stream
in,
// initialize state
[=](unit_t&) {
// nop
},
// processing step
[=](unit_t&, value_type x) {
self->state.log.emplace_back(std::move(x));
},
// cleanup and produce result message
[=](unit_t&, const error&) { CAF_MESSAGE(self->name() << " is done"); });
}};
}
struct config : actor_system_config {
......@@ -180,7 +160,7 @@ struct config : actor_system_config {
using fixture = test_coordinator_fixture<config>;
} // namespace <anonymous>
} // namespace
// -- unit tests ---------------------------------------------------------------
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment