Commit bcacc6d8 authored by Dominik Charousset's avatar Dominik Charousset

Remove make_sink_result

parent 1b98e243
...@@ -152,13 +152,6 @@ public: ...@@ -152,13 +152,6 @@ public:
(*this)(x.in(), x.out(), 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())`. /// Calls `(*this)(x.in(), x.out(), x.ptr())`.
template <class T, class S, class... Ts> template <class T, class S, class... Ts>
void operator()(make_source_result<T, S, Ts...>& x) { void operator()(make_source_result<T, S, Ts...>& x) {
...@@ -174,9 +167,10 @@ public: ...@@ -174,9 +167,10 @@ public:
} }
/// Calls `(*this)(x.ptr)`. /// Calls `(*this)(x.ptr)`.
template <class T> template <class T, class P>
void operator()(stream_result<T>& x) { void operator()(stream_result<T, P>& x) {
(*this)(x.in(), 0, x.ptr()); stream_manager_ptr ptr{std::move(x.ptr())};
(*this)(x.in(), 0, ptr);
} }
// -- visit API: return true if T was visited, false if T was skipped -------- // -- visit API: return true if T was visited, false if T was skipped --------
......
...@@ -33,7 +33,6 @@ template <class> class intrusive_ptr; ...@@ -33,7 +33,6 @@ template <class> class intrusive_ptr;
template <class> class optional; template <class> class optional;
template <class> class param; template <class> class param;
template <class> class stream; 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;
...@@ -229,6 +228,10 @@ using stream_manager_ptr = intrusive_ptr<stream_manager>; ...@@ -229,6 +228,10 @@ using stream_manager_ptr = intrusive_ptr<stream_manager>;
using type_erased_value_ptr = std::unique_ptr<type_erased_value>; using type_erased_value_ptr = std::unique_ptr<type_erased_value>;
using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>; using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
// -- templates that depend on others ------------------------------------------
template <class, class = stream_manager_ptr> class stream_result;
} // namespace caf } // namespace caf
#endif // CAF_FWD_HPP #endif // CAF_FWD_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://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
...@@ -39,7 +39,6 @@ ...@@ -39,7 +39,6 @@
#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_source_result.hpp"
#include "caf/make_stage_result.hpp" #include "caf/make_stage_result.hpp"
#include "caf/no_stages.hpp" #include "caf/no_stages.hpp"
...@@ -526,19 +525,19 @@ public: ...@@ -526,19 +525,19 @@ public:
scatterer_arg); scatterer_arg);
} }
template <class Driver, class Input, class... Ts> template <class Driver, class... Ts>
stream_result<typename Driver::result_type> stream_result<typename Driver::result_type, typename Driver::sink_ptr_type>
make_sink(const stream<Input>& src, Ts&&... xs) { make_sink(const stream<typename Driver::input_type>& src, 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)...);
return mgr->add_inbound_path(src); return mgr->add_inbound_path(src);
} }
template <class Input, class Init, class Fun, class Finalize, template <class Input, class Init, class Fun, class Finalize,
class Trait = stream_sink_trait_t<Fun, Finalize>> class Trait = stream_sink_trait_t<Fun, Finalize>>
stream_result<typename Trait::output> stream_result<typename Trait::result, typename Trait::pointer>
make_sink(const stream<Input>& in, Init init, Fun fun, Finalize fin) { make_sink(const stream<Input>& in, Init init, Fun fun, Finalize fin) {
using driver = detail::stream_sink_driver_impl<typename Trait::input, using driver = detail::stream_sink_driver_impl<typename Trait::input,
typename Trait::output, typename Trait::result,
Fun, Finalize>; Fun, Finalize>;
return make_sink<driver>(in, std::move(init), std::move(fun), return make_sink<driver>(in, std::move(init), std::move(fun),
std::move(fin)); std::move(fin));
......
...@@ -20,7 +20,6 @@ ...@@ -20,7 +20,6 @@
#define CAF_STREAM_RESULT_HPP #define CAF_STREAM_RESULT_HPP
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/make_sink_result.hpp"
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
...@@ -30,9 +29,15 @@ ...@@ -30,9 +29,15 @@
namespace caf { namespace caf {
/// Terminates a stream by reducing it to a single value. /// Terminates a stream by reducing it to a single value.
template <class T> template <class T, class Pointer /* = stream_manager_ptr */>
class stream_result { class stream_result {
public: public:
// -- member types -----------------------------------------------------------
using pointer_type = Pointer;
// -- constructors, destructors, and assignment operators --------------------
stream_result(stream_result&&) = default; stream_result(stream_result&&) = default;
stream_result(const stream_result&) = default; stream_result(const stream_result&) = default;
stream_result& operator=(stream_result&&) = default; stream_result& operator=(stream_result&&) = default;
...@@ -42,48 +47,45 @@ public: ...@@ -42,48 +47,45 @@ public:
// nop // nop
} }
stream_result(stream_slot id) : in_(id) { stream_result(stream_slot id, Pointer mgr = nullptr)
// nop
}
/// Convenience constructor for returning the result of `self->new_stream_result`
/// and similar functions.
stream_result(stream_slot id, stream_manager_ptr sptr)
: in_(id), : in_(id),
ptr_(std::move(sptr)) { ptr_(std::move(mgr)) {
// nop // nop
} }
/// Convenience constructor for returning the result of `self->new_stream_result` template <class CompatiblePointer>
/// and similar functions. stream_result(stream_result<T, CompatiblePointer> other)
stream_result(stream_result other, stream_manager_ptr sptr) : in_(other.in()),
: in_(other.in_), ptr_(std::move(other.ptr())) {
ptr_(std::move(sptr)) {
// nop // nop
} }
template <class In> template <class CompatiblePointer>
stream_result(make_sink_result<In, T> other) stream_result& operator=(stream_result<T, CompatiblePointer> other) {
: in_(other.in()), in_ = other.in();
ptr_(other.ptr()) { ptr_ = std::move(other.ptr());
// nop return *this;
} }
// -- properties -------------------------------------------------------------
/// Returns the unique identifier for this stream_result. /// Returns the unique identifier for this stream_result.
inline stream_slot in() const noexcept { inline stream_slot in() const noexcept {
return in_; return in_;
} }
/// Returns the handler assigned to this stream on this actor. /// Returns the handler assigned to this stream on this actor.
inline stream_manager_ptr& ptr() noexcept { inline pointer_type& 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 stream_manager_ptr& ptr() const noexcept { inline const pointer_type& ptr() const noexcept {
return ptr_; return ptr_;
} }
// -- serialization support --------------------------------------------------
template <class Inspector> template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, friend typename Inspector::result_type inspect(Inspector& f,
stream_result& x) { stream_result& x) {
...@@ -91,8 +93,10 @@ public: ...@@ -91,8 +93,10 @@ public:
} }
private: private:
// -- member variables -------------------------------------------------------
stream_slot in_; stream_slot in_;
stream_manager_ptr ptr_; pointer_type ptr_;
}; };
} // namespace caf } // namespace caf
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_result.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
...@@ -35,6 +36,8 @@ public: ...@@ -35,6 +36,8 @@ public:
using input_type = In; using input_type = In;
using result_type = Result;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_sink(scheduled_actor* self) : stream_manager(self) { stream_sink(scheduled_actor* self) : stream_manager(self) {
...@@ -44,7 +47,10 @@ public: ...@@ -44,7 +47,10 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Creates a new input path to the current sender. /// Creates a new input path to the current sender.
make_sink_result<In, Result> add_inbound_path(const stream<In>& in); stream_result<result_type, intrusive_ptr<stream_sink>>
add_inbound_path(const stream<input_type>&) {
return {this->assign_next_slot(), this};
}
}; };
template <class In, class Result> template <class In, class Result>
...@@ -52,16 +58,4 @@ using stream_sink_ptr = intrusive_ptr<stream_sink<In, Result>>; ...@@ -52,16 +58,4 @@ using stream_sink_ptr = intrusive_ptr<stream_sink<In, Result>>;
} // namespace caf } // namespace caf
#include "caf/make_sink_result.hpp"
namespace caf {
template <class In, class Result>
make_sink_result<In, Result>
stream_sink<In, Result>::add_inbound_path(const stream<In>&) {
return {this->assign_next_slot(), this};
}
} // namespace caf
#endif // CAF_DETAIL_STREAM_SINK_HPP #endif // CAF_DETAIL_STREAM_SINK_HPP
...@@ -24,8 +24,9 @@ ...@@ -24,8 +24,9 @@
#include <vector> #include <vector>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/message.hpp"
#include "caf/make_message.hpp" #include "caf/make_message.hpp"
#include "caf/message.hpp"
#include "caf/stream_sink.hpp"
namespace caf { namespace caf {
...@@ -45,9 +46,6 @@ public: ...@@ -45,9 +46,6 @@ public:
/// Smart pointer to the interface type. /// Smart pointer to the interface type.
using sink_ptr_type = intrusive_ptr<sink_type>; using sink_ptr_type = intrusive_ptr<sink_type>;
/// Wrapper type holding a pointer to `sink_type`.
using make_sink_result_type = make_sink_result<input_type, result_type>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
virtual ~stream_sink_driver() { virtual ~stream_sink_driver() {
......
...@@ -19,8 +19,9 @@ ...@@ -19,8 +19,9 @@
#ifndef CAF_STREAM_SINK_TRAIT_HPP #ifndef CAF_STREAM_SINK_TRAIT_HPP
#define CAF_STREAM_SINK_TRAIT_HPP #define CAF_STREAM_SINK_TRAIT_HPP
#include "caf/message.hpp"
#include "caf/make_message.hpp" #include "caf/make_message.hpp"
#include "caf/message.hpp"
#include "caf/stream_sink.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
...@@ -67,7 +68,7 @@ struct stream_sink_trait_void_finalize { ...@@ -67,7 +68,7 @@ struct stream_sink_trait_void_finalize {
// -- trait implementation ----------------------------------------------------- // -- trait implementation -----------------------------------------------------
/// Base type for all sink traits. /// Base type for all sink traits.
template <class State, class In, class Out> template <class State, class In, class Result>
struct stream_sink_trait_base { struct stream_sink_trait_base {
/// Defines the state element for the function objects. /// Defines the state element for the function objects.
using state = State; using state = State;
...@@ -76,7 +77,10 @@ struct stream_sink_trait_base { ...@@ -76,7 +77,10 @@ struct stream_sink_trait_base {
using input = In; using input = In;
/// Defines the result type of the sink. /// Defines the result type of the sink.
using output = Out; using result = Result;
/// Defines a pointer to a sink.
using pointer = stream_sink_ptr<input, result>;
}; };
/// Defines required type aliases for stream sinks. /// Defines required type aliases for stream sinks.
......
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