Commit 71d2e3e9 authored by Dominik Charousset's avatar Dominik Charousset

Make composition of stream stages type-safe

parent 4c886d32
......@@ -26,6 +26,7 @@
#include "caf/expected.hpp"
#include "caf/optional.hpp"
#include "caf/replies_to.hpp"
#include "caf/stream_result.hpp"
#include "caf/detail/implicit_conversions.hpp"
......@@ -51,14 +52,21 @@ struct dmi<result<Ys...> (Xs...)> {
output_tuple<implicit_conversions_t<Ys>...>>;
};
// case #2b: function returning a std::tuple<...>
// 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 std::decay<Xs>::type...>,
output_tuple<implicit_conversions_t<Y>>>;
};
// case #2c: function returning a std::tuple<...>
template <class... Ys, class... Xs>
struct dmi<std::tuple<Ys...> (Xs...)> {
using type = typed_mpi<type_list<typename std::decay<Xs>::type...>,
output_tuple<implicit_conversions_t<Ys>...>>;
};
// case #2c: function returning a std::tuple<...>
// case #2d: function returning a std::tuple<...>
template <class... Ys, class... Xs>
struct dmi<delegated<Ys...> (Xs...)> {
using type = typed_mpi<type_list<typename std::decay<Xs>::type...>,
......@@ -80,6 +88,14 @@ struct dmi<optional<Y> (Xs...)> : dmi<Y (Xs...)> {};
template <class Y, class... Xs>
struct dmi<expected<Y> (Xs...)> : dmi<Y (Xs...)> {};
// case #5: function returning an annotated_stream<>
template <class Y, class... Ys, class... Xs>
struct dmi<annotated_stream<Y, Ys...> (Xs...)> : dmi<Y (Xs...)> {
using type =
typed_mpi<type_list<typename std::decay<Xs>::type...>,
output_tuple<stream<Y>, strip_and_convert_t<Ys>...>>;
};
// -- dmfou = deduce_mpi_function_object_unboxing
template <class T, bool isClass = std::is_class<T>::value>
......
......@@ -53,6 +53,10 @@ template <class...> class typed_actor_pointer;
template <class...> class typed_response_promise;
template <class...> class typed_event_based_actor;
// -- variadic templates with 1 fixed argument ---------------------------------
template <class, class...> class annotated_stream;
// -- classes ------------------------------------------------------------------
class actor;
......@@ -183,6 +187,7 @@ using weak_actor_ptr = weak_intrusive_ptr<actor_control_block>;
// -- intrusive pointer aliases ------------------------------------------------
using stream_handler_ptr = intrusive_ptr<stream_handler>;
using strong_actor_ptr = intrusive_ptr<actor_control_block>;
// -- unique pointer aliases ---------------------------------------------------
......
......@@ -34,6 +34,7 @@
#include "caf/extend.hpp"
#include "caf/local_actor.hpp"
#include "caf/actor_marker.hpp"
#include "caf/stream_result.hpp"
#include "caf/response_handle.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/stream_sink_impl.hpp"
......@@ -82,6 +83,9 @@ class scheduled_actor : public local_actor, public resumable {
public:
// -- member types -----------------------------------------------------------
/// A reference-counting pointer to a `stream_handler`.
using stream_handler_ptr = intrusive_ptr<stream_handler>;
/// The message ID of an outstanding response with its callback.
using pending_response = std::pair<const message_id, behavior>;
......@@ -89,7 +93,8 @@ public:
using pointer = scheduled_actor*;
/// Function object for handling unmatched messages.
using default_handler = std::function<result<message> (pointer, message_view&)>;
using default_handler =
std::function<result<message>(pointer, message_view&)>;
/// Function object for handling error messages.
using error_handler = std::function<void (pointer, error&)>;
......@@ -298,12 +303,19 @@ public:
// -- stream management ------------------------------------------------------
/// Owning poiner to a `downstream_policy`.
using downstream_policy_ptr = std::unique_ptr<downstream_policy>;
/// Owning poiner to an `upstream_policy`.
using upstream_policy_ptr = std::unique_ptr<upstream_policy>;
// Starts a new stream.
template <class Handle, class... Ts, class Init, class Getter,
class ClosedPredicate, class ResHandler>
stream<typename stream_source_trait_t<Getter>::output>
annotated_stream<typename stream_source_trait_t<Getter>::output, Ts...>
new_stream(const Handle& dest, std::tuple<Ts...> xs, Init init, Getter getter,
ClosedPredicate pred, ResHandler res_handler) {
ClosedPredicate pred, ResHandler res_handler,
downstream_policy_ptr dpolicy = nullptr) {
using type = typename stream_source_trait_t<Getter>::output;
using state_type = typename stream_source_trait_t<Getter>::state;
static_assert(std::is_same<
......@@ -313,13 +325,15 @@ public:
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
bool (const state_type&),
typename detail::get_callable_trait<ClosedPredicate>::fun_sig
typename detail::get_callable_trait<
ClosedPredicate
>::fun_sig
>::value,
"Expected signature `bool (const State&)` for "
"closed_predicate function");
if (!dest) {
CAF_LOG_ERROR("cannot stream to an invalid actor handle");
return stream_id{nullptr, 0};
return {stream_id{nullptr, 0}, nullptr};
}
// generate new stream ID
stream_id sid{ctrl(),
......@@ -340,12 +354,13 @@ public:
stream_result_trait_t<ResHandler>::make_result_handler(res_handler));
// install stream handler
using impl = stream_source_impl<Getter, ClosedPredicate>;
std::unique_ptr<downstream_policy> p{new policy::anycast};
auto ptr = make_counted<impl>(this, sid, std::move(p), std::move(getter),
std::move(pred));
if (dpolicy == nullptr)
dpolicy.reset(new policy::anycast);
auto ptr = make_counted<impl>(this, sid, std::move(dpolicy),
std::move(getter), std::move(pred));
init(ptr->state());
streams_.emplace(std::move(sid), std::move(ptr));
return sid;
streams_.emplace(sid, ptr);
return {std::move(sid), std::move(ptr)};
}
// Starts a new stream.
......@@ -361,9 +376,9 @@ public:
/// Adds a stream source to this actor.
template <class Init, class... Ts, class Getter, class ClosedPredicate>
stream<typename stream_source_trait_t<Getter>::output>
add_source(std::tuple<Ts...> xs, Init init,
Getter getter, ClosedPredicate pred) {
annotated_stream<typename stream_source_trait_t<Getter>::output, Ts...>
add_source(std::tuple<Ts...> xs, Init init, Getter getter,
ClosedPredicate pred, downstream_policy_ptr dpolicy = nullptr) {
CAF_ASSERT(current_mailbox_element() != nullptr);
using type = typename stream_source_trait_t<Getter>::output;
using state_type = typename stream_source_trait_t<Getter>::state;
......@@ -374,7 +389,9 @@ public:
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
bool (const state_type&),
typename detail::get_callable_trait<ClosedPredicate>::fun_sig
typename detail::get_callable_trait<
ClosedPredicate
>::fun_sig
>::value,
"Expected signature `bool (const State&)` for "
"closed_predicate function");
......@@ -382,18 +399,19 @@ public:
CAF_LOG_ERROR("cannot create a stream data source without downstream");
auto rp = make_response_promise();
rp.deliver(sec::no_downstream_stages_defined);
return stream_id{nullptr, 0};
return {stream_id{nullptr, 0}, nullptr};
}
stream_id sid{ctrl(),
new_request_id(message_priority::normal).integer_value()};
fwd_stream_handshake<type>(sid, xs);
using impl = stream_source_impl<Getter, ClosedPredicate>;
std::unique_ptr<downstream_policy> p{new policy::anycast};
auto ptr = make_counted<impl>(this, sid, std::move(p), std::move(getter),
std::move(pred));
if (dpolicy == nullptr)
dpolicy.reset(new policy::anycast);
auto ptr = make_counted<impl>(this, sid, std::move(dpolicy),
std::move(getter), std::move(pred));
init(ptr->state());
streams_.emplace(std::move(sid), std::move(ptr));
return sid;
streams_.emplace(sid, ptr);
return {std::move(sid), std::move(ptr)};
}
template <class Init, class Getter, class ClosedPredicate>
......@@ -405,9 +423,10 @@ public:
/// Adds a stream stage to this actor.
template <class In, class... Ts, class Init, class Fun, class Cleanup>
stream<typename stream_stage_trait_t<Fun>::output>
add_stage(const stream<In>& in, std::tuple<Ts...> xs,
Init init, Fun fun, Cleanup cleanup) {
annotated_stream<typename stream_stage_trait_t<Fun>::output, Ts...>
add_stage(const stream<In>& in, std::tuple<Ts...> xs, Init init, Fun fun,
Cleanup cleanup, upstream_policy_ptr upolicy = nullptr,
downstream_policy_ptr dpolicy = nullptr) {
CAF_ASSERT(current_mailbox_element() != nullptr);
using output_type = typename stream_stage_trait_t<Fun>::output;
using state_type = typename stream_stage_trait_t<Fun>::state;
......@@ -423,13 +442,16 @@ public:
auto sid = in.id();
fwd_stream_handshake<output_type>(sid, xs);
using impl = stream_stage_impl<Fun, Cleanup>;
std::unique_ptr<downstream_policy> dptr{new policy::anycast};
std::unique_ptr<upstream_policy> uptr{new policy::greedy};
auto ptr = make_counted<impl>(this, sid, std::move(uptr), std::move(dptr),
std::move(fun), std::move(cleanup));
if (upolicy == nullptr)
upolicy.reset(new policy::greedy);
if (dpolicy == nullptr)
dpolicy.reset(new policy::anycast);
auto ptr = make_counted<impl>(this, sid, std::move(upolicy),
std::move(dpolicy), std::move(fun),
std::move(cleanup));
init(ptr->state());
streams_.emplace(sid, std::move(ptr));
return std::move(sid);
streams_.emplace(sid, ptr);
return {std::move(sid), std::move(ptr)};
}
/// Adds a stream stage to this actor.
......@@ -455,10 +477,10 @@ public:
/// Adds a stream sink to this actor.
template <class In, class Init, class Fun, class Finalize>
result<typename stream_sink_trait_t<Fun, Finalize>::output>
add_sink(const stream<In>& in, Init init, Fun fun, Finalize finalize) {
stream_result<typename stream_sink_trait_t<Fun, Finalize>::output>
add_sink(const stream<In>& in, Init init, Fun fun, Finalize finalize,
upstream_policy_ptr upolicy = nullptr) {
CAF_ASSERT(current_mailbox_element() != nullptr);
delegated<typename stream_sink_trait_t<Fun, Finalize>::output> dummy_res;
//using output_type = typename stream_sink_trait_t<Fun, Finalize>::output;
using state_type = typename stream_sink_trait_t<Fun, Finalize>::state;
static_assert(std::is_same<
......@@ -475,20 +497,21 @@ public:
auto mptr = current_mailbox_element();
if (!mptr) {
CAF_LOG_ERROR("add_sink called outside of a message handler");
return dummy_res;
return {stream_id{nullptr, 0}, nullptr};
}
using impl = stream_sink_impl<Fun, Finalize>;
std::unique_ptr<upstream_policy> p{new policy::greedy};
auto ptr = make_counted<impl>(this, std::move(p), std::move(mptr->sender),
if (upolicy == nullptr)
upolicy.reset(new policy::greedy);
auto ptr = make_counted<impl>(this, std::move(upolicy),
std::move(mptr->sender),
std::move(mptr->stages), mptr->mid,
std::move(fun), std::move(finalize));
init(ptr->state());
streams_.emplace(in.id(), std::move(ptr));
return dummy_res;
streams_.emplace(in.id(), ptr);
return {in.id(), std::move(ptr)};
}
inline std::unordered_map<stream_id, intrusive_ptr<stream_handler>>&
streams() {
inline std::unordered_map<stream_id, stream_handler_ptr>& streams() {
return streams_;
}
......@@ -654,7 +677,7 @@ protected:
// TODO: this type is quite heavy in terms of memory, maybe use vector?
/// Holds state for all streams running through this actor.
std::unordered_map<stream_id, intrusive_ptr<stream_handler>> streams_;
std::unordered_map<stream_id, stream_handler_ptr> streams_;
# ifndef CAF_NO_EXCEPTIONS
/// Customization point for setting a default exception callback.
......
......@@ -20,11 +20,14 @@
#ifndef CAF_STREAM_HPP
#define CAF_STREAM_HPP
#include "caf/fwd.hpp"
#include "caf/stream_id.hpp"
#include "caf/stream_handler.hpp"
#include "caf/meta/type_name.hpp"
namespace caf {
/// Marker type for constructing invalid `stream` objects.
struct invalid_stream_t {};
/// Identifies an unbound sequence of messages.
......@@ -41,14 +44,36 @@ public:
// nop
}
/// Convenience constructor for returning the result of `self->new_stream`
/// and similar functions.
stream(stream_id sid, stream_handler_ptr sptr)
: id_(std::move(sid)),
ptr_(std::move(sptr)) {
// nop
}
/// Convenience constructor for returning the result of `self->new_stream`
/// and similar functions.
stream(stream other, stream_handler_ptr sptr)
: id_(std::move(other.id_)),
ptr_(std::move(sptr)) {
// nop
}
stream(invalid_stream_t) {
// nop
}
/// Returns the unique identifier for this stream.
inline const stream_id& id() const {
return id_;
}
/// Returns the handler assigned to this stream on this actor.
inline const stream_handler_ptr& ptr() const {
return ptr_;
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, stream& x) {
return f(meta::type_name("stream"), x.id_);
......@@ -56,10 +81,24 @@ public:
private:
stream_id id_;
stream_handler_ptr ptr_;
};
/// @relates stream
constexpr invalid_stream_t invalid_stream = invalid_stream_t{};
/// Identifies an unbound sequence of messages annotated with the additional
/// handshake arguments emitted to the next stage.
template <class T, class... Ts>
class annotated_stream final : public stream<T> {
public:
/// Dennotes the supertype.
using super = stream<T>;
// Import constructors.
using super::super;
};
} // namespace caf
#endif // CAF_STREAM_HPP
......@@ -49,6 +49,7 @@ struct stream_msg : tag::boxing_type {
/// Identifies content types that propagate errors in both directions.
flows_both_ways
};
/// Initiates a stream handshake.
struct open {
/// Allows visitors to dispatch on this tag.
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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_STREAM_RESULT_HPP
#define CAF_STREAM_RESULT_HPP
#include "caf/fwd.hpp"
#include "caf/stream_id.hpp"
#include "caf/stream_handler.hpp"
#include "caf/meta/type_name.hpp"
namespace caf {
/// Terminates a stream by reducing it to a single value.
template <class T>
class stream_result {
public:
stream_result() = default;
stream_result(stream_result&&) = default;
stream_result(const stream_result&) = default;
stream_result& operator=(stream_result&&) = default;
stream_result& operator=(const stream_result&) = default;
stream_result(stream_id sid) : id_(std::move(sid)) {
// nop
}
/// Convenience constructor for returning the result of `self->new_stream_result`
/// and similar functions.
stream_result(stream_id sid, stream_handler_ptr sptr)
: id_(std::move(sid)),
ptr_(std::move(sptr)) {
// nop
}
/// Convenience constructor for returning the result of `self->new_stream_result`
/// and similar functions.
stream_result(stream_result other, stream_handler_ptr sptr)
: id_(std::move(other.id_)),
ptr_(std::move(sptr)) {
// nop
}
/// Returns the unique identifier for this stream_result.
inline const stream_id& id() const {
return id_;
}
/// Returns the handler assigned to this stream_result on this actor.
inline const stream_handler_ptr& ptr() const {
return ptr_;
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, stream_result& x) {
return f(meta::type_name("stream_result"), x.id_);
}
private:
stream_id id_;
stream_handler_ptr ptr_;
};
} // namespace caf
#endif // CAF_STREAM_RESULT_HPP
......@@ -379,7 +379,7 @@ scheduled_actor::categorize(mailbox_element& x) {
return message_category::internal;
}
case make_type_token<stream_msg>(): {
auto sm = content.move_if_unshared<stream_msg>(0);
auto& sm = content.get_mutable_as<stream_msg>(0);
auto e = streams_.end();
stream_msg_visitor f{this, sm.sid, streams_.find(sm.sid), e};
auto res = apply_visitor(f, sm.content);
......
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