Commit c693c48d authored by Dominik Charousset's avatar Dominik Charousset

Make make_source_result a utility trait

parent 7a1bc768
......@@ -76,46 +76,49 @@ struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>,
// case #2
template <class... In, class... Out, class... Xs, class Ys,
class... MapsTo, class... Zs, class... Rs>
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<MapsTo...>>,
output_stream_t<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<MapsTo...>>>> {};
output_stream_t<M, Ms...>>>> {
};
// case #3
template <class... In, class... Out, class... Xs, class Ys,
class... MapsTo, class... Zs, class... Rs>
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<Out...>>, Xs...>,
output_stream_t<O, Out...>>, Xs...>,
Ys,
detail::type_list<typed_mpi<detail::type_list<Out...>,
output_tuple<MapsTo...>>, Zs...>,
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<MapsTo...>>>> {};
output_stream_t<M, Ms...>>>> {
};
// case #4
template <class... In, class... Out, class... Xs, class Ys,
class... MapsTo, class... Zs, class... Rs>
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<Out...>>, Xs...>,
output_stream_t<O, Out...>>, Xs...>,
Ys,
detail::type_list<typed_mpi<detail::type_list<Out...>,
output_stream<MapsTo...>>, Zs...>,
detail::type_list<typed_mpi<detail::type_list<O, Out...>,
output_stream_t<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<MapsTo...>>>> {};
output_stream_t<M, Ms...>>>> {
};
// default case (recurse over Zs)
template <class In, class Out, class... Xs, class Ys,
......
......@@ -89,8 +89,8 @@ 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... Xs>
struct dmi<output_stream<Y, Ys...> (Xs...)> : dmi<Y (Xs...)> {
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>...>>;
......
......@@ -61,9 +61,8 @@ public:
/// Called if the message handler returned any "ordinary" value.
virtual void operator()(message&) = 0;
/// Called if the message handler returned a `stream<...>`,
/// `output_stream<...>`, or `stream_result<...>`, `make_source_result<...>`,
/// `make_sink_result<...>`, or `make_stage_result<...>`.
/// Called if the message handler returned an `output_stream<...>` or a
/// `stream_result<...>`.
virtual void operator()(stream_slot in, stream_slot out,
stream_manager_ptr& mgr) = 0;
......@@ -146,19 +145,12 @@ public:
}
/// Calls `(*this)(x.in(), x.out(), x.ptr())`.
template <class Out, class... Ts>
void operator()(output_stream<Out, Ts...>& x) {
template <class Out, class Tuple, class P>
void operator()(output_stream<Out, Tuple, P>& 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 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) {
......@@ -166,7 +158,7 @@ public:
(*this)(x.in(), x.out(), ptr);
}
/// Calls `(*this)(x.ptr)`.
/// Calls `(*this)(x.in(), 0, x.ptr())`.
template <class T, class P>
void operator()(stream_result<T, P>& x) {
stream_manager_ptr ptr{std::move(x.ptr())};
......
......@@ -19,8 +19,9 @@
#ifndef CAF_FWD_HPP
#define CAF_FWD_HPP
#include <memory>
#include <cstdint>
#include <memory>
#include <tuple>
namespace caf {
......@@ -63,9 +64,7 @@ template <class...> class typed_event_based_actor;
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;
// -- classes ------------------------------------------------------------------
......@@ -228,9 +227,13 @@ 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 that depend on others ------------------------------------------
// -- templates with default parameters ----------------------------------------
template <class, class = std::tuple<>, class = stream_manager_ptr>
class output_stream;
template <class, class = stream_manager_ptr> class stream_result;
template <class, class = stream_manager_ptr>
class stream_result;
} // namespace caf
......
......@@ -29,71 +29,28 @@
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;
/// Helper trait for deducing an `output_stream` from the arguments to
/// `scheduled_actor::make_source`.
template <class Scatterer, class... Ts>
struct make_source_result {
/// Element type.
using value_type = typename Scatterer::value_type;
/// Fully typed stream manager as returned by `make_source`.
using type = stream_source<T, Scatterer, Ts...>;
using source_type = stream_source<value_type, 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 -------------------------------------------------------
using source_ptr_type = intrusive_ptr<source_type>;
stream_slot out_;
ptr_type ptr_;
/// The return type for `scheduled_actor::make_source`.
using type = output_stream<value_type, std::tuple<Ts...>, source_ptr_type>;
};
/// Helper type for defining a `make_source_result` from a `Scatterer` plus
/// additional handshake types.
/// Helper type for defining an `output_stream` from a `Scatterer` plus
/// the types of the handshake arguments.
template <class Scatterer, class... Ts>
using make_source_result_t =
make_source_result<typename Scatterer::value_type, Scatterer,
detail::decay_t<Ts>...>;
typename make_source_result<Scatterer, detail::decay_t<Ts>...>::type;
} // namespace caf
......
......@@ -20,9 +20,8 @@
#define CAF_OUTPUT_STREAM_HPP
#include "caf/fwd.hpp"
#include "caf/make_source_result.hpp"
#include "caf/stream_slot.hpp"
#include "caf/invalid_stream.hpp"
#include "caf/stream_slot.hpp"
#include "caf/meta/type_name.hpp"
......@@ -30,79 +29,90 @@ 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 output_stream {
template <class T, class... Ts, class Pointer /* = stream_manager_ptr */>
class output_stream<T, std::tuple<Ts...>, Pointer> {
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
}
template <class S>
output_stream(make_source_result<T, S, Ts...> x)
: in_(0),
out_(x.out()),
ptr_(std::move(x.ptr())) {
// nop
}
output_stream(stream_slot in_slot, stream_slot out_slot,
stream_manager_ptr mgr)
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 {
inline stream_slot in() const noexcept {
return in_;
}
/// Returns the output slot.
inline stream_slot out() const {
inline stream_slot out() const noexcept {
return out_;
}
/// Returns the handler assigned to this stream on this actor.
inline stream_manager_ptr& ptr() noexcept {
inline pointer_type& ptr() noexcept {
return ptr_;
}
/// Returns the handler assigned to this stream on this actor.
inline const stream_manager_ptr& ptr() const noexcept {
inline const pointer_type& ptr() const noexcept {
return ptr_;
}
// -- serialization support --------------------------------------------------
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
output_stream& x) {
return f(meta::type_name("output_stream"), x.in_, x.out_);
}
private:
// -- member variables -------------------------------------------------------
stream_slot in_;
stream_slot out_;
stream_manager_ptr ptr_;
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
......@@ -51,7 +51,7 @@ struct replies_to {
/// @private
template <class O, class... Os>
using with_stream = typed_mpi<detail::type_list<Is...>,
output_stream<O, Os...>>;
output_stream_t<O, Os...>>;
};
template <class... Is>
......
......@@ -39,7 +39,6 @@
#include "caf/invoke_message_result.hpp"
#include "caf/local_actor.hpp"
#include "caf/logger.hpp"
#include "caf/make_source_result.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/no_stages.hpp"
#include "caf/output_stream.hpp"
......@@ -422,7 +421,7 @@ public:
/// @param scatterer_type Configures the policy for downstream communication.
/// @returns A stream object with a pointer to the generated `stream_manager`.
template <class Driver, class... Ts>
typename Driver::make_source_result_type make_source(Ts&&... xs) {
typename Driver::output_stream_type make_source(Ts&&... xs) {
using detail::make_stream_source;
auto mgr = make_stream_source<Driver>(this, std::forward<Ts>(xs)...);
return mgr->add_outbound_path();
......@@ -483,8 +482,9 @@ public:
class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<detail::is_actor_handle<ActorHandle>::value,
typename make_source_result_t<Scatterer, Ts...>::ptr_type>
detail::enable_if_t<
detail::is_actor_handle<ActorHandle>::value,
typename make_source_result_t<Scatterer, Ts...>::pointer_type>
make_source(const ActorHandle& dest, std::tuple<Ts...> xs, Init init,
Pull pull, Done done, HandleResult handle_res,
policy::arg<Scatterer> scatterer_arg = {}) {
......@@ -516,7 +516,7 @@ public:
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<detail::is_actor_handle<ActorHandle>::value,
typename make_source_result_t<Scatterer>::ptr_type>
typename make_source_result_t<Scatterer>::pointer_type>
make_source(const ActorHandle& dest, Init init, Pull pull, Done done,
HandleResult handle_res,
policy::arg<Scatterer> scatterer_arg = {}) {
......
......@@ -84,14 +84,6 @@ public:
return ptr_;
}
// -- serialization support --------------------------------------------------
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
stream_result& x) {
return f(meta::type_name("stream_result"), x.in_);
}
private:
// -- member variables -------------------------------------------------------
......
......@@ -47,7 +47,10 @@ public:
}
/// Creates a new output path to the current sender.
make_source_result<Out, Scatterer, Ts...> add_outbound_path();
output_stream<Out, std::tuple<Ts...>, intrusive_ptr<stream_source>>
add_outbound_path() {
return {0, this->assign_next_pending_slot(), this};
}
protected:
Scatterer out_;
......@@ -64,16 +67,4 @@ using stream_source_ptr_t =
} // namespace caf
#include "caf/make_source_result.hpp"
namespace caf {
template <class Out, class Scatterer, class... Ts>
make_source_result<Out, Scatterer, Ts...>
stream_source<Out, Scatterer, Ts...>::add_outbound_path() {
return {this->assign_next_pending_slot(), this};
}
} // namespace caf
#endif // CAF_DETAIL_STREAM_SOURCE_HPP
......@@ -41,9 +41,6 @@ public:
/// Type of the output stream.
using stream_type = stream<output_type>;
/// Type of the output stream including handshake argument types.
using output_stream_type = output_stream<output_type, Ts...>;
/// Tuple for creating the `open_stream_msg` handshake.
using handshake_tuple_type = std::tuple<stream_type, Ts...>;
......@@ -53,9 +50,9 @@ public:
/// Smart pointer to the interface 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...>;
/// Type of the output stream including handshake argument types.
using output_stream_type = output_stream<output_type, std::tuple<Ts...>,
source_ptr_type>;
// -- constructors, destructors, and assignment operators --------------------
......
......@@ -39,7 +39,7 @@ TESTEE_SETUP();
VARARGS_TESTEE(file_reader, size_t buf_size) {
using buf = std::deque<int>;
return {
[=](string& fname) -> output_stream<int, string> {
[=](string& fname) -> output_stream_t<int, string> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return self->make_source(
......
......@@ -65,7 +65,7 @@ std::function<bool(const buf&)> is_done(scheduled_actor* self) {
VARARGS_TESTEE(file_reader, size_t buf_size) {
return {
[=](string& fname) -> output_stream<int, string> {
[=](string& fname) -> output_stream_t<int, string> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return self->make_source(
......
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