Unverified Commit b877c738 authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #909

Support streaming in composable behaviors
parents 94a262f9 e24ad712
......@@ -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[48-78]{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[80-106]{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[121-125]{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, 48-78, 80-107, 121-125 (Streaming)
#include <iostream>
#include <vector>
......@@ -16,114 +16,101 @@ 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 attach_stream_stage(
self,
// 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 attach_stream_sink(
self,
// 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 +125,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,98 @@
#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_continuous_stream_source.hpp"
#include "caf/attach_continuous_stream_stage.hpp"
#include "caf/attach_stream_sink.hpp"
#include "caf/attach_stream_source.hpp"
#include "caf/attach_stream_stage.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 "caf/broadcast_downstream_manager.hpp"
#include "caf/detail/stream_source_driver_impl.hpp"
#include "caf/detail/stream_source_impl.hpp"
#include "caf/fwd.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream_source.hpp"
#include "caf/stream_source_driver.hpp"
#include "caf/stream_source_trait.hpp"
namespace caf {
/// Creates a new continuous stream source by instantiating the default source
/// implementation with `Driver`. The returned manager is not connected to any
/// slot and thus not stored by the actor automatically.
/// @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 new `stream_manager`.
template <class Driver, class... Ts>
typename Driver::source_ptr_type
attach_continuous_stream_source(scheduled_actor* self, Ts&&... xs) {
using detail::make_stream_source;
auto mgr = make_stream_source<Driver>(self, std::forward<Ts>(xs)...);
mgr->continuous(true);
return mgr;
}
/// Creates a new continuous stream source by instantiating the default source
/// implementation with `Driver`. The returned manager is not connected to any
/// slot and thus not stored by the actor automatically.
/// @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 new `stream_manager`.
template <class Init, class Pull, class Done, class Finalize = unit_t,
class Trait = stream_source_trait_t<Pull>,
class DownstreamManager = broadcast_downstream_manager<
typename Trait::output>>
stream_source_ptr<DownstreamManager>
attach_continuous_stream_source(scheduled_actor* self, Init init, Pull pull,
Done done, Finalize fin = {},
policy::arg<DownstreamManager> = {}) {
using state_type = typename Trait::state;
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<Done>::fun_sig>::value,
"Expected signature `bool (const State&)` "
"for done predicate function");
using driver = detail::stream_source_driver_impl<DownstreamManager, Pull,
Done, Finalize>;
return attach_continuous_stream_source<driver>(self, std::move(init),
std::move(pull),
std::move(done),
std::move(fin));
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_stage_driver_impl.hpp"
#include "caf/detail/stream_stage_impl.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/fwd.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream.hpp"
#include "caf/stream_stage.hpp"
#include "caf/unit.hpp"
namespace caf {
/// Returns a stream manager (implementing a continuous stage) without in- or
/// outbound path. The returned manager is not connected to any slot and thus
/// not stored by the actor automatically.
/// @param self Points to the hosting actor.
/// @param xs User-defined arguments for the downstream handshake.
/// @returns The new `stream_manager`.
template <class Driver, class... Ts>
typename Driver::stage_ptr_type
attach_continuous_stream_stage(scheduled_actor* self, Ts&&... xs) {
auto ptr = detail::make_stream_stage<Driver>(self, std::forward<Ts>(xs)...);
ptr->continuous(true);
return ptr;
}
/// @param self Points to the hosting actor.
/// @param init Function object for initializing the state of the stage.
/// @param fun Processing function.
/// @param fin Optional cleanup handler.
/// @param token Policy token for selecting a downstream manager
/// implementation.
template <class Init, class Fun, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
stream_stage_ptr<typename Trait::input, DownstreamManager>
attach_continuous_stream_stage(scheduled_actor* self, Init init, Fun fun,
Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
CAF_IGNORE_UNUSED(token);
using input_type = typename Trait::input;
using output_type = typename Trait::output;
using state_type = typename Trait::state;
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<
void(state_type&, downstream<output_type>&, input_type),
typename detail::get_callable_trait<Fun>::fun_sig>::value,
"Expected signature `void (State&, downstream<Out>&, In)` "
"for consume function");
using detail::stream_stage_driver_impl;
using driver = stream_stage_driver_impl<typename Trait::input,
DownstreamManager, Fun, Finalize>;
return attach_continuous_stream_stage<driver>(self, std::move(init),
std::move(fun), std::move(fin));
}
} // namespace caf
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* 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 *
......@@ -18,14 +18,45 @@
#pragma once
#include "caf/detail/stream_sink_driver_impl.hpp"
#include "caf/detail/stream_sink_impl.hpp"
#include "caf/fwd.hpp"
#include "caf/make_sink_result.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream.hpp"
#include "caf/stream_sink.hpp"
namespace caf {
/// Empty marker type for type-checking of stream sources and stages.
template <class T, class... Ts>
class output_stream {
public:
using value_type = T;
};
/// 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)};
}
} // 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 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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/response_type.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... CtorArgs>
make_source_result_t<typename Driver::downstream_manager_type, Ts...>
attach_stream_source(scheduled_actor* self, std::tuple<Ts...> xs,
CtorArgs&&... ctor_args) {
using namespace detail;
auto mgr = make_stream_source<Driver>(self,
std::forward<CtorArgs>(ctor_args)...);
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 Trait = stream_source_trait_t<Pull>,
class DownstreamManager = broadcast_downstream_manager<
typename Trait::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 state_type = typename Trait::state;
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<Done>::fun_sig>::value,
"Expected signature `bool (const State&)` "
"for done predicate function");
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> = {}) {
using namespace detail;
using token = type_list<stream<typename DownstreamManager::output_type>,
strip_and_convert_t<Ts>...>;
static_assert(response_type_unbox<signatures_of_t<ActorHandle>, token>::valid,
"receiver does not accept the stream handshake");
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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_stage_driver_impl.hpp"
#include "caf/detail/stream_stage_impl.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/fwd.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream.hpp"
#include "caf/stream_stage.hpp"
#include "caf/unit.hpp"
namespace caf {
/// Attaches a new stream stage to `self` by creating a default stream stage
/// manager with `Driver`.
/// @param self Points to the hosting actor.
/// @param in Stream handshake from upstream path.
/// @param xs User-defined arguments for the downstream handshake.
/// @param ys Additional constructor arguments for `Driver`.
/// @returns The new `stream_manager`, an inbound slot, and an outbound slot.
template <class Driver, class In, class... Ts, class... Us>
make_stage_result_t<In, typename Driver::downstream_manager_type, Ts...>
attach_stream_stage(scheduled_actor* self, const stream<In>& in,
std::tuple<Ts...> xs, Us&&... ys) {
using detail::make_stream_stage;
auto mgr = make_stream_stage<Driver>(self, std::forward<Us>(ys)...);
auto islot = mgr->add_inbound_path(in);
auto oslot = mgr->add_outbound_path(std::move(xs));
return {islot, oslot, std::move(mgr)};
}
/// Attaches a new stream stage to `self` by creating a default stream stage
/// manager from given callbacks.
/// @param self Points to the hosting actor.
/// @param in Stream handshake from upstream path.
/// @param xs User-defined arguments for the downstream handshake.
/// @param init Function object for initializing the state of the stage.
/// @param fun Processing function.
/// @param fin Optional cleanup handler.
/// @param token Policy token for selecting a downstream manager
/// implementation.
/// @returns The new `stream_manager`, an inbound slot, and an outbound slot.
template <class In, class... Ts, class Init, class Fun, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
make_stage_result_t<In, DownstreamManager, Ts...>
attach_stream_stage(scheduled_actor* self, const stream<In>& in,
std::tuple<Ts...> xs, Init init, Fun fun, Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
CAF_IGNORE_UNUSED(token);
using output_type = typename stream_stage_trait_t<Fun>::output;
using state_type = typename stream_stage_trait_t<Fun>::state;
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<
void(state_type&, downstream<output_type>&, In),
typename detail::get_callable_trait<Fun>::fun_sig>::value,
"Expected signature `void (State&, downstream<Out>&, In)` "
"for consume function");
using driver = detail::stream_stage_driver_impl<
typename Trait::input, DownstreamManager, Fun, Finalize>;
return attach_stream_stage<driver>(self, in, std::move(xs), std::move(init),
std::move(fun), std::move(fin));
}
/// Attaches a new stream stage to `self` by creating a default stream stage
/// 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 stage.
/// @param fun Processing function.
/// @param fin Optional cleanup handler.
/// @param token Policy token for selecting a downstream manager
/// implementation.
/// @returns The new `stream_manager`, an inbound slot, and an outbound slot.
template <class In, class Init, class Fun, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
make_stage_result_t<In, DownstreamManager>
attach_stream_stage(scheduled_actor* self, const stream<In>& in, Init init,
Fun fun, Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
return attach_stream_stage(self, in, std::make_tuple(), std::move(init),
std::move(fun), std::move(fin), token);
}
} // namespace caf
......@@ -18,12 +18,12 @@
#pragma once
#include "caf/param.hpp"
#include "caf/abstract_composable_behavior.hpp"
#include "caf/behavior.hpp"
#include "caf/param.hpp"
#include "caf/replies_to.hpp"
#include "caf/typed_actor.hpp"
#include "caf/typed_actor_pointer.hpp"
#include "caf/abstract_composable_behavior.hpp"
namespace caf {
......@@ -34,8 +34,8 @@ template <class MPI>
class composable_behavior_base;
template <class... Xs, class... Ys>
class composable_behavior_base<typed_mpi<detail::type_list<Xs...>,
output_tuple<Ys...>>> {
class composable_behavior_base<
typed_mpi<detail::type_list<Xs...>, output_tuple<Ys...>>> {
public:
virtual ~composable_behavior_base() noexcept {
// nop
......@@ -44,16 +44,16 @@ public:
virtual result<Ys...> operator()(param_t<Xs>...) = 0;
// C++14 and later
# if __cplusplus > 201103L
#if __cplusplus > 201103L
auto make_callback() {
return [=](param_t<Xs>... xs) { return (*this)(std::move(xs)...); };
}
# else
#else
// C++11
std::function<result<Ys...> (param_t<Xs>...)> make_callback() {
std::function<result<Ys...>(param_t<Xs>...)> make_callback() {
return [=](param_t<Xs>... xs) { return (*this)(std::move(xs)...); };
}
# endif
#endif
};
/// Base type for composable actor states.
......@@ -67,11 +67,7 @@ class composable_behavior<typed_actor<Clauses...>>
public:
using signatures = detail::type_list<Clauses...>;
using handle_type =
typename detail::tl_apply<
signatures,
typed_actor
>::type;
using handle_type = typename detail::tl_apply<signatures, typed_actor>::type;
using actor_base = typename handle_type::base;
......@@ -106,4 +102,3 @@ public:
};
} // namespace caf
......@@ -18,122 +18,71 @@
#pragma once
#include "caf/detail/type_list.hpp"
#include "caf/fwd.hpp"
#include "caf/replies_to.hpp"
#include "caf/detail/type_list.hpp"
namespace caf {
/// Computes the type for f*g (actor composition).
///
/// ~~~
/// let output_type x = case x of Stream y -> y ; Single y -> y
/// This metaprogramming function implements the following pseudo-code (with
/// `f` and `g` modelled as pairs where the first element is the input type and
/// the second type is the output type).
///
/// let propagate_stream from to = case from of
/// Stream _ -> Stream (output_type to)
/// Single _ -> to
/// ~~~
/// let composed_type f g =
/// [(fst x, propagate_stream (snd x) (snd y)) | x <- g, y <- f,
/// output_type (snd x) == fst y]
/// [(fst x, snd y) | x <- g, y <- f,
/// snd x == fst y]
/// ~~~
///
/// This class implements the list comprehension above in a
/// single shot with worst case n*m template instantiations using an
/// inner and outer loop, where n is the size
/// of Xs and m the size of Ys. Zs is a helper that models the
/// "inner loop variable" for generating the cross product of Xs and Ys.
/// The helper function propagate_stream is integrated into the loop with
/// four cases for the matching case. Rs collects the results.
/// This class implements the list comprehension above in a single shot with
/// worst case n*m template instantiations using an inner and outer loop, where
/// n is the size of `Xs` and m the size of `Ys`. `Zs` is a helper that models
/// the "inner loop variable" for generating the cross product of `Xs` and
/// `Ys`. `Rs` collects the results.
template <class Xs, class Ys, class Zs, class Rs>
struct composed_type;
// end of outer loop over Xs
// End of outer loop over Xs.
template <class Ys, class Zs, class... Rs>
struct composed_type<detail::type_list<>, Ys, Zs, detail::type_list<Rs...>> {
using type = typed_actor<Rs...>;
};
// end of inner loop Ys (Zs)
// End of inner loop Ys (Zs).
template <class X, class... Xs, class Ys, class Rs>
struct composed_type<detail::type_list<X, Xs...>, Ys, detail::type_list<>, Rs>
: composed_type<detail::type_list<Xs...>, Ys, Ys, Rs> {};
// case #1
template <class... In, class... Out, class... Xs, class Ys,
class... MapsTo, class... Zs, class... Rs>
struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>,
output_tuple<Out...>>, Xs...>,
Ys,
detail::type_list<typed_mpi<detail::type_list<Out...>,
output_tuple<MapsTo...>>, Zs...>,
detail::type_list<Rs...>>
: composed_type<detail::type_list<Xs...>, Ys, Ys,
detail::type_list<Rs..., typed_mpi<detail::type_list<In...>,
output_tuple<MapsTo...>>>> {};
// case #2
template <class... In, class... Out, class... Xs, class Ys,
class M, class... Ms, class... Zs, class... Rs>
struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>,
output_tuple<Out...>>, Xs...>,
Ys,
detail::type_list<typed_mpi<detail::type_list<Out...>,
output_stream<M, Ms...>>,
Zs...>,
detail::type_list<Rs...>>
: composed_type<detail::type_list<Xs...>, Ys, Ys,
detail::type_list<Rs...,
typed_mpi<detail::type_list<In...>,
output_stream<M, Ms...>>>> {
};
// case #3
template <class... In, class O, class... Out, class... Xs, class Ys,
class M, class... Ms, class... Zs, class... Rs>
struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>,
output_stream<O, Out...>>, Xs...>,
Ys,
detail::type_list<typed_mpi<detail::type_list<O, Out...>,
output_tuple<M, Ms...>>, Zs...>,
detail::type_list<Rs...>>
: composed_type<detail::type_list<Xs...>, Ys, Ys,
detail::type_list<Rs...,
typed_mpi<detail::type_list<In...>,
output_stream<M, Ms...>>>> {
: composed_type<detail::type_list<Xs...>, Ys, Ys, Rs> {};
// Output type matches the input type of the next actor.
template <class... In, class... Out, class... Xs, class Ys, class... MapsTo,
class... Zs, class... Rs>
struct composed_type<
detail::type_list<typed_mpi<detail::type_list<In...>, output_tuple<Out...>>,
Xs...>,
Ys,
detail::type_list<
typed_mpi<detail::type_list<Out...>, output_tuple<MapsTo...>>, Zs...>,
detail::type_list<Rs...>>
: composed_type<
detail::type_list<Xs...>, Ys, Ys,
detail::type_list<
Rs..., typed_mpi<detail::type_list<In...>, output_tuple<MapsTo...>>>> {
};
// case #4
template <class... In, class O, class... Out, class... Xs, class Ys,
class M, class... Ms, class... Zs, class... Rs>
struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>,
output_stream<O, Out...>>, Xs...>,
Ys,
detail::type_list<typed_mpi<detail::type_list<O, Out...>,
output_stream<M, Ms...>>, Zs...>,
detail::type_list<Rs...>>
: composed_type<detail::type_list<Xs...>, Ys, Ys,
detail::type_list<Rs...,
typed_mpi<detail::type_list<In...>,
output_stream<M, Ms...>>>> {
};
// default case (recurse over Zs)
template <class In, class Out, class... Xs, class Ys,
class Unrelated, class MapsTo, class... Zs, class Rs>
struct composed_type<detail::type_list<typed_mpi<In, Out>, Xs...>,
Ys,
detail::type_list<typed_mpi<Unrelated, MapsTo>, Zs...>,
Rs>
: composed_type<detail::type_list<typed_mpi<In, Out>, Xs...>,
Ys, detail::type_list<Zs...>, Rs> {};
// No match, recurse over Zs.
template <class In, class Out, class... Xs, class Ys, class Unrelated,
class MapsTo, class... Zs, class Rs>
struct composed_type<detail::type_list<typed_mpi<In, Out>, Xs...>, Ys,
detail::type_list<typed_mpi<Unrelated, MapsTo>, Zs...>, Rs>
: composed_type<detail::type_list<typed_mpi<In, Out>, Xs...>, Ys,
detail::type_list<Zs...>, Rs> {};
/// Convenience type alias.
/// @relates composed_type
template <class F, class G>
using composed_type_t =
typename composed_type<G, F, F, detail::type_list<>>::type;
using composed_type_t = typename composed_type<G, F, F,
detail::type_list<>>::type;
} // namespace caf
......@@ -79,14 +79,6 @@ struct dmi<optional<Y> (Xs...)> : dmi<Y (Xs...)> {};
template <class Y, class... Xs>
struct dmi<expected<Y> (Xs...)> : dmi<Y (Xs...)> {};
// case #5: function returning an output_stream<>
template <class Y, class... Ys, class P, class... Xs>
struct dmi<output_stream<Y, std::tuple<Ys...>, P> (Xs...)> : dmi<Y (Xs...)> {
using type =
typed_mpi<type_list<typename param_decay<Xs>::type...>,
output_tuple<stream<Y>, strip_and_convert_t<Ys>...>>;
};
// -- dmfou = deduce_mpi_function_object_unboxing
template <class T, bool isClass = std::is_class<T>::value>
......
......@@ -140,12 +140,6 @@ public:
(*this)();
}
/// Calls `(*this)()`.
template <class Out, class... Ts>
void operator()(output_stream<Out, Ts...>&) {
(*this)();
}
/// Calls `(*this)()`.
template <class Out, class... Ts>
void operator()(outbound_stream_slot<Out, Ts...>&) {
......
......@@ -769,6 +769,19 @@ CAF_HAS_ALIAS_TRAIT(mapped_type);
// -- constexpr functions for use in enable_if & friends -----------------------
template <class List1, class List2>
struct all_constructible : std::false_type {};
template <>
struct all_constructible<type_list<>, type_list<>> : std::true_type {};
template <class T, class... Ts, class U, class... Us>
struct all_constructible<type_list<T, Ts...>, type_list<U, Us...>> {
static constexpr bool value = std::is_constructible<T, U>::value
&& all_constructible<type_list<Ts...>,
type_list<Us...>>::value;
};
/// Checks whether T behaves like `std::map`.
template <class T>
struct is_map_like {
......@@ -785,6 +798,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
......
......@@ -73,10 +73,6 @@ template <class...> class typed_event_based_actor;
template <class...> class typed_response_promise;
template <class...> class variant;
// -- variadic templates with fixed arguments ----------------------------------
//
template <class, class...> class output_stream;
// clang-format on
// -- classes ------------------------------------------------------------------
......
......@@ -18,6 +18,7 @@
#pragma once
#include "caf/delegated.hpp"
#include "caf/fwd.hpp"
#include "caf/stream_sink.hpp"
#include "caf/stream_slot.hpp"
......@@ -26,7 +27,7 @@ namespace caf {
/// Returns a stream sink with the slot ID of its first inbound path.
template <class In>
class make_sink_result {
class make_sink_result : public delegated<void> {
public:
// -- member types -----------------------------------------------------------
......@@ -39,9 +40,6 @@ public:
/// Pointer to a fully typed stream manager.
using sink_ptr_type = intrusive_ptr<sink_type>;
/// The return type for `scheduled_actor::make_sink`.
using output_stream_type = stream<input_type>;
// -- constructors, destructors, and assignment operators --------------------
make_sink_result() noexcept : slot_(0) {
......@@ -61,15 +59,15 @@ public:
// -- properties -------------------------------------------------------------
inline stream_slot inbound_slot() const noexcept {
stream_slot inbound_slot() const noexcept {
return slot_;
}
inline sink_ptr_type& ptr() noexcept {
sink_ptr_type& ptr() noexcept {
return ptr_;
}
inline const sink_ptr_type& ptr() const noexcept {
const sink_ptr_type& ptr() const noexcept {
return ptr_;
}
......
......@@ -18,17 +18,22 @@
#pragma once
#include <tuple>
#include "caf/delegated.hpp"
#include "caf/detail/implicit_conversions.hpp"
#include "caf/fwd.hpp"
#include "caf/stream.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_source.hpp"
#include "caf/detail/implicit_conversions.hpp"
namespace caf {
/// Returns a stream source with the slot ID of its first outbound path.
template <class DownstreamManager, class... Ts>
struct make_source_result {
class make_source_result
: public delegated<stream<typename DownstreamManager::output_type>, Ts...> {
public:
// -- member types -----------------------------------------------------------
/// Type of a single element.
......@@ -40,8 +45,11 @@ struct make_source_result {
/// Pointer to a fully typed stream manager.
using source_ptr_type = intrusive_ptr<source_type>;
/// The return type for `scheduled_actor::make_source`.
using output_stream_type = output_stream<output_type, Ts...>;
/// The return type for `scheduled_actor::make_stage`.
using stream_type = stream<output_type>;
/// Type of user-defined handshake arguments.
using handshake_arguments = std::tuple<Ts...>;
// -- constructors, destructors, and assignment operators --------------------
......@@ -60,23 +68,17 @@ struct make_source_result {
make_source_result& operator=(make_source_result&&) = default;
make_source_result& operator=(const make_source_result&) = default;
// -- conversion operators ---------------------------------------------------
inline operator output_stream_type() const noexcept {
return {};
}
// -- properties -------------------------------------------------------------
inline stream_slot outbound_slot() const noexcept {
stream_slot outbound_slot() const noexcept {
return slot_;
}
inline source_ptr_type& ptr() noexcept {
source_ptr_type& ptr() noexcept {
return ptr_;
}
inline const source_ptr_type& ptr() const noexcept {
const source_ptr_type& ptr() const noexcept {
return ptr_;
}
......
......@@ -18,19 +18,22 @@
#pragma once
#include <tuple>
#include "caf/delegated.hpp"
#include "caf/detail/implicit_conversions.hpp"
#include "caf/fwd.hpp"
#include "caf/output_stream.hpp"
#include "caf/stream.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_stage.hpp"
#include "caf/detail/implicit_conversions.hpp"
namespace caf {
/// Returns a stream stage with the slot IDs of its first in- and outbound
/// paths.
template <class In, class DownstreamManager, class... Ts>
class make_stage_result {
class make_stage_result
: public delegated<stream<typename DownstreamManager::output_type>, Ts...> {
public:
// -- member types -----------------------------------------------------------
......@@ -47,7 +50,10 @@ public:
using stage_ptr_type = intrusive_ptr<stage_type>;
/// The return type for `scheduled_actor::make_stage`.
using output_stream_type = output_stream<output_type, Ts...>;
using stream_type = stream<output_type>;
/// Type of user-defined handshake arguments.
using handshake_arguments = std::tuple<Ts...>;
// -- constructors, destructors, and assignment operators --------------------
......@@ -67,27 +73,21 @@ public:
make_stage_result& operator=(make_stage_result&&) = default;
make_stage_result& operator=(const make_stage_result&) = default;
// -- conversion operators ---------------------------------------------------
inline operator output_stream_type() const noexcept {
return {};
}
// -- properties -------------------------------------------------------------
inline stream_slot inbound_slot() const noexcept {
stream_slot inbound_slot() const noexcept {
return inbound_slot_;
}
inline stream_slot outbound_slot() const noexcept {
stream_slot outbound_slot() const noexcept {
return outbound_slot_;
}
inline stage_ptr_type& ptr() noexcept {
stage_ptr_type& ptr() noexcept {
return ptr_;
}
inline const stage_ptr_type& ptr() const noexcept {
const stage_ptr_type& ptr() const noexcept {
return ptr_;
}
......
......@@ -84,15 +84,27 @@ private:
flag flag_;
};
/// Convenience alias that wraps `T` into `param<T>`
/// unless `T` is arithmetic or an atom constant.
/// Converts `T` to `param<T>` unless `T` is arithmetic, an atom constant, or
/// a stream handshake.
template <class T>
using param_t =
typename std::conditional<
std::is_arithmetic<T>::value || is_atom_constant<T>::value,
T,
param<T>
>::type;
struct add_param : std::conditional<std::is_arithmetic<T>::value, T, param<T>> {
// nop
};
template <atom_value V>
struct add_param<atom_constant<V>> {
using type = atom_constant<V>;
};
template <class T>
struct add_param<stream<T>> {
using type = stream<T>;
};
/// Convenience alias that wraps `T` into `param<T>` unless `T` is arithmetic,
/// a stream handshake or an atom constant.
template <class T>
using param_t = typename add_param<T>::type;
/// Unpacks `param<T>` to `T`.
template <class T>
......@@ -112,4 +124,3 @@ struct param_decay {
};
} // namespace caf
......@@ -20,13 +20,7 @@
#include <string>
#include "caf/illegal_message_element.hpp"
#include "caf/output_stream.hpp"
#include "caf/stream.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_pair.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf {
......@@ -47,15 +41,9 @@ template <class... Is>
struct replies_to {
template <class... Os>
using with = typed_mpi<detail::type_list<Is...>, output_tuple<Os...>>;
/// @private
template <class O, class... Os>
using with_stream = typed_mpi<detail::type_list<Is...>,
output_stream<O, Os...>>;
};
template <class... Is>
using reacts_to = typed_mpi<detail::type_list<Is...>, output_tuple<void>>;
} // namespace caf
......@@ -18,15 +18,17 @@
#pragma once
#include "caf/fwd.hpp"
#include "caf/none.hpp"
#include "caf/skip.hpp"
#include <type_traits>
#include "caf/delegated.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/fwd.hpp"
#include "caf/message.hpp"
#include "caf/delegated.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/none.hpp"
#include "caf/skip.hpp"
namespace caf {
......@@ -40,13 +42,15 @@ enum result_runtime_type {
template <class... Ts>
class result {
public:
result(Ts... xs) : flag(rt_value), value(make_message(std::move(xs)...)) {
// nop
}
template <class U, class... Us>
result(U x, Us... xs) : flag(rt_value) {
init(std::move(x), std::move(xs)...);
// clang-format off
template <class... Us,
class = detail::enable_if_tt<
detail::all_constructible<
detail::type_list<Ts...>,
detail::type_list<detail::decay_t<Us>...>>>>
// clang-format on
result(Us&&... xs) : flag(rt_value) {
value = make_message(Ts{std::forward<Us>(xs)}...);
}
template <class E, class = enable_if_has_make_error_t<E>>
......@@ -58,16 +62,11 @@ public:
// nop
}
template <
class T,
class = typename std::enable_if<
sizeof...(Ts) == 1
&& std::is_convertible<
T,
detail::tl_head_t<detail::type_list<Ts...>>
>::value
>::type
>
template <class T,
class = typename std::enable_if<
sizeof...(Ts) == 1
&& std::is_convertible<
T, detail::tl_head_t<detail::type_list<Ts...>>>::value>::type>
result(expected<T> x) {
if (x) {
flag = rt_value;
......
......@@ -44,7 +44,6 @@
#include "caf/make_source_result.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/no_stages.hpp"
#include "caf/output_stream.hpp"
#include "caf/response_handle.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/sec.hpp"
......@@ -420,14 +419,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 +433,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 +447,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 +459,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 +480,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>,
......@@ -507,9 +494,7 @@ public:
std::move(pull), std::move(done), std::move(fin), token);
}
/// Creates a new continuous stream source by instantiating the default
/// source implementation with `Driver. `The returned manager is not
/// connected to any slot and thus not stored by the actor automatically.
/// @deprecated Please use `attach_continuous_stream_source` instead.
template <class Driver, class Init, class Pull, class Done,
class Finalize = unit_t>
typename Driver::source_ptr_type
......@@ -522,9 +507,7 @@ public:
return mgr;
}
/// Creates a new continuous stream source by instantiating the default
/// source implementation with `Driver. `The returned manager is not
/// connected to any slot and thus not stored by the actor automatically.
/// @deprecated Please use `attach_continuous_stream_source` instead.
template <class Init, class Pull, class Done, class Finalize = unit_t,
class DownstreamManager = broadcast_downstream_manager<
typename stream_source_trait_t<Pull>::output>>
......@@ -537,6 +520,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) {
......@@ -545,6 +529,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,
......@@ -554,6 +539,7 @@ public:
std::move(fin));
}
/// @deprecated Please use `attach_stream_stage` instead.
template <class Driver, class In, class... Ts, class... Us>
make_stage_result_t<In, typename Driver::downstream_manager_type, Ts...>
make_stage(const stream<In>& src, std::tuple<Ts...> xs, Us&&... ys) {
......@@ -564,6 +550,7 @@ public:
return {in, out, std::move(mgr)};
}
/// @deprecated Please use `attach_stream_stage` instead.
template <class In, class... Ts, class Init, class Fun,
class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Fun>,
......@@ -595,6 +582,7 @@ public:
std::move(fun), std::move(fin));
}
/// @deprecated Please use `attach_stream_stage` instead.
template <class In, class Init, class Fun, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
......@@ -605,9 +593,7 @@ public:
std::move(fin), token);
}
/// Returns a stream manager (implementing a continuous stage) without in- or
/// outbound path. The returned manager is not connected to any slot and thus
/// not stored by the actor automatically.
/// @deprecated Please use `attach_continuous_stream_stage` instead.
template <class Driver, class... Ts>
typename Driver::stage_ptr_type make_continuous_stage(Ts&&... xs) {
auto ptr = detail::make_stream_stage<Driver>(this, std::forward<Ts>(xs)...);
......@@ -615,6 +601,7 @@ public:
return ptr;
}
/// @deprecated Please use `attach_continuous_stream_stage` instead.
template <class Init, class Fun, class Cleanup,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
......
......@@ -31,7 +31,6 @@
#include "caf/mailbox_element.hpp"
#include "caf/make_message.hpp"
#include "caf/message_builder.hpp"
#include "caf/output_stream.hpp"
#include "caf/ref_counted.hpp"
#include "caf/stream.hpp"
#include "caf/stream_slot.hpp"
......
......@@ -20,8 +20,6 @@
#include <cstdint>
#include "caf/output_stream.hpp"
#include "caf/detail/comparable.hpp"
namespace caf {
......@@ -35,7 +33,7 @@ constexpr stream_slot invalid_stream_slot = 0;
/// Maps two `stream_slot` values into a pair for storing sender and receiver
/// slot information.
struct stream_slots : detail::comparable<stream_slots>{
struct stream_slots : detail::comparable<stream_slots> {
stream_slot sender;
stream_slot receiver;
......@@ -46,8 +44,7 @@ struct stream_slots : detail::comparable<stream_slots>{
}
constexpr stream_slots(stream_slot sender_slot, stream_slot receiver_slot)
: sender(sender_slot),
receiver(receiver_slot) {
: sender(sender_slot), receiver(receiver_slot) {
// nop
}
......@@ -82,7 +79,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
constexpr inbound_stream_slot(stream_slot value = 0): value_(value) {
constexpr inbound_stream_slot(stream_slot value = 0) : value_(value) {
// nop
}
......@@ -117,26 +114,28 @@ public:
/// Type of a single element.
using output_type = OutputType;
/// The return type for `scheduled_actor::make_source`.
using output_stream_type = output_stream<output_type, HandshakeArgs...>;
/// Type of a stream over the elements.
using stream_type = stream<output_type>;
/// Type of user-defined handshake arguments.
using handshake_arguments = std::tuple<HandshakeArgs...>;
// -- constructors, destructors, and assignment operators --------------------
constexpr outbound_stream_slot(stream_slot value = 0): value_(value) {
constexpr outbound_stream_slot(stream_slot value = 0) : value_(value) {
// nop
}
outbound_stream_slot(outbound_stream_slot&&) = default;
outbound_stream_slot(const outbound_stream_slot&) = default;
outbound_stream_slot& operator=(outbound_stream_slot&&) = default;
outbound_stream_slot& operator=(const outbound_stream_slot&) = default;
// -- conversion operators ---------------------------------------------------
constexpr operator output_stream_type() const noexcept {
return {};
}
constexpr operator stream_slot() const noexcept {
return value_;
}
......
......@@ -62,7 +62,11 @@ public:
}
/// @private
scheduled_actor* internal_ptr() const {
scheduled_actor* internal_ptr() const noexcept {
return view_.internal_ptr();
}
operator scheduled_actor*() const noexcept {
return view_.internal_ptr();
}
......
......@@ -116,23 +116,27 @@ public:
/// Returns a pointer to the sender of the current message.
/// @pre `current_mailbox_element() != nullptr`
inline strong_actor_ptr& current_sender() {
strong_actor_ptr& current_sender() {
return self_->current_sender();
}
/// Returns a pointer to the currently processed mailbox element.
inline mailbox_element* current_mailbox_element() {
mailbox_element* current_mailbox_element() {
return self_->current_mailbox_element();
}
/// @private
actor_control_block* ctrl() const {
actor_control_block* ctrl() const noexcept {
CAF_ASSERT(self_ != nullptr);
return actor_control_block::from(self_);;
}
/// @private
scheduled_actor* internal_ptr() const {
scheduled_actor* internal_ptr() const noexcept {
return self_;
}
operator scheduled_actor*() const noexcept {
return self_;
}
......@@ -141,4 +145,3 @@ private:
};
} // namespace caf
This diff is collapsed.
......@@ -25,6 +25,8 @@
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/attach_continuous_stream_stage.hpp"
#include "caf/attach_stream_sink.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/stateful_actor.hpp"
......@@ -46,38 +48,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 self->make_source(
// 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) -> result<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) {
......@@ -85,32 +86,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) {
......@@ -118,20 +113,16 @@ TESTEE_STATE(stream_multiplexer) {
};
TESTEE(stream_multiplexer) {
self->state.stage = self->make_continuous_stage(
self->state.stage = attach_continuous_stream_stage(
self,
// initialize state
[](unit_t&) {
// 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");
......@@ -151,7 +142,7 @@ TESTEE(stream_multiplexer) {
using fixture = test_coordinator_fixture<>;
} // namespace <anonymous>
} // namespace
// -- unit tests ---------------------------------------------------------------
......
......@@ -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"
......@@ -25,8 +27,8 @@
#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/fused_downstream_manager.hpp"
#include "caf/stateful_actor.hpp"
using std::string;
......@@ -57,9 +59,10 @@ 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> {
[=](string& fname) -> result<stream<int32_t>> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->make_source(
return attach_stream_source(
self,
// initialize state
[=](buf& xs) {
xs.resize(buf_size);
......@@ -70,19 +73,18 @@ VARARGS_TESTEE(int_file_reader, size_t buf_size) {
push(xs, out, num);
},
// check whether we reached the end
[=](const buf& xs) {
return xs.empty();
});
}
[=](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> {
[=](string& fname) -> result<stream<string>> {
CAF_CHECK_EQUAL(fname, "strings.txt");
return self->make_source(
return attach_stream_source(
self,
// initialize state
[=](buf& xs) {
for (size_t i = 0; i < buf_size; ++i)
......@@ -93,10 +95,8 @@ VARARGS_TESTEE(string_file_reader, size_t buf_size) {
push(xs, out, num);
},
// check whether we reached the end
[=](const buf& xs) {
return xs.empty();
});
}
[=](const buf& xs) { return xs.empty(); });
},
};
}
......@@ -108,27 +108,23 @@ 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
[=](intptr& x) {
x = &self->state.x;
},
[=](intptr& x) { x = &self->state.x; },
// processing step
[](intptr& x, int32_t y) {
*x += y;
},
[](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);
}
},
};
}
......@@ -139,7 +135,8 @@ TESTEE_STATE(collect) {
TESTEE(collect) {
return {
[=](stream<string>& in) {
return self->make_sink(
return attach_stream_sink(
self,
// input stream
in,
// initialize state
......@@ -153,13 +150,12 @@ TESTEE(collect) {
// 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);
}
},
};
}
......@@ -167,8 +163,8 @@ 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:
......@@ -248,7 +244,7 @@ TESTEE(stream_multiplexer) {
[=](const stream<string>& in) {
CAF_MESSAGE("received handshake for strings");
return self->state.stage->add_unchecked_inbound_path(in);
}
},
};
}
......@@ -260,11 +256,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);
......
......@@ -253,34 +253,6 @@ CAF_TEST(typed_behavior_assignment) {
h5, h6, h7, h8));
}
CAF_TEST(composed_types) {
// message type for test message #1
auto msg_1 = tk<type_list<int>>();
// message type for test message #1
auto msg_2 = tk<type_list<double>>();
// interface type a
auto if_a = tk<type_list<replies_to<int>::with<double>,
replies_to<double, double>::with<int, int>>>();
// interface type b
auto if_b = tk<type_list<replies_to<double>::with<std::string>>>();
// interface type c
auto if_c = tk<type_list<replies_to<int>::with_stream<double>>>();
// interface type b . a
auto if_ba = tk<typed_actor<replies_to<int>::with<std::string>>>();
// interface type b . c
auto if_bc = tk<typed_actor<replies_to<int>::with_stream<std::string>>>();
CAF_MESSAGE("check whether actors return the correct types");
auto nil = tk<none_t>();
auto dbl = tk<type_list<double>>();
//auto dbl_stream = tk<output_stream<double>>();
CAF_CHECK_EQUAL(res(if_a, msg_1), dbl);
CAF_CHECK_EQUAL(res(if_a, msg_2), nil);
//CAF_CHECK_EQUAL(res(if_c, msg_1), dbl_stream);
CAF_MESSAGE("check types of actor compositions");
CAF_CHECK_EQUAL(dot_op(if_b, if_a), if_ba);
CAF_CHECK_EQUAL(dot_op(if_b, if_c), if_bc);
}
struct foo {};
struct bar {};
bool operator==(const bar&, const bar&);
......
......@@ -25,6 +25,8 @@
#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"
......@@ -64,7 +66,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 +77,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) -> result<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) -> result<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 +111,16 @@ 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 attach_stream_sink(
self,
// 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 +130,67 @@ 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 attach_stream_sink(
self,
// 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 attach_stream_stage(
self,
// 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 attach_stream_stage(
self,
// 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 +205,7 @@ struct fixture : test_coordinator_fixture<> {
}
};
} // namespace <anonymous>
} // namespace
// -- unit tests ---------------------------------------------------------------
......@@ -426,8 +386,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,9 @@
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/atom.hpp"
#include "caf/attach_continuous_stream_stage.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"
#include "caf/stateful_actor.hpp"
......@@ -36,13 +39,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 +52,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,47 +73,43 @@ 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) -> result<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) {
self->state.stage = self->make_continuous_stage(
self->state.stage = attach_continuous_stream_stage(
self,
// initialize state
[](unit_t&) {
// nop
......@@ -126,23 +119,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 +138,22 @@ 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 attach_stream_sink(
self,
// 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 +164,7 @@ struct config : actor_system_config {
using fixture = test_coordinator_fixture<config>;
} // namespace <anonymous>
} // namespace
// -- unit tests ---------------------------------------------------------------
......
......@@ -18,13 +18,14 @@
#pragma once
#include <type_traits>
#include "caf/all.hpp"
#include "caf/config.hpp"
#include "caf/detail/gcd.hpp"
#include "caf/meta/annotation.hpp"
#include "caf/test/unit_test.hpp"
#include "caf/detail/gcd.hpp"
CAF_PUSH_WARNINGS
/// The type of `_`.
......@@ -375,6 +376,71 @@ protected:
std::function<void ()> peek_;
};
template <class... Ts>
class inject_clause {
public:
inject_clause(caf::scheduler::test_coordinator& sched)
: sched_(sched), dest_(nullptr) {
// nop
}
inject_clause(inject_clause&& other) = default;
template <class Handle>
inject_clause& from(const Handle& whom) {
src_ = caf::actor_cast<caf::strong_actor_ptr>(whom);
return *this;
}
template <class Handle>
inject_clause& to(const Handle& whom) {
dest_ = caf::actor_cast<caf::strong_actor_ptr>(whom);
return *this;
}
inject_clause& to(const caf::scoped_actor& whom) {
dest_ = whom.ptr();
return *this;
}
void with(Ts... xs) {
if (src_ == nullptr)
CAF_FAIL("missing .from() in inject() statement");
if (dest_ == nullptr)
CAF_FAIL("missing .to() in inject() statement");
caf::send_as(caf::actor_cast<caf::actor>(src_),
caf::actor_cast<caf::actor>(dest_), xs...);
CAF_REQUIRE(sched_.prioritize(dest_));
auto dest_ptr = &sched_.next_job<caf::abstract_actor>();
auto ptr = dest_ptr->peek_at_next_mailbox_element();
CAF_REQUIRE(ptr != nullptr);
CAF_REQUIRE_EQUAL(ptr->sender, src_);
// TODO: replace this workaround with the make_tuple() line when dropping
// support for GCC 4.8.
std::tuple<Ts...> tmp{std::move(xs)...};
using namespace caf::detail;
elementwise_compare_inspector<decltype(tmp)> inspector{tmp};
auto ys = extract<Ts...>(dest_);
auto ys_indices = get_indices(ys);
CAF_REQUIRE(apply_args(inspector, ys_indices, ys));
run_once();
}
protected:
void run_once() {
auto ptr = caf::actor_cast<caf::abstract_actor*>(dest_);
auto dptr = dynamic_cast<caf::blocking_actor*>(ptr);
if (dptr == nullptr)
sched_.run_once();
else
dptr->dequeue(); // Drop message.
}
caf::scheduler::test_coordinator& sched_;
caf::strong_actor_ptr src_;
caf::strong_actor_ptr dest_;
};
template <class... Ts>
class allow_clause {
public:
......@@ -797,6 +863,12 @@ T unbox(T* x) {
expect_clause<CAF_EXPAND(CAF_DSL_LIST types)>{sched}.fields; \
} while (false)
#define inject(types, fields) \
do { \
CAF_MESSAGE("inject" << #types << "." << #fields); \
inject_clause<CAF_EXPAND(CAF_DSL_LIST types)>{sched}.fields; \
} while (false)
/// Convenience macro for defining allow clauses.
#define allow(types, fields) \
([&] { \
......
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