Commit 6b66ed28 authored by Dominik Charousset's avatar Dominik Charousset

Add scatterer and result to streaming types

Make the scatterer as well as the result type part of `stream_source`,
`stream_stage` and `stream_sink`. This allows more type-checking at
compile time and (most importantly) gives users access to the scatterer
when using continuous stages.
parent b4e0d75b
...@@ -61,7 +61,8 @@ public: ...@@ -61,7 +61,8 @@ public:
virtual void operator()(message&) = 0; virtual void operator()(message&) = 0;
/// Called if the message handler returned a `stream<...>`, /// Called if the message handler returned a `stream<...>`,
/// `output_stream<...>`, or `stream_result<...>`. /// `output_stream<...>`, or `stream_result<...>`, `make_source_result<...>`,
/// `make_sink_result<...>`, or `make_stage_result<...>`.
virtual void operator()(stream_slot in, stream_slot out, virtual void operator()(stream_slot in, stream_slot out,
stream_manager_ptr& mgr) = 0; stream_manager_ptr& mgr) = 0;
...@@ -146,11 +147,32 @@ public: ...@@ -146,11 +147,32 @@ public:
(*this)(x.slot(), x.ptr()); (*this)(x.slot(), x.ptr());
} }
/// Calls `(*this)(x.ptr)`. /// Calls `(*this)(x.in(), x.out(), x.ptr())`.
template <class T, class... Us> template <class Out, class... Ts>
void operator()(output_stream<T, Us...>& x) { void operator()(output_stream<Out, Ts...>& x) {
stream_manager_ptr ptr{std::move(x.ptr())};
(*this)(x.in(), x.out(), ptr);
}
/// Calls `(*this)(x.in(), x.out(), x.ptr())`.
template <class In, class Result>
void operator()(make_sink_result<In, Result>& x) {
stream_manager_ptr ptr{std::move(x.ptr())};
(*this)(x.in(), 0, ptr);
}
/// Calls `(*this)(x.in(), x.out(), x.ptr())`.
template <class T, class S, class... Ts>
void operator()(make_source_result<T, S, Ts...>& x) {
stream_manager_ptr ptr{std::move(x.ptr())};
(*this)(0, x.out(), ptr);
}
/// Calls `(*this)(x.in(), x.out(), x.ptr())`.
template <class In, class Result, class Out, class S, class... Ts>
void operator()(make_stage_result<In, Result, Out, S, Ts...>& x) {
stream_manager_ptr ptr{std::move(x.ptr())}; stream_manager_ptr ptr{std::move(x.ptr())};
(*this)(x.origin_slot(), x.slot(), ptr); (*this)(x.in(), x.out(), ptr);
} }
/// Calls `(*this)(x.ptr)`. /// Calls `(*this)(x.ptr)`.
......
...@@ -27,18 +27,19 @@ ...@@ -27,18 +27,19 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
template <class Output, class Pull, class Done, class Finalize, template <class Scatterer, class Pull, class Done, class Finalize,
class HandshakeData> class HandshakeData>
class stream_source_driver_impl; class stream_source_driver_impl;
/// Identifies an unbound sequence of messages. /// Identifies an unbound sequence of messages.
template <class Output, class Pull, class Done, class Finalize, class... Ts> template <class Scatterer, class Pull, class Done, class Finalize, class... Ts>
class stream_source_driver_impl<Output, Pull, Done, Finalize, std::tuple<Ts...>> class stream_source_driver_impl<Scatterer, Pull, Done, Finalize,
final : public stream_source_driver<Output, Ts...> { std::tuple<Ts...>>
final : public stream_source_driver<Scatterer, Ts...> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = stream_source_driver<Output, Ts...>; using super = stream_source_driver<Scatterer, Ts...>;
using output_type = typename super::output_type; using output_type = typename super::output_type;
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/make_source_result.hpp"
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
#include "caf/stream_source.hpp" #include "caf/stream_source.hpp"
#include "caf/stream_source_trait.hpp" #include "caf/stream_source_trait.hpp"
...@@ -30,59 +31,45 @@ ...@@ -30,59 +31,45 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
template <class Driver, class Scatterer> template <class Driver>
class stream_source_impl : public Driver::source_type { class stream_source_impl : public Driver::source_type {
public: public:
// -- static asserts ---------------------------------------------------------
static_assert(std::is_same<typename Driver::output_type,
typename Scatterer::value_type>::value,
"Driver and Scatterer have incompatible types.");
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = typename Driver::source_type; using super = typename Driver::source_type;
using driver_type = Driver; using driver_type = Driver;
using scatterer_type = Scatterer;
using output_type = typename driver_type::output_type;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
template <class... Ts> template <class... Ts>
stream_source_impl(local_actor* self, Ts&&... xs) stream_source_impl(local_actor* self, Ts&&... xs)
: super(self), : stream_manager(self),
super(self),
at_end_(false), at_end_(false),
driver_(std::forward<Ts>(xs)...), driver_(std::forward<Ts>(xs)...) {
out_(self) {
// nop // nop
} }
// -- implementation of virtual functions ------------------------------------ // -- implementation of virtual functions ------------------------------------
bool done() const override { bool done() const override {
return this->pending_handshakes_ == 0 && at_end_ && out_.clean(); return this->pending_handshakes_ == 0 && at_end_ && this->out_.clean();
}
Scatterer& out() override {
return out_;
} }
bool generate_messages() override { bool generate_messages() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (at_end_) if (at_end_)
return false; return false;
auto hint = out_.capacity(); auto hint = this->out_.capacity();
CAF_LOG_DEBUG(CAF_ARG(hint)); CAF_LOG_DEBUG(CAF_ARG(hint));
if (hint == 0) if (hint == 0)
return false; return false;
downstream<output_type> ds{out_.buf()}; downstream<typename Driver::output_type> ds{this->out_.buf()};
driver_.pull(ds, hint); driver_.pull(ds, hint);
if (driver_.done()) if (driver_.done())
at_end_ = true; at_end_ = true;
return hint != out_.capacity(); return hint != this->out_.capacity();
} }
message make_handshake(stream_slot slot) const override { message make_handshake(stream_slot slot) const override {
...@@ -97,15 +84,12 @@ protected: ...@@ -97,15 +84,12 @@ protected:
private: private:
bool at_end_; bool at_end_;
Driver driver_; Driver driver_;
Scatterer out_;
}; };
template <class Driver, template <class Driver, class... Ts>
class Scatterer = broadcast_scatterer<typename Driver::output_type>,
class... Ts>
typename Driver::source_ptr_type make_stream_source(local_actor* self, typename Driver::source_ptr_type make_stream_source(local_actor* self,
Ts&&... xs) { Ts&&... xs) {
using impl = stream_source_impl<Driver, Scatterer>; using impl = stream_source_impl<Driver>;
return make_counted<impl>(self, std::forward<Ts>(xs)...); return make_counted<impl>(self, std::forward<Ts>(xs)...);
} }
......
...@@ -20,38 +20,37 @@ ...@@ -20,38 +20,37 @@
#ifndef CAF_STREAM_STAGE_DRIVER_IMPL_HPP #ifndef CAF_STREAM_STAGE_DRIVER_IMPL_HPP
#define CAF_STREAM_STAGE_DRIVER_IMPL_HPP #define CAF_STREAM_STAGE_DRIVER_IMPL_HPP
#include "caf/none.hpp" #include "caf/stream_slot.hpp"
#include "caf/stream_stage_driver.hpp" #include "caf/stream_stage_driver.hpp"
#include "caf/stream_stage_trait.hpp" #include "caf/stream_stage_trait.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
template <class Input, class Output, class Process, class Finalize, template <class Input, class Scatterer, class Process, class Finalize,
class HandshakeData> class HandshakeData>
class stream_stage_driver_impl; class stream_stage_driver_impl;
/// Identifies an unbound sequence of messages. /// Default implementation for a `stream_stage_driver` that hardwires `message`
template <class Input, class Output, class Process, class Finalize, class... Ts> /// as result type and implements `process` and `finalize` using user-provided
class stream_stage_driver_impl<Input, Output, Process, Finalize, /// function objects (usually lambdas).
std::tuple<Ts...>> final template <class Input, class Scatterer, class Process, class Finalize,
: public stream_stage_driver<Input, Output, Ts...> { class... Ts>
class stream_stage_driver_impl<Input, Scatterer, Process, Finalize,
std::tuple<Ts...>>
final : public stream_stage_driver<Input, message, Scatterer, Ts...> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = stream_stage_driver<Input, Output, Ts...>; using super = stream_stage_driver<Input, message, Scatterer, Ts...>;
using input_type = typename super::output_type; using typename super::input_type;
using output_type = typename super::output_type; using typename super::output_type;
using stream_type = stream<output_type>; using typename super::stream_type;
using output_stream_type = typename super::output_stream_type; using typename super::handshake_tuple_type;
using tuple_type = std::tuple<Ts...>;
using handshake_tuple_type = typename super::handshake_tuple_type;
using trait = stream_stage_trait_t<Process>; using trait = stream_stage_trait_t<Process>;
...@@ -74,6 +73,15 @@ public: ...@@ -74,6 +73,15 @@ public:
trait::process::invoke(process_, state_, std::move(batch), out); trait::process::invoke(process_, state_, std::move(batch), out);
} }
void add_result(message& x) override {
// The default driver assumes to receive only a single result.
result_ = std::move(x);
}
message make_final_result() override {
return std::move(result_);
}
void finalize(const error&) override { void finalize(const error&) override {
return fin_(state_); return fin_(state_);
} }
...@@ -82,7 +90,8 @@ private: ...@@ -82,7 +90,8 @@ private:
state_type state_; state_type state_;
Process process_; Process process_;
Finalize fin_; Finalize fin_;
tuple_type hs_; std::tuple<Ts...> hs_;
message result_;
}; };
} // namespace detail } // namespace detail
......
...@@ -31,22 +31,16 @@ ...@@ -31,22 +31,16 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
template <class Driver, class Scatterer> template <class Driver>
class stream_stage_impl : public Driver::stage_type { class stream_stage_impl : public Driver::stage_type {
public: public:
// -- static asserts ---------------------------------------------------------
static_assert(std::is_same<typename Driver::output_type,
typename Scatterer::value_type>::value,
"Driver and Scatterer have incompatible types.");
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = typename Driver::stage_type; using super = typename Driver::stage_type;
using driver_type = Driver; using driver_type = Driver;
using scatterer_type = Scatterer; using scatterer_type = typename Driver::scatterer_type;
using input_type = typename driver_type::input_type; using input_type = typename driver_type::input_type;
...@@ -56,21 +50,17 @@ public: ...@@ -56,21 +50,17 @@ public:
template <class... Ts> template <class... Ts>
stream_stage_impl(local_actor* self, Ts&&... xs) stream_stage_impl(local_actor* self, Ts&&... xs)
: super(self), : stream_manager(self),
driver_(std::forward<Ts>(xs)...), super(self),
out_(self) { driver_(std::forward<Ts>(xs)...) {
// nop // nop
} }
// -- implementation of virtual functions ------------------------------------ // -- implementation of virtual functions ------------------------------------
scatterer_type& out() override {
return out_;
}
bool done() const override { bool done() const override {
return !this->continuous() && this->pending_handshakes_ == 0 return !this->continuous() && this->pending_handshakes_ == 0
&& this->inbound_paths_.empty() && out_.clean(); && this->inbound_paths_.empty() && this->out_.clean();
} }
void handle(inbound_path*, downstream_msg::batch& x) override { void handle(inbound_path*, downstream_msg::batch& x) override {
...@@ -78,7 +68,7 @@ public: ...@@ -78,7 +68,7 @@ public:
using vec_type = std::vector<input_type>; using vec_type = std::vector<input_type>;
if (x.xs.match_elements<vec_type>()) { if (x.xs.match_elements<vec_type>()) {
auto& xs = x.xs.get_mutable_as<vec_type>(0); auto& xs = x.xs.get_mutable_as<vec_type>(0);
downstream<output_type> ds{out_.buf()}; downstream<output_type> ds{this->out_.buf()};
driver_.process(std::move(xs), ds); driver_.process(std::move(xs), ds);
return; return;
} }
...@@ -90,7 +80,7 @@ public: ...@@ -90,7 +80,7 @@ public:
} }
bool congested() const noexcept override { bool congested() const noexcept override {
return out_.capacity() == 0; return this->out_.capacity() == 0;
} }
protected: protected:
...@@ -100,15 +90,12 @@ protected: ...@@ -100,15 +90,12 @@ protected:
private: private:
driver_type driver_; driver_type driver_;
scatterer_type out_;
}; };
template <class Driver, template <class Driver, class... Ts>
class Scatterer = broadcast_scatterer<typename Driver::output_type>,
class... Ts>
typename Driver::stage_ptr_type make_stream_stage(local_actor* self, typename Driver::stage_ptr_type make_stream_stage(local_actor* self,
Ts&&... xs) { Ts&&... xs) {
using impl = stream_stage_impl<Driver, Scatterer>; using impl = stream_stage_impl<Driver>;
return make_counted<impl>(self, std::forward<Ts>(xs)...); return make_counted<impl>(self, std::forward<Ts>(xs)...);
} }
......
...@@ -26,19 +26,24 @@ namespace caf { ...@@ -26,19 +26,24 @@ namespace caf {
// -- 1 param templates -------------------------------------------------------- // -- 1 param templates --------------------------------------------------------
template <class> class param; template <class> class behavior_type_of;
template <class> class stream; template <class> class broadcast_scatterer;
template <class> class optional;
template <class> class expected;
template <class> class downstream; template <class> class downstream;
template <class> class expected;
template <class> class intrusive_ptr; template <class> class intrusive_ptr;
template <class> class behavior_type_of; template <class> class optional;
template <class> class param;
template <class> class stream;
template <class> class stream_result;
template <class> class trivial_match_case; template <class> class trivial_match_case;
template <class> class weak_intrusive_ptr; template <class> class weak_intrusive_ptr;
template <class> class broadcast_scatterer;
template <class> struct timeout_definition; template <class> struct timeout_definition;
// -- 2 param templates --------------------------------------------------------
template <class, class> class make_sink_result;
// -- 3 param templates -------------------------------------------------------- // -- 3 param templates --------------------------------------------------------
template <class, class, int> class actor_cast_access; template <class, class, int> class actor_cast_access;
...@@ -55,14 +60,13 @@ template <class...> class typed_actor_pointer; ...@@ -55,14 +60,13 @@ template <class...> class typed_actor_pointer;
template <class...> class typed_response_promise; template <class...> class typed_response_promise;
template <class...> class typed_event_based_actor; template <class...> class typed_event_based_actor;
// -- variadic templates with 1 fixed argument --------------------------------- // -- variadic templates with fixed arguments ----------------------------------
template <class, class, class, class, class...> class make_stage_result;
template <class, class, class, class, class...> class stream_stage;
template <class, class, class...> class make_source_result;
template <class, class, class...> class stream_source;
template <class, class...> class output_stream; template <class, class...> class output_stream;
template <class, class...> class stream_source;
// -- variadic templates with 2 fixed argument ---------------------------------
template <class, class, class...> class stream_stage;
// -- classes ------------------------------------------------------------------ // -- classes ------------------------------------------------------------------
...@@ -135,6 +139,7 @@ enum class atom_value : uint64_t; ...@@ -135,6 +139,7 @@ enum class atom_value : uint64_t;
// -- aliases ------------------------------------------------------------------ // -- aliases ------------------------------------------------------------------
using actor_id = uint64_t; using actor_id = uint64_t;
using stream_slot = uint16_t;
// -- intrusive containers ----------------------------------------------------- // -- intrusive containers -----------------------------------------------------
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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://opensink.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_MAKE_SINK_RESULT_HPP
#define CAF_MAKE_SINK_RESULT_HPP
#include "caf/fwd.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_sink.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/meta/type_name.hpp"
namespace caf {
/// Wraps the result of a `make_sink` or `add_output_path` function call.
template <class In, class Result>
class make_sink_result {
public:
// -- member types -----------------------------------------------------------
/// A single element.
using value_type = In;
/// Fully typed stream manager as returned by `make_sink`.
using type = stream_sink<In, Result>;
/// Pointer to a fully typed stream manager.
using ptr_type = intrusive_ptr<type>;
/// Type of the final result message.
using result_type = stream_result<Result>;
// -- constructors, destructors, and assignment operators --------------------
make_sink_result(make_sink_result&&) = default;
make_sink_result(const make_sink_result&) = default;
make_sink_result(stream_slot in_slot, ptr_type ptr)
: in_(in_slot),
ptr_(std::move(ptr)) {
// nop
}
// -- properties -------------------------------------------------------------
/// Returns the output slot.
inline stream_slot in() const {
return in_;
}
/// Returns the handler assigned to this stream on this actor.
inline ptr_type& ptr() noexcept {
return ptr_;
}
/// Returns the handler assigned to this stream on this actor.
inline const ptr_type& ptr() const noexcept {
return ptr_;
}
// -- serialization support --------------------------------------------------
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
make_sink_result& x) {
return f(meta::type_name("make_sink_result"), x.in_);
}
private:
// -- member variables -------------------------------------------------------
stream_slot in_;
ptr_type ptr_;
};
} // namespace caf
#endif // CAF_MAKE_SINK_RESULT_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#ifndef CAF_MAKE_SOURCE_RESULT_HPP
#define CAF_MAKE_SOURCE_RESULT_HPP
#include "caf/fwd.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_source.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/meta/type_name.hpp"
namespace caf {
/// Wraps the result of a `make_source` or `add_output_path` function call.
template <class T, class Scatterer, class... Ts>
class make_source_result {
public:
// -- member types -----------------------------------------------------------
/// Type of a single element.
using value_type = T;
/// Fully typed stream manager as returned by `make_source`.
using type = stream_source<T, Scatterer, Ts...>;
/// Pointer to a fully typed stream manager.
using ptr_type = intrusive_ptr<type>;
// -- constructors, destructors, and assignment operators --------------------
make_source_result(make_source_result&&) = default;
make_source_result(const make_source_result&) = default;
make_source_result(stream_slot out_slot, ptr_type ptr)
: out_(out_slot),
ptr_(std::move(ptr)) {
// nop
}
// -- properties -------------------------------------------------------------
/// Returns the output slot.
inline stream_slot out() const {
return out_;
}
/// Returns the handler assigned to this stream on this actor.
inline ptr_type& ptr() noexcept {
return ptr_;
}
/// Returns the handler assigned to this stream on this actor.
inline const ptr_type& ptr() const noexcept {
return ptr_;
}
// -- serialization support --------------------------------------------------
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
make_source_result& x) {
return f(meta::type_name("make_source_result"), x.out_);
}
private:
// -- member variables -------------------------------------------------------
stream_slot out_;
ptr_type ptr_;
};
/// Helper type for defining a `make_source_result` from a `Scatterer` plus
/// additional handshake types.
template <class Scatterer, class... Ts>
using make_source_result_t =
make_source_result<typename Scatterer::value_type, Scatterer,
detail::decay_t<Ts>...>;
} // namespace caf
#endif // CAF_MAKE_SOURCE_RESULT_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#ifndef CAF_MAKE_STAGE_RESULT_HPP
#define CAF_MAKE_STAGE_RESULT_HPP
#include "caf/fwd.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_stage.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/meta/type_name.hpp"
namespace caf {
/// Identifies an unbound sequence of elements annotated with the additional
/// handshake arguments emitted to the next stage.
template <class In, class Result, class Out, class Scatterer, class... Ts>
class make_stage_result {
public:
// -- member types -----------------------------------------------------------
/// Type of a single element.
using value_type = Out;
/// Fully typed stream manager as returned by `make_stage`.
using ptr_type = stream_stage_ptr<In, Result, Out, Scatterer, Ts...>;
// -- constructors, destructors, and assignment operators --------------------
make_stage_result(make_stage_result&&) = default;
make_stage_result(const make_stage_result&) = default;
make_stage_result(stream_slot in_slot, stream_slot out_slot, ptr_type ptr)
: in_(in_slot),
out_(out_slot),
ptr_(std::move(ptr)) {
// nop
}
// -- properties -------------------------------------------------------------
/// Returns the slot of the origin stream if `ptr()` is a stage or 0 if
/// `ptr()` is a source.
inline stream_slot in() const {
return in_;
}
/// Returns the output slot.
inline stream_slot out() const {
return out_;
}
/// Returns the handler assigned to this stream on this actor.
inline ptr_type& ptr() noexcept {
return ptr_;
}
/// Returns the handler assigned to this stream on this actor.
inline const ptr_type& ptr() const noexcept {
return ptr_;
}
// -- serialization support --------------------------------------------------
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
make_stage_result& x) {
return f(meta::type_name("make_stage_result"), x.in_, x.out_);
}
private:
// -- member variables -------------------------------------------------------
stream_slot in_;
stream_slot out_;
ptr_type ptr_;
};
/// Helper type for defining a `make_stage_result` from a `Scatterer` plus
/// additional handshake types. Hardwires `message` as result type.
template <class In, class Scatterer, class... Ts>
using make_stage_result_t =
make_stage_result<In, message, typename Scatterer::value_type, Scatterer,
detail::decay_t<Ts>...>;
} // namespace caf
#endif // CAF_MAKE_STAGE_RESULT_HPP
...@@ -20,9 +20,9 @@ ...@@ -20,9 +20,9 @@
#define CAF_OUTPUT_STREAM_HPP #define CAF_OUTPUT_STREAM_HPP
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/make_source_result.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/stream_source.hpp"
#include "caf/stream_stage.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
...@@ -35,32 +35,28 @@ class output_stream { ...@@ -35,32 +35,28 @@ class output_stream {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
/// Dennotes the supertype.
using super = stream<T>;
/// Type of a single element. /// Type of a single element.
using value_type = T; using value_type = T;
/// Smart pointer to a source.
using source_pointer = stream_source_ptr<T, Ts...>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
output_stream(output_stream&&) = default; output_stream(output_stream&&) = default;
output_stream(const output_stream&) = default; output_stream(const output_stream&) = default;
output_stream(stream_slot id, source_pointer sptr) template <class S>
: origin_(0), output_stream(make_source_result<T, S, Ts...> x)
slot_(id), : in_(0),
ptr_(std::move(sptr)) { out_(x.out()),
ptr_(std::move(x.ptr())) {
// nop // nop
} }
template <class In> template <class I, class R, class S>
output_stream(stream_slot origin, stream_slot id, output_stream(make_stage_result<I, R, T, S, Ts...> x)
stream_stage_ptr<In, T, Ts...> sptr) : in_(x.in()),
: origin_(origin), slot_(id), ptr_(std::move(sptr)) { out_(x.out()),
ptr_(std::move(x.ptr())) {
// nop // nop
} }
...@@ -68,22 +64,22 @@ public: ...@@ -68,22 +64,22 @@ public:
/// Returns the slot of the origin stream if `ptr()` is a stage or 0 if /// Returns the slot of the origin stream if `ptr()` is a stage or 0 if
/// `ptr()` is a source. /// `ptr()` is a source.
inline stream_slot origin_slot() const { inline stream_slot in() const {
return origin_; return in_;
} }
/// Returns the output slot. /// Returns the output slot.
inline stream_slot slot() const { inline stream_slot out() const {
return slot_; return out_;
} }
/// Returns the handler assigned to this stream on this actor. /// Returns the handler assigned to this stream on this actor.
inline source_pointer& ptr() noexcept { inline stream_manager_ptr& ptr() noexcept {
return ptr_; return ptr_;
} }
/// Returns the handler assigned to this stream on this actor. /// Returns the handler assigned to this stream on this actor.
inline const source_pointer& ptr() const noexcept { inline const stream_manager_ptr& ptr() const noexcept {
return ptr_; return ptr_;
} }
...@@ -92,15 +88,15 @@ public: ...@@ -92,15 +88,15 @@ public:
template <class Inspector> template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, friend typename Inspector::result_type inspect(Inspector& f,
output_stream& x) { output_stream& x) {
return f(meta::type_name("output_stream"), x.origin_, x.slot_); return f(meta::type_name("output_stream"), x.in_, x.out_);
} }
private: private:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
stream_slot origin_; stream_slot in_;
stream_slot slot_; stream_slot out_;
source_pointer ptr_; stream_manager_ptr ptr_;
}; };
} // namespace caf } // namespace caf
......
...@@ -39,6 +39,9 @@ ...@@ -39,6 +39,9 @@
#include "caf/invoke_message_result.hpp" #include "caf/invoke_message_result.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/make_sink_result.hpp"
#include "caf/make_source_result.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/no_stages.hpp" #include "caf/no_stages.hpp"
#include "caf/output_stream.hpp" #include "caf/output_stream.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
...@@ -413,27 +416,28 @@ public: ...@@ -413,27 +416,28 @@ public:
// -- stream management ------------------------------------------------------ // -- stream management ------------------------------------------------------
/// Creates an output path for given source. /// Creates an output path for given source.
template <class Out, class... Ts> template <class Out, class Scatterer, class... Ts>
output_stream<Out, Ts...> make_source_result<Out, Scatterer, Ts...>
add_output_path(stream_source_ptr<Out, Ts...> mgr) { add_output_path(stream_source_ptr<Out, Scatterer, Ts...> mgr) {
auto slot = assign_next_pending_slot(mgr); auto slot = assign_next_pending_slot(mgr);
return {slot, std::move(mgr)}; return {slot, std::move(mgr)};
} }
/// Creates an output path for given stage. /// Creates an output path for given stage.
template <class In, class Out, class... Ts> template <class In, class Result, class Out, class Scatterer, class... Ts>
output_stream<Out, Ts...> make_source_result<Out, Scatterer, Ts...>
add_output_path(stream_stage_ptr<In, Out, Ts...> mgr) { add_output_path(stream_stage_ptr<In, Result, Out, Scatterer, Ts...> mgr) {
auto slot = assign_next_pending_slot(mgr); auto slot = assign_next_pending_slot(mgr);
return {slot, std::move(mgr)}; return {slot, std::move(mgr)};
} }
/// Creates an input path for given stage. /// Creates an input path for given stage.
template <class In, class Out, class... Ts> template <class In, class Result, class Out, class Scatterer, class... Ts>
output_stream<Out, Ts...> make_sink_result<In, Result>
add_input_path(const stream<In>&, stream_stage_ptr<In, Out, Ts...> mgr) { add_input_path(const stream<In>&,
stream_stage_ptr<In, Result, Out, Scatterer, Ts...> mgr) {
auto slot = assign_next_slot(mgr); auto slot = assign_next_slot(mgr);
return {slot, 0, std::move(mgr)}; return {slot, std::move(mgr)};
} }
/// Creates a new stream source and starts streaming to `dest`. /// Creates a new stream source and starts streaming to `dest`.
...@@ -445,37 +449,33 @@ public: ...@@ -445,37 +449,33 @@ public:
/// @param res_handler Function object for receiving the stream result. /// @param res_handler Function object for receiving the stream result.
/// @param scatterer_type Configures the policy for downstream communication. /// @param scatterer_type Configures the policy for downstream communication.
/// @returns A stream object with a pointer to the generated `stream_manager`. /// @returns A stream object with a pointer to the generated `stream_manager`.
template <class Driver, template <class Driver, class... Ts>
class Scatterer = broadcast_scatterer<typename Driver::output_type>, typename Driver::make_source_result_type make_source(Ts&&... xs) {
class... Ts> using detail::make_stream_source;
typename Driver::output_stream_type make_source(Ts&&... xs) { auto mgr = make_stream_source<Driver>(this, std::forward<Ts>(xs)...);
auto mgr = detail::make_stream_source<Driver, Scatterer>(
this, std::forward<Ts>(xs)...);
auto slot = assign_next_pending_slot(mgr); auto slot = assign_next_pending_slot(mgr);
return {slot, std::move(mgr)}; return {slot, std::move(mgr)};
} }
template <class... Ts, class Init, class Pull, class Done, class Finalize, template <class... Ts, class Init, class Pull, class Done, class Finalize,
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>, broadcast_scatterer<typename stream_source_trait_t<Pull>::output>>
class Trait = stream_source_trait_t<Pull>> make_source_result_t<Scatterer, Ts...>
output_stream<typename Trait::output, detail::decay_t<Ts>...>
make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done, make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done,
Finalize finalize, policy::arg<Scatterer> = {}) { Finalize finalize, policy::arg<Scatterer> = {}) {
using tuple_type = std::tuple<detail::decay_t<Ts>...>; using tuple_type = std::tuple<detail::decay_t<Ts>...>;
using driver = using driver = detail::stream_source_driver_impl<Scatterer, Pull, Done,
detail::stream_source_driver_impl<typename Trait::output, Pull, Done, Finalize, tuple_type>;
Finalize, tuple_type>; return make_source<driver>(std::move(init), std::move(pull),
return make_source<driver, Scatterer>(std::move(init), std::move(pull), std::move(done), std::move(finalize),
std::move(done), std::move(finalize), std::move(xs));
std::move(xs));
} }
template <class... Ts, class Init, class Pull, class Done, template <class Init, class Pull, class Done,
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>, broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>> class Trait = stream_source_trait_t<Pull>>
output_stream<typename Trait::output, detail::decay_t<Ts>...> make_source_result_t<Scatterer>
make_source(Init init, Pull pull, Done done, make_source(Init init, Pull pull, Done done,
policy::arg<Scatterer> scatterer_type = {}) { policy::arg<Scatterer> scatterer_type = {}) {
auto finalize = [](typename Trait::state, const error&) {}; auto finalize = [](typename Trait::state, const error&) {};
...@@ -483,11 +483,11 @@ public: ...@@ -483,11 +483,11 @@ public:
scatterer_type); scatterer_type);
} }
template <class... Ts, class Init, class Pull, class Done, class Finalize, template <class Init, class Pull, class Done, class Finalize,
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>, broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>> class Trait = stream_source_trait_t<Pull>>
output_stream<typename Trait::output, detail::decay_t<Ts>...> make_source_result_t<Scatterer>
make_source(Init init, Pull pull, Done done, Finalize finalize, make_source(Init init, Pull pull, Done done, Finalize finalize,
policy::arg<Scatterer> scatterer_type = {}) { policy::arg<Scatterer> scatterer_type = {}) {
return make_source(std::make_tuple(), init, pull, done, finalize, return make_source(std::make_tuple(), init, pull, done, finalize,
...@@ -498,7 +498,7 @@ public: ...@@ -498,7 +498,7 @@ public:
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>, broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>> class Trait = stream_source_trait_t<Pull>>
output_stream<typename Trait::output, detail::decay_t<Ts>...> make_source_result_t<Scatterer, Ts...>
make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done, make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done,
policy::arg<Scatterer> scatterer_type = {}) { policy::arg<Scatterer> scatterer_type = {}) {
auto finalize = [](typename Trait::state, const error&) {}; auto finalize = [](typename Trait::state, const error&) {};
...@@ -507,7 +507,9 @@ public: ...@@ -507,7 +507,9 @@ public:
} }
template <class Driver, class Input, class... Ts> template <class Driver, class Input, class... Ts>
stream_result<typename Driver::output_type> make_sink(const stream<Input>&, stream_result<typename Driver::output_type>
make_sink(const stream<Input>&,
Ts&&... xs) { Ts&&... xs) {
auto mgr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...); auto mgr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...);
auto slot = assign_next_slot(mgr); auto slot = assign_next_slot(mgr);
...@@ -524,24 +526,21 @@ public: ...@@ -524,24 +526,21 @@ public:
std::move(fin)); std::move(fin));
} }
template <class Driver, template <class Driver, class In, class... Ts>
class Scatterer = broadcast_scatterer<typename Driver::output_type>, typename Driver::make_stage_result_type make_stage(const stream<In>&,
class Input = int, class... Ts> Ts&&... xs) {
typename Driver::output_stream_type make_stage(const stream<Input>& in,
Ts&&... xs) {
using detail::make_stream_stage; using detail::make_stream_stage;
auto mgr = make_stream_stage<Driver, Scatterer>(this, auto mgr = make_stream_stage<Driver>(this, std::forward<Ts>(xs)...);
std::forward<Ts>(xs)...); auto in = assign_next_slot(mgr);
assign_slot(in.slot(), mgr); auto out = assign_next_pending_slot(mgr);
auto slot = assign_next_pending_slot(mgr); return {in, out, std::move(mgr)};
return {in.slot(), slot, std::move(mgr)};
} }
template <class In, class... Ts, class Init, class Fun, class Cleanup, template <class In, class... Ts, class Init, class Fun, class Cleanup,
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>, broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>,
class Trait = stream_stage_trait_t<Fun>> class Trait = stream_stage_trait_t<Fun>>
output_stream<typename Trait::output, detail::decay_t<Ts>...> make_stage_result_t<In, Scatterer, Ts...>
make_stage(const stream<In>& in, std::tuple<Ts...> xs, Init init, Fun fun, make_stage(const stream<In>& in, std::tuple<Ts...> xs, Init init, Fun fun,
Cleanup cleanup, policy::arg<Scatterer> scatterer_type = {}) { Cleanup cleanup, policy::arg<Scatterer> scatterer_type = {}) {
CAF_IGNORE_UNUSED(scatterer_type); CAF_IGNORE_UNUSED(scatterer_type);
...@@ -562,22 +561,19 @@ public: ...@@ -562,22 +561,19 @@ public:
"Expected signature `void (State&, downstream<Out>&, In)` " "Expected signature `void (State&, downstream<Out>&, In)` "
"for consume function"); "for consume function");
using driver = using driver =
detail::stream_stage_driver_impl<typename Trait::input, output_type, Fun, detail::stream_stage_driver_impl<typename Trait::input, Scatterer, Fun,
Cleanup, Cleanup,
std::tuple<detail::decay_t<Ts>...>>; std::tuple<detail::decay_t<Ts>...>>;
return make_stage<driver, Scatterer>(in, std::move(init), std::move(fun), return make_stage<driver>(in, std::move(init), std::move(fun),
std::move(cleanup), std::move(xs)); std::move(cleanup), std::move(xs));
} }
/// Returns a stream manager (implementing a continuous stage) without in- or /// Returns a stream manager (implementing a continuous stage) without in- or
/// outbound path. The returned manager is not connected to any slot and thus /// outbound path. The returned manager is not connected to any slot and thus
/// not stored by the actor automatically. /// not stored by the actor automatically.
template <class Driver, template <class Driver, class... Ts>
class Scatterer = broadcast_scatterer<typename Driver::output_type>,
class Input = int, class... Ts>
typename Driver::stage_ptr_type make_continuous_stage(Ts&&... xs) { typename Driver::stage_ptr_type make_continuous_stage(Ts&&... xs) {
auto ptr = detail::make_stream_stage<Driver, Scatterer>( auto ptr = detail::make_stream_stage<Driver>(this, std::forward<Ts>(xs)...);
this, std::forward<Ts>(xs)...);
ptr->continuous(true); ptr->continuous(true);
return std::move(ptr); return std::move(ptr);
} }
...@@ -586,8 +582,8 @@ public: ...@@ -586,8 +582,8 @@ public:
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>, broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>,
class Trait = stream_stage_trait_t<Fun>> class Trait = stream_stage_trait_t<Fun>>
stream_stage_ptr<typename Trait::input, typename Trait::output, stream_stage_ptr<typename Trait::input, message, typename Trait::output,
detail::decay_t<Ts>...> Scatterer, detail::decay_t<Ts>...>
make_continuous_stage(std::tuple<Ts...> xs, Init init, Fun fun, make_continuous_stage(std::tuple<Ts...> xs, Init init, Fun fun,
Cleanup cleanup, Cleanup cleanup,
policy::arg<Scatterer> scatterer_type = {}) { policy::arg<Scatterer> scatterer_type = {}) {
...@@ -607,19 +603,19 @@ public: ...@@ -607,19 +603,19 @@ public:
"Expected signature `void (State&, downstream<Out>&, In)` " "Expected signature `void (State&, downstream<Out>&, In)` "
"for consume function"); "for consume function");
using driver = using driver =
detail::stream_stage_driver_impl<typename Trait::input, output_type, Fun, detail::stream_stage_driver_impl<typename Trait::input, Scatterer, Fun,
Cleanup, Cleanup,
std::tuple<detail::decay_t<Ts>...>>; std::tuple<detail::decay_t<Ts>...>>;
return make_continuous_stage<driver, Scatterer>( return make_continuous_stage<driver>(std::move(init), std::move(fun),
std::move(init), std::move(fun), std::move(cleanup), std::move(xs)); std::move(cleanup), std::move(xs));
} }
template <class Init, class Fun, class Cleanup, template <class Init, class Fun, class Cleanup,
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>, broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>,
class Trait = stream_stage_trait_t<Fun>> class Trait = stream_stage_trait_t<Fun>>
stream_stage_ptr<typename stream_stage_trait_t<Fun>::input, stream_stage_ptr<typename stream_stage_trait_t<Fun>::input, message,
typename stream_stage_trait_t<Fun>::output> typename stream_stage_trait_t<Fun>::output, Scatterer>
make_continuous_stage(Init init, Fun fun, Cleanup cleanup, make_continuous_stage(Init init, Fun fun, Cleanup cleanup,
policy::arg<Scatterer> scatterer_type = {}) { policy::arg<Scatterer> scatterer_type = {}) {
return make_continuous_stage(std::make_tuple(), std::move(init), return make_continuous_stage(std::make_tuple(), std::move(init),
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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://opensink.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_DETAIL_STREAM_SINK_HPP
#define CAF_DETAIL_STREAM_SINK_HPP
#include <tuple>
#include "caf/intrusive_ptr.hpp"
#include "caf/stream_manager.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf {
template <class In, class Result>
class stream_sink : public virtual stream_manager {
public:
// -- member types -----------------------------------------------------------
using input_type = In;
// -- constructors, destructors, and assignment operators --------------------
stream_sink(local_actor* self) : stream_manager(self) {
// nop
}
};
template <class In, class Result>
using stream_sink_ptr = intrusive_ptr<stream_sink<In, Result>>;
} // namespace caf
#endif // CAF_DETAIL_STREAM_SINK_HPP
...@@ -45,7 +45,7 @@ public: ...@@ -45,7 +45,7 @@ public:
// -- virtual functions ------------------------------------------------------ // -- virtual functions ------------------------------------------------------
/// Cleans up any state and produces a result message. /// Produces a result message after closing the last inbound path.
virtual message make_final_result() { virtual message make_final_result() {
return make_message(); return make_message();
} }
......
...@@ -24,10 +24,12 @@ ...@@ -24,10 +24,12 @@
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
template <class Out, class... Ts> template <class Out, class Scatterer, class... Ts>
class stream_source : public stream_manager { class stream_source : public virtual stream_manager {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -35,13 +37,26 @@ public: ...@@ -35,13 +37,26 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_source(local_actor* self) : stream_manager(self) { stream_source(local_actor* self) : stream_manager(self), out_(self) {
// nop // nop
} }
Scatterer& out() override {
return out_;
}
protected:
Scatterer out_;
}; };
template <class Out, class... HandshakeData> template <class Out, class Scatterer, class... HandshakeData>
using stream_source_ptr = intrusive_ptr<stream_source<Out, HandshakeData...>>; using stream_source_ptr =
intrusive_ptr<stream_source<Out, Scatterer, HandshakeData...>>;
template <class Scatterer, class... Ts>
using stream_source_ptr_t =
stream_source_ptr<typename Scatterer::value_type, Scatterer,
detail::decay_t<Ts>...>;
} // namespace caf } // namespace caf
......
...@@ -27,23 +27,36 @@ ...@@ -27,23 +27,36 @@
namespace caf { namespace caf {
/// Identifies an unbound sequence of messages. /// Identifies an unbound sequence of messages.
template <class Output, class... HandshakeData> template <class Scatterer, class... Ts>
class stream_source_driver { class stream_source_driver {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using output_type = Output; /// Policy for distributing data to outbound paths.
using scatterer_type = Scatterer;
/// Element type of the output stream.
using output_type = typename scatterer_type::value_type;
/// Type of the output stream.
using stream_type = stream<output_type>; using stream_type = stream<output_type>;
using output_stream_type = output_stream<output_type, HandshakeData...>; /// Type of the output stream including handshake argument types.
using output_stream_type = output_stream<output_type, Ts...>;
using handshake_tuple_type = std::tuple<stream_type, HandshakeData...>; /// Tuple for creating the `open_stream_msg` handshake.
using handshake_tuple_type = std::tuple<stream_type, Ts...>;
using source_type = stream_source<output_type, HandshakeData...>; /// Implemented `stream_source` interface.
using source_type = stream_source<output_type, scatterer_type, Ts...>;
/// Smart pointer to the interface type.
using source_ptr_type = intrusive_ptr<source_type>; using source_ptr_type = intrusive_ptr<source_type>;
/// Wrapper type holding a pointer to `source_type`.
using make_source_result_type = make_source_result<output_type,
scatterer_type, Ts...>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
virtual ~stream_source_driver() { virtual ~stream_source_driver() {
......
...@@ -22,28 +22,36 @@ ...@@ -22,28 +22,36 @@
#include <tuple> #include <tuple>
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/stream_sink.hpp"
#include "caf/stream_source.hpp" #include "caf/stream_source.hpp"
namespace caf { namespace caf {
template <class In, class Out, class... HandshakeData> template <class In, class Result, class Out, class Scatterer,
class stream_stage : public stream_source<Out, HandshakeData...> { class... HandshakeData>
class stream_stage : public stream_source<Out, Scatterer, HandshakeData...>,
public stream_sink<In, Result> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = stream_source<Out, HandshakeData...>; using left_super = stream_source<Out, Scatterer, HandshakeData...>;
using input_type = In; using right_super = stream_sink<In, Result>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_stage(local_actor* self) : super(self) { stream_stage(local_actor* self)
: stream_manager(self),
left_super(self),
right_super(self) {
// nop // nop
} }
}; };
template <class In, class Out, class... HandshakeData> template <class In, class Result, class Out, class Scatterer,
using stream_stage_ptr = intrusive_ptr<stream_stage<In, Out, HandshakeData...>>; class... HandshakeData>
using stream_stage_ptr =
intrusive_ptr<stream_stage<In, Result, Out, Scatterer, HandshakeData...>>;
} // namespace caf } // namespace caf
......
...@@ -24,29 +24,50 @@ ...@@ -24,29 +24,50 @@
#include <vector> #include <vector>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/message.hpp"
namespace caf { namespace caf {
/// Identifies an unbound sequence of messages. /// Encapsulates user-provided functionality for generating a stream stage.
template <class Input, class Output, class... HandshakeData> template <class Input, class Result, class Scatterer, class... Ts>
class stream_stage_driver { class stream_stage_driver {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
/// Element type of the input stream.
using input_type = Input; using input_type = Input;
using output_type = Output; /// Result type shipped to the client after processing all elements of the
/// stream.
using result_type = Result;
/// Policy for distributing data to outbound paths.
using scatterer_type = Scatterer;
/// Element type of the output stream.
using output_type = typename scatterer_type::value_type;
/// Type of the output stream.
using stream_type = stream<output_type>; using stream_type = stream<output_type>;
using output_stream_type = output_stream<output_type, HandshakeData...>; /// Type of the output stream including handshake argument types.
using output_stream_type = output_stream<output_type, Ts...>;
using handshake_tuple_type = std::tuple<stream_type, HandshakeData...>; /// Tuple for creating the `open_stream_msg` handshake.
using handshake_tuple_type = std::tuple<stream_type, Ts...>;
using stage_type = stream_stage<input_type, output_type, HandshakeData...>; /// Implemented `stream_stage` interface.
using stage_type = stream_stage<input_type, result_type,
output_type, Scatterer, Ts...>;
/// Smart pointer to the interface type.
using stage_ptr_type = intrusive_ptr<stage_type>; using stage_ptr_type = intrusive_ptr<stage_type>;
/// Wrapper type holding a pointer to `stage_type`.
using make_stage_result_type =
make_stage_result<input_type, result_type, output_type, scatterer_type,
Ts...>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
virtual ~stream_stage_driver() { virtual ~stream_stage_driver() {
...@@ -55,19 +76,22 @@ public: ...@@ -55,19 +76,22 @@ public:
// -- virtual functions ------------------------------------------------------ // -- virtual functions ------------------------------------------------------
/// Cleans up any state.
virtual void finalize(const error&) {
// nop
}
// -- pure virtual functions -------------------------------------------------
/// Generates handshake data for the next actor in the pipeline. /// Generates handshake data for the next actor in the pipeline.
virtual handshake_tuple_type make_handshake(stream_slot slot) const = 0; virtual handshake_tuple_type make_handshake(stream_slot slot) const = 0;
/// Processes a single batch. /// Processes a single batch.
virtual void process(std::vector<input_type>&& batch, virtual void process(std::vector<input_type>&& batch,
downstream<output_type>& out) = 0; downstream<output_type>& out) = 0;
/// Handles the result of an outbound path.
virtual void add_result(message&) = 0;
/// Produces a result message after receiving the result messages of all
/// outbound paths and closing all paths.
virtual message make_final_result() = 0;
/// Cleans up any state.
virtual void finalize(const error&) = 0;
}; };
} // namespace caf } // namespace caf
......
...@@ -104,7 +104,7 @@ TESTEE(sum_up) { ...@@ -104,7 +104,7 @@ TESTEE(sum_up) {
} }
TESTEE_STATE(stream_multiplexer) { TESTEE_STATE(stream_multiplexer) {
stream_stage_ptr<int, int, std::string> stage; stream_stage_ptr<int, message, int, broadcast_scatterer<int>, string> stage;
}; };
TESTEE(stream_multiplexer) { TESTEE(stream_multiplexer) {
......
...@@ -239,7 +239,8 @@ public: ...@@ -239,7 +239,8 @@ public:
auto slot = next_slot_++; auto slot = next_slot_++;
CAF_MESSAGE(name_ << " starts streaming to " << ref.name() CAF_MESSAGE(name_ << " starts streaming to " << ref.name()
<< " on slot " << slot); << " on slot " << slot);
struct driver final : public stream_source_driver<int> { using scatterer = broadcast_scatterer<int>;
struct driver final : public stream_source_driver<scatterer> {
public: public:
driver(int sentinel) : x_(0), sentinel_(sentinel) { driver(int sentinel) : x_(0), sentinel_(sentinel) {
// nop // nop
...@@ -272,7 +273,8 @@ public: ...@@ -272,7 +273,8 @@ public:
auto slot = next_slot_++; auto slot = next_slot_++;
CAF_MESSAGE(name_ << " starts forwarding to " << ref.name() CAF_MESSAGE(name_ << " starts forwarding to " << ref.name()
<< " on slot " << slot); << " on slot " << slot);
struct driver final : public stream_stage_driver<int, int> { using scatterer = broadcast_scatterer<int>;
struct driver final : public stream_stage_driver<int, none_t, scatterer> {
public: public:
driver(vector<int>* log) : log_(log) { driver(vector<int>* log) : log_(log) {
// nop // nop
...@@ -286,6 +288,19 @@ public: ...@@ -286,6 +288,19 @@ public:
log_->insert(log_->end(), batch.begin(), batch.end()); log_->insert(log_->end(), batch.begin(), batch.end());
out.append(batch.begin(), batch.end()); out.append(batch.begin(), batch.end());
} }
void add_result(message&) override {
// nop
}
message make_final_result() override {
return none;
}
void finalize(const error&) override {
// nop
}
private: private:
vector<int>* log_; vector<int>* log_;
}; };
......
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