Commit c8a35b24 authored by Dominik Charousset's avatar Dominik Charousset

Redefine role of output_stream

An `output_stream` now has the same role as `stream`: a marker type for
type-checking only. All `make_(stage|source|sink)` factory functions now
return `make_(stage|source|sink)_result` that wraps IDs for created
slots and a pointer to the created manager. All variants of
`add_..._path` now return a typed wrapper holding the slot ID.
parent ef999854
......@@ -81,20 +81,20 @@ 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_t<M, Ms...>>,
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_t<M, Ms...>>>> {
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_t<O, Out...>>, Xs...>,
output_stream<O, Out...>>, Xs...>,
Ys,
detail::type_list<typed_mpi<detail::type_list<O, Out...>,
output_tuple<M, Ms...>>, Zs...>,
......@@ -102,22 +102,22 @@ struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>,
: composed_type<detail::type_list<Xs...>, Ys, Ys,
detail::type_list<Rs...,
typed_mpi<detail::type_list<In...>,
output_stream_t<M, Ms...>>>> {
output_stream<M, Ms...>>>> {
};
// 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_t<O, Out...>>, Xs...>,
output_stream<O, Out...>>, Xs...>,
Ys,
detail::type_list<typed_mpi<detail::type_list<O, Out...>,
output_stream_t<M, Ms...>>, Zs...>,
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_t<M, Ms...>>>> {
output_stream<M, Ms...>>>> {
};
// default case (recurse over Zs)
......
......@@ -26,7 +26,6 @@
#include "caf/expected.hpp"
#include "caf/optional.hpp"
#include "caf/replies_to.hpp"
#include "caf/stream_result.hpp"
#include "caf/detail/implicit_conversions.hpp"
......@@ -52,21 +51,14 @@ struct dmi<result<Ys...> (Xs...)> {
output_tuple<implicit_conversions_t<Ys>...>>;
};
// case #2b: function returning a stream_result<...>
template <class Y, class... Xs>
struct dmi<stream_result<Y> (Xs...)> {
using type = typed_mpi<type_list<typename param_decay<Xs>::type...>,
output_tuple<implicit_conversions_t<Y>>>;
};
// case #2c: function returning a std::tuple<...>
// case #2b: function returning a std::tuple<...>
template <class... Ys, class... Xs>
struct dmi<std::tuple<Ys...> (Xs...)> {
using type = typed_mpi<type_list<typename param_decay<Xs>::type...>,
output_tuple<implicit_conversions_t<Ys>...>>;
};
// case #2d: function returning a std::tuple<...>
// case #2c: function returning a std::tuple<...>
template <class... Ys, class... Xs>
struct dmi<delegated<Ys...> (Xs...)> {
using type = typed_mpi<type_list<typename param_decay<Xs>::type...>,
......
......@@ -21,16 +21,18 @@
#include <tuple>
#include "caf/expected.hpp"
#include "caf/fwd.hpp"
#include "caf/none.hpp"
#include "caf/unit.hpp"
#include "caf/skip.hpp"
#include "caf/result.hpp"
#include "caf/make_message.hpp"
#include "caf/make_sink_result.hpp"
#include "caf/make_source_result.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/message.hpp"
#include "caf/expected.hpp"
#include "caf/none.hpp"
#include "caf/optional.hpp"
#include "caf/make_message.hpp"
#include "caf/stream_result.hpp"
#include "caf/result.hpp"
#include "caf/skip.hpp"
#include "caf/unit.hpp"
#include "caf/detail/apply_args.hpp"
#include "caf/detail/int_list.hpp"
......@@ -139,15 +141,39 @@ public:
(*this)();
}
/// Calls `(*this)(x.in(), x.out(), x.ptr())`.
template <class Out, class Tuple, class P>
void operator()(output_stream<Out, Tuple, P>&) {
/// Calls `(*this)()`.
template <class Out, class... Ts>
void operator()(output_stream<Out, Ts...>&) {
(*this)();
}
/// Calls `(*this)(x.in(), 0, x.ptr())`.
template <class T>
void operator()(stream_result<T>&) {
/// Calls `(*this)()`.
template <class Out, class... Ts>
void operator()(outbound_stream_slot<Out, Ts...>&) {
(*this)();
}
/// Calls `(*this)()`.
template <class In>
void operator()(inbound_stream_slot<In>&) {
(*this)();
}
/// Calls `(*this)()`.
template <class In>
void operator()(make_sink_result<In>&) {
(*this)();
}
/// Calls `(*this)()`.
template <class DownstreamManager, class... Ts>
void operator()(make_source_result<DownstreamManager, Ts...>&) {
(*this)();
}
/// Calls `(*this)()`.
template <class In, class DownstreamManager, class... Ts>
void operator()(make_stage_result<In, DownstreamManager, Ts...>&) {
(*this)();
}
......
......@@ -61,6 +61,8 @@ template <class...> class typed_event_based_actor;
template <class...> class typed_response_promise;
// -- variadic templates with fixed arguments ----------------------------------
//
template <class, class...> class output_stream;
// -- classes ------------------------------------------------------------------
......@@ -225,14 +227,6 @@ using stream_manager_ptr = intrusive_ptr<stream_manager>;
using type_erased_value_ptr = std::unique_ptr<type_erased_value>;
using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
// -- templates with default parameters ----------------------------------------
template <class, class = std::tuple<>, class = stream_manager_ptr>
class output_stream;
template <class = stream_manager_ptr>
class stream_result;
} // namespace caf
#endif // CAF_FWD_HPP
......@@ -16,78 +16,71 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_RESULT_HPP
#define CAF_STREAM_RESULT_HPP
#ifndef CAF_MAKE_SINK_RESULT_HPP
#define CAF_MAKE_SINK_RESULT_HPP
#include "caf/fwd.hpp"
#include "caf/none.hpp"
#include "caf/stream_sink.hpp"
#include "caf/stream_slot.hpp"
namespace caf {
/// Terminates a stream by reducing it to a single value.
template <class Pointer /* = stream_manager_ptr */>
class stream_result {
/// Returns a stream sink with the slot ID of its first inbound path.
template <class In>
class make_sink_result {
public:
// -- member types -----------------------------------------------------------
using pointer_type = Pointer;
/// Type of a single element.
using input_type = In;
// -- constructors, destructors, and assignment operators --------------------
/// Fully typed stream manager as returned by `make_source`.
using sink_type = stream_sink<In>;
stream_result(stream_result&&) = default;
stream_result(const stream_result&) = default;
stream_result& operator=(stream_result&&) = default;
stream_result& operator=(const stream_result&) = default;
/// Pointer to a fully typed stream manager.
using sink_ptr_type = intrusive_ptr<sink_type>;
stream_result(none_t = none) : in_(0) {
// nop
}
/// The return type for `scheduled_actor::make_sink`.
using output_stream_type = stream<input_type>;
stream_result(stream_slot id, Pointer mgr = nullptr)
: in_(id),
ptr_(std::move(mgr)) {
// -- constructors, destructors, and assignment operators --------------------
make_sink_result() noexcept : slot_(0) {
// nop
}
template <class CompatiblePointer>
stream_result(stream_result<CompatiblePointer> other)
: in_(other.in()),
ptr_(std::move(other.ptr())) {
make_sink_result(stream_slot slot, sink_ptr_type ptr) noexcept
: slot_(slot),
ptr_(std::move(ptr)) {
// nop
}
template <class CompatiblePointer>
stream_result& operator=(stream_result<CompatiblePointer> other) {
in_ = other.in();
ptr_ = std::move(other.ptr());
return *this;
}
make_sink_result(make_sink_result&&) = default;
make_sink_result(const make_sink_result&) = default;
make_sink_result& operator=(make_sink_result&&) = default;
make_sink_result& operator=(const make_sink_result&) = default;
// -- properties -------------------------------------------------------------
/// Returns the unique identifier for this stream_result.
inline stream_slot in() const noexcept {
return in_;
inline stream_slot inbound_slot() const noexcept {
return slot_;
}
/// Returns the handler assigned to this stream on this actor.
inline pointer_type& ptr() noexcept {
inline sink_ptr_type& ptr() noexcept {
return ptr_;
}
/// Returns the handler assigned to this stream on this actor.
inline const pointer_type& ptr() const noexcept {
inline const sink_ptr_type& ptr() const noexcept {
return ptr_;
}
private:
// -- member variables -------------------------------------------------------
stream_slot in_;
pointer_type ptr_;
stream_slot slot_;
sink_ptr_type ptr_;
};
} // namespace caf
#endif // CAF_STREAM_RESULT_HPP
#endif // CAF_MAKE_SINK_RESULT_HPP
......@@ -27,10 +27,11 @@
namespace caf {
/// Helper trait for deducing an `output_stream` from the arguments to
/// `scheduled_actor::make_source`.
/// Returns a stream source with the slot ID of its first outbound path.
template <class DownstreamManager, class... Ts>
struct make_source_result {
// -- member types -----------------------------------------------------------
/// Type of a single element.
using output_type = typename DownstreamManager::output_type;
......@@ -41,15 +42,54 @@ struct make_source_result {
using source_ptr_type = intrusive_ptr<source_type>;
/// The return type for `scheduled_actor::make_source`.
using type = output_stream<output_type, std::tuple<Ts...>, source_ptr_type>;
using output_stream_type = output_stream<output_type, Ts...>;
// -- constructors, destructors, and assignment operators --------------------
make_source_result() noexcept : slot_(0) {
// nop
}
make_source_result(stream_slot slot, source_ptr_type ptr) noexcept
: slot_(slot),
ptr_(std::move(ptr)) {
// nop
}
make_source_result(make_source_result&&) = default;
make_source_result(const make_source_result&) = default;
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 {
return slot_;
}
inline source_ptr_type& ptr() noexcept {
return ptr_;
}
inline const source_ptr_type& ptr() const noexcept {
return ptr_;
}
private:
stream_slot slot_;
source_ptr_type ptr_;
};
/// Helper type for defining an `output_stream` from a downstream manager plus
/// the types of the handshake arguments.
/// @relates make_source_result
template <class DownstreamManager, class... Ts>
using make_source_result_t =
typename make_source_result<DownstreamManager,
detail::strip_and_convert_t<Ts>...>::type;
make_source_result<DownstreamManager, detail::strip_and_convert_t<Ts>...>;
} // namespace caf
......
......@@ -28,30 +28,81 @@
namespace caf {
/// Helper trait for deducing an `output_stream` from the arguments to
/// `scheduled_actor::make_stage`.
/// 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 {
public:
// -- member types -----------------------------------------------------------
/// Type of a single element.
using input_type = In;
/// Type of a single element.
using output_type = typename DownstreamManager::output_type;
/// Fully typed stream manager as returned by `make_stage`.
/// Fully typed stream manager as returned by `make_source`.
using stage_type = stream_stage<In, DownstreamManager>;
/// Pointer to a fully typed stream manager.
using stage_ptr_type = intrusive_ptr<stage_type>;
/// The return type for `scheduled_actor::make_sink`.
using type = output_stream<output_type, std::tuple<Ts...>, stage_ptr_type>;
/// The return type for `scheduled_actor::make_stage`.
using output_stream_type = output_stream<output_type, Ts...>;
// -- constructors, destructors, and assignment operators --------------------
make_stage_result() noexcept : inbound_slot_(0), outbound_slot_(0) {
// nop
}
make_stage_result(stream_slot in, stream_slot out, stage_ptr_type ptr)
: inbound_slot_(in),
outbound_slot_(out),
ptr_(std::move(ptr)) {
// nop
}
make_stage_result(make_stage_result&&) = default;
make_stage_result(const make_stage_result&) = default;
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 {
return inbound_slot_;
}
inline stream_slot outbound_slot() const noexcept {
return outbound_slot_;
}
inline stage_ptr_type& ptr() noexcept {
return ptr_;
}
inline const stage_ptr_type& ptr() const noexcept {
return ptr_;
}
private:
stream_slot inbound_slot_;
stream_slot outbound_slot_;
stage_ptr_type ptr_;
};
/// Helper type for defining a `make_stage_result` from a downstream manager
/// plus additional handshake types. Hardwires `message` as result type.
template <class In, class DownstreamManager, class... Ts>
using make_stage_result_t =
typename make_stage_result<In, DownstreamManager,
detail::strip_and_convert_t<Ts>...>::type;
make_stage_result<In, DownstreamManager, detail::strip_and_convert_t<Ts>...>;
} // namespace caf
......
......@@ -19,107 +19,15 @@
#ifndef CAF_OUTPUT_STREAM_HPP
#define CAF_OUTPUT_STREAM_HPP
#include "caf/fwd.hpp"
#include "caf/invalid_stream.hpp"
#include "caf/stream_slot.hpp"
namespace caf {
/// Identifies an unbound sequence of elements annotated with the additional
/// handshake arguments emitted to the next stage.
template <class T, class... Ts, class Pointer /* = stream_manager_ptr */>
class output_stream<T, std::tuple<Ts...>, Pointer> {
/// Empty marker type for type-checking of stream sources and stages.
template <class T, class... Ts>
class output_stream {
public:
// -- member types -----------------------------------------------------------
/// Type of a single element.
using value_type = T;
/// Type of the handshake.
using tuple_type = std::tuple<Ts...>;
/// Type of the stream manager pointer.
using pointer_type = Pointer;
// -- constructors, destructors, and assignment operators --------------------
output_stream(output_stream&&) = default;
output_stream(const output_stream&) = default;
output_stream& operator=(output_stream&&) = default;
output_stream& operator=(const output_stream&) = default;
output_stream(invalid_stream_t) : in_(0), out_(0) {
// nop
}
output_stream(stream_slot in_slot, stream_slot out_slot,
pointer_type mgr)
: in_(in_slot),
out_(out_slot),
ptr_(std::move(mgr)) {
// nop
}
template <class CompatiblePointer>
output_stream(output_stream<T, tuple_type, CompatiblePointer> x)
: in_(x.in()),
out_(x.out()),
ptr_(std::move(x.ptr())) {
// nop
}
template <class CompatiblePointer>
output_stream& operator=(output_stream<T, tuple_type, CompatiblePointer> x) {
in_ = x.in();
out_ = x.out();
ptr_ = std::move(x.ptr());
return *this;
}
// -- 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 noexcept {
return in_;
}
/// Returns the output slot.
inline stream_slot out() const noexcept {
return out_;
}
/// Returns the handler assigned to this stream on this actor.
inline pointer_type& ptr() noexcept {
return ptr_;
}
/// Returns the handler assigned to this stream on this actor.
inline const pointer_type& ptr() const noexcept {
return ptr_;
}
// -- conversions ------------------------------------------------------------
/// Rebinds the `ptr()` property.
template <class Manager>
output_stream<value_type, tuple_type, intrusive_ptr<Manager>>
rebind(Manager* ptr) const noexcept {
return {in_, out_, ptr};
}
private:
// -- member variables -------------------------------------------------------
stream_slot in_;
stream_slot out_;
pointer_type ptr_;
};
/// Convenience alias for `output_stream<T, std::tuple<Ts...>>`.
template <class T, class... Ts>
using output_stream_t = output_stream<T, std::tuple<Ts...>>;
} // namespace caf
#endif // CAF_OUTPUT_STREAM_HPP
......@@ -22,6 +22,7 @@
#include <string>
#include "caf/illegal_message_element.hpp"
#include "caf/output_stream.hpp"
#include "caf/stream.hpp"
#include "caf/detail/type_list.hpp"
......@@ -51,7 +52,7 @@ struct replies_to {
/// @private
template <class O, class... Os>
using with_stream = typed_mpi<detail::type_list<Is...>,
output_stream_t<O, Os...>>;
output_stream<O, Os...>>;
};
template <class... Is>
......
......@@ -40,6 +40,8 @@
#include "caf/invoke_message_result.hpp"
#include "caf/local_actor.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/output_stream.hpp"
......@@ -48,8 +50,9 @@
#include "caf/sec.hpp"
#include "caf/stream.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_result.hpp"
#include "caf/stream_sink_trait.hpp"
#include "caf/stream_source_trait.hpp"
#include "caf/stream_stage_trait.hpp"
#include "caf/to_string.hpp"
#include "caf/policy/arg.hpp"
......@@ -437,7 +440,8 @@ public:
auto mgr = make_stream_source<Driver>(this, std::move(init),
std::move(pull), std::move(done),
std::move(fin));
return mgr->add_outbound_path(std::move(xs));
auto slot = mgr->add_outbound_path(std::move(xs));
return {slot, std::move(mgr)};
}
/// Creates a new stream source from given arguments.
......@@ -463,7 +467,8 @@ public:
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>>
make_source_result_t<DownstreamManager>
detail::enable_if_t<!detail::is_actor_handle<Init>::value,
make_source_result_t<DownstreamManager>>
make_source(Init init, Pull pull, Done done, Finalize finalize = {},
policy::arg<DownstreamManager> token = {}) {
return make_source(std::make_tuple(), init, pull, done, finalize,
......@@ -487,15 +492,8 @@ public:
std::move(pull),
std::move(done),
std::move(fin));
return mgr->add_outbound_path(dest, std::move(xs));
/*
auto slot = mgr->add_unchecked_outbound_path_impl(
actor_cast<strong_actor_ptr>(dest), std::move(handshake));
if (slot == invalid_stream_slot) {
CAF_LOG_WARNING("unable to assign a slot in make_source");
}
return mgr;
*/
auto slot = mgr->add_outbound_path(dest, std::move(xs));
return {slot, std::move(mgr)};
}
/// Creates a new stream source and adds `dest` as first outbound path to it.
......@@ -513,18 +511,18 @@ public:
}
template <class Driver, class... Ts>
stream_result<typename Driver::sink_ptr_type>
make_sink_result<typename Driver::input_type>
make_sink(const stream<typename Driver::input_type>& src, Ts&&... xs) {
auto mgr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...);
return mgr->add_inbound_path(src);
auto slot = mgr->add_inbound_path(src);
return {slot, std::move(mgr)};
}
template <class Input, class Init, class Fun, class Finalize = unit_t,
template <class In, class Init, class Fun, class Finalize = unit_t,
class Trait = stream_sink_trait_t<Fun>>
stream_result<typename Trait::pointer>
make_sink(const stream<Input>& in, Init init, Fun fun, Finalize fin = {}) {
using driver = detail::stream_sink_driver_impl<typename Trait::input,
Fun, Finalize>;
make_sink_result<In> make_sink(const stream<In>& in, Init init, Fun fun,
Finalize fin = {}) {
using driver = detail::stream_sink_driver_impl<In, Fun, Finalize>;
return make_sink<driver>(in, std::move(init), std::move(fun),
std::move(fin));
}
......@@ -534,8 +532,8 @@ public:
make_stage(const stream<In>& src, std::tuple<Ts...> xs, Us&&... ys) {
using detail::make_stream_stage;
auto mgr = make_stream_stage<Driver>(this, std::forward<Us>(ys)...);
auto in = mgr->add_inbound_path(src).in();
auto out = mgr->add_outbound_path(std::move(xs)).out();
auto in = mgr->add_inbound_path(src);
auto out = mgr->add_outbound_path(std::move(xs));
return {in, out, std::move(mgr)};
}
......
......@@ -33,7 +33,6 @@
#include "caf/message_builder.hpp"
#include "caf/output_stream.hpp"
#include "caf/ref_counted.hpp"
#include "caf/stream_result.hpp"
#include "caf/stream_slot.hpp"
#include "caf/upstream_msg.hpp"
......@@ -155,20 +154,20 @@ public:
/// @pre `out().terminal() == false`
/// @private
template <class Out>
output_stream_t<Out> add_unchecked_outbound_path() {
outbound_stream_slot<Out> add_unchecked_outbound_path() {
auto handshake = make_message(stream<Out>{});
return {0, add_unchecked_outbound_path_impl(std::move(handshake)), this};
return add_unchecked_outbound_path_impl(std::move(handshake));
}
/// Creates an outbound path to the current sender without any type checking.
/// @pre `out().terminal() == false`
/// @private
template <class Out, class... Ts>
output_stream_t<Out, detail::strip_and_convert_t<Ts>...>
outbound_stream_slot<Out, detail::strip_and_convert_t<Ts>...>
add_unchecked_outbound_path(std::tuple<Ts...> xs) {
auto tk = std::make_tuple(stream<Out>{});
auto handshake = make_message_from_tuple(std::tuple_cat(tk, std::move(xs)));
return {0, add_unchecked_outbound_path_impl(std::move(handshake)), this};
return add_unchecked_outbound_path_impl(std::move(handshake));
}
/// Creates an outbound path to `next`, only checking whether the interface
......@@ -178,13 +177,12 @@ public:
/// @pre `out().terminal() == false`
/// @private
template <class Out, class Handle>
output_stream_t<Out> add_unchecked_outbound_path(Handle next) {
// TODO: type checking
outbound_stream_slot<Out> add_unchecked_outbound_path(Handle next) {
// TODO: type-check whether `next` accepts our handshake
auto handshake = make_message(stream<Out>{});
auto hdl = actor_cast<strong_actor_ptr>(std::move(next));
auto slot = add_unchecked_outbound_path_impl(std::move(hdl),
std::move(handshake));
return {0, slot, this};
return add_unchecked_outbound_path_impl(std::move(hdl),
std::move(handshake));
}
/// Creates an outbound path to `next`, only checking whether the interface
......@@ -194,15 +192,14 @@ public:
/// @pre `out().terminal() == false`
/// @private
template <class Out, class Handle, class... Ts>
output_stream_t<Out, detail::strip_and_convert_t<Ts>...>
outbound_stream_slot<Out, detail::strip_and_convert_t<Ts>...>
add_unchecked_outbound_path(const Handle& next, std::tuple<Ts...> xs) {
// TODO: type checking
// TODO: type-check whether `next` accepts our handshake
auto tk = std::make_tuple(stream<Out>{});
auto handshake = make_message_from_tuple(std::tuple_cat(tk, std::move(xs)));
auto hdl = actor_cast<strong_actor_ptr>(std::move(next));
auto slot = add_unchecked_outbound_path_impl(std::move(hdl),
std::move(handshake));
return {0, slot, this};
return add_unchecked_outbound_path_impl(std::move(hdl),
std::move(handshake));
}
/// Creates an inbound path to the current sender without any type checking.
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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://openresult_handler.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_RESULT_HANDLER_TRAIT_HPP
#define CAF_STREAM_RESULT_HANDLER_TRAIT_HPP
#include "caf/fwd.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf {
/// Deduces the input type for a stream result handler from its signature.
template <class F>
struct stream_result_handler_trait {
static constexpr bool valid = false;
using result = void;
};
template <class T>
struct stream_result_handler_trait<void (expected<T>)> {
static constexpr bool valid = true;
using result = T;
};
/// Convenience alias for extracting the function signature from `Pull` and
/// passing it to `stream_result_handler_trait`.
template <class Pull>
using stream_result_handler_trait_t =
stream_result_handler_trait<typename detail::get_callable_trait<Pull>::fun_sig>;
} // namespace caf
#endif // CAF_STREAM_RESULT_HANDLER_TRAIT_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://openresult.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_RESULT_TRAIT_HPP
#define CAF_STREAM_RESULT_TRAIT_HPP
#include "caf/unit.hpp"
#include "caf/behavior.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf {
template <class F>
struct stream_result_trait;
template <class T>
struct stream_result_trait<void (expected<T>)> {
using type = T;
template <class OnResult>
static behavior make_result_handler(OnResult f) {
return {
[=](T& res) {
f(std::move(res));
},
[=](error& err) {
f(std::move(err));
}
};
}
};
template <>
struct stream_result_trait<void (expected<void>)> {
using type = void;
template <class OnResult>
static behavior make_result_handler(OnResult f) {
return {
[=]() {
f(unit);
},
[=](error& err) {
f(std::move(err));
}
};
}
};
template <class F>
using stream_result_trait_t =
stream_result_trait<typename detail::get_callable_trait<F>::fun_sig>;
} // namespace caf
#endif // CAF_STREAM_RESULT_TRAIT_HPP
......@@ -23,7 +23,6 @@
#include "caf/intrusive_ptr.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_result.hpp"
#include "caf/detail/type_traits.hpp"
......@@ -55,9 +54,8 @@ public:
// -- properties -------------------------------------------------------------
/// Creates a new input path to the current sender.
stream_result<intrusive_ptr<stream_sink>>
add_inbound_path(const stream<input_type>&) {
return {add_unchecked_inbound_path_impl(), this};
inbound_stream_slot<input_type> add_inbound_path(const stream<input_type>&) {
return {add_unchecked_inbound_path_impl()};
}
private:
......
......@@ -21,6 +21,8 @@
#include <cstdint>
#include "caf/output_stream.hpp"
#include "caf/detail/comparable.hpp"
namespace caf {
......@@ -65,6 +67,90 @@ struct stream_slots : detail::comparable<stream_slots>{
}
};
/// Wraps a stream slot ID for inbound paths with the full type information of
/// the path creation.
template <class In>
class inbound_stream_slot {
public:
// -- member types -----------------------------------------------------------
/// Type of a single element.
using input_type = In;
/// The return type for `scheduled_actor::make_source`.
using input_stream_type = stream<input_type>;
// -- constructors, destructors, and assignment operators --------------------
constexpr inbound_stream_slot(stream_slot value = 0): value_(value) {
// nop
}
inbound_stream_slot(inbound_stream_slot&&) = default;
inbound_stream_slot(const inbound_stream_slot&) = default;
inbound_stream_slot& operator=(inbound_stream_slot&&) = default;
inbound_stream_slot& operator=(const inbound_stream_slot&) = default;
// -- conversion operators ---------------------------------------------------
constexpr operator stream_slot() const noexcept {
return value_;
}
// -- properties -------------------------------------------------------------
constexpr stream_slot value() const noexcept {
return value_;
}
private:
stream_slot value_;
};
/// Wraps a stream slot ID for outbound paths with the full type information of
/// the path creation.
template <class OutputType, class... HandshakeArgs>
class outbound_stream_slot {
public:
// -- member types -----------------------------------------------------------
/// 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...>;
// -- constructors, destructors, and assignment operators --------------------
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_;
}
// -- properties -------------------------------------------------------------
constexpr stream_slot value() const noexcept {
return value_;
}
private:
stream_slot value_;
};
/// @relates stream_slots
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_slots& x) {
......
......@@ -48,38 +48,32 @@ public:
}
/// Creates a new output path to the current sender.
output_stream<output_type, std::tuple<>, intrusive_ptr<stream_source>>
add_outbound_path() {
outbound_stream_slot<output_type> add_outbound_path() {
CAF_LOG_TRACE("");
return this->add_unchecked_outbound_path<output_type>().rebind(this);
return this->add_unchecked_outbound_path<output_type>();
}
/// Creates a new output path to the current sender with custom handshake.
template <class... Ts>
output_stream<output_type, std::tuple<detail::strip_and_convert_t<Ts>...>,
intrusive_ptr<stream_source>>
outbound_stream_slot<output_type, detail::strip_and_convert_t<Ts>...>
add_outbound_path(std::tuple<Ts...> xs) {
CAF_LOG_TRACE(CAF_ARG(xs));
return this->add_unchecked_outbound_path<output_type>(std::move(xs))
.rebind(this);
return this->add_unchecked_outbound_path<output_type>(std::move(xs));
}
/// Creates a new output path to the current sender.
template <class Handle>
output_stream<output_type, std::tuple<>, intrusive_ptr<stream_source>>
add_outbound_path(const Handle& next) {
outbound_stream_slot<output_type> add_outbound_path(const Handle& next) {
CAF_LOG_TRACE(CAF_ARG(next));
return this->add_unchecked_outbound_path<output_type>(next).rebind(this);
return this->add_unchecked_outbound_path<output_type>(next);
}
/// Creates a new output path to the current sender with custom handshake.
template <class Handle, class... Ts>
output_stream<output_type, std::tuple<detail::strip_and_convert_t<Ts>...>,
intrusive_ptr<stream_source>>
outbound_stream_slot<output_type, detail::strip_and_convert_t<Ts>...>
add_outbound_path(const Handle& next, std::tuple<Ts...> xs) {
CAF_LOG_TRACE(CAF_ARG(next) << CAF_ARG(xs));
return this->add_unchecked_outbound_path<output_type>(next, std::move(xs))
.rebind(this);
return this->add_unchecked_outbound_path<output_type>(next, std::move(xs));
}
protected:
......
......@@ -39,7 +39,7 @@ TESTEE_SETUP();
VARARGS_TESTEE(file_reader, size_t buf_size) {
using buf = std::deque<int>;
return {
[=](string& fname) -> output_stream_t<int, string> {
[=](string& fname) -> output_stream<int, string> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return self->make_source(
......
......@@ -226,14 +226,14 @@ TESTEE(stream_multiplexer) {
auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request for integers");
auto result = stg->add_unchecked_outbound_path<int>();
stg->out().assign<int_downstream_manager>(result.out());
stg->out().assign<int_downstream_manager>(result);
return result;
},
[=](join_atom, strings_atom) {
auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request for strings");
auto result = stg->add_unchecked_outbound_path<string>();
stg->out().assign<string_downstream_manager>(result.out());
stg->out().assign<string_downstream_manager>(result);
return result;
},
[=](const stream<int>& in) {
......
......@@ -258,7 +258,7 @@ public:
auto mgr = detail::make_stream_source<driver>(this, num_messages);
auto res = mgr->add_outbound_path(ref.ctrl());
CAF_MESSAGE(name_ << " starts streaming to " << ref.name()
<< " on slot " << res.out());
<< " on slot " << res.value());
}
void forward_to(entity& ref) {
......@@ -284,7 +284,7 @@ public:
forwarder = detail::make_stream_stage<driver>(this, &data);
auto res = forwarder->add_outbound_path(ref.ctrl());
CAF_MESSAGE(name_ << " starts forwarding to " << ref.name()
<< " on slot " << res.out());
<< " on slot " << res.value());
}
void operator()(open_stream_msg& hs) {
......
......@@ -76,7 +76,7 @@ std::function<void (T&, const error&)> fin(scheduled_actor* self) {
VARARGS_TESTEE(file_reader, size_t buf_size) {
return {
[=](string& fname) -> output_stream_t<int> {
[=](string& fname) -> output_stream<int> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return self->make_source(
......
......@@ -106,7 +106,7 @@ TESTEE(log_producer) {
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.out(), lvl);
out.set_filter(res.outbound_slot(), lvl);
return res;
}
};
......@@ -137,11 +137,11 @@ TESTEE(log_dispatcher) {
auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request");
auto result = stg->add_outbound_path();
stg->out().set_filter(result.out(), lvl);
stg->out().set_filter(result, lvl);
return result;
},
[=](const stream<value_type>& in) {
return self->state.stage->add_inbound_path(in);
self->state.stage->add_inbound_path(in);
}
};
}
......
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