Commit e839d34c authored by Dominik Charousset's avatar Dominik Charousset

Propagate typed stream managers from factories

Introduce `stream_source` and `stream_stage` to be able to properly
type-check `output_stream`.
parent 0731e31d
......@@ -148,7 +148,8 @@ public:
/// Calls `(*this)(x.ptr)`.
template <class T, class... Us>
void operator()(output_stream<T, Us...>& x) {
(*this)(x.slot(), x.ptr());
stream_manager_ptr ptr{std::move(x.ptr())};
(*this)(x.slot(), ptr);
}
/// Calls `(*this)(x.ptr)`.
......
......@@ -24,16 +24,14 @@
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/outbound_path.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_source.hpp"
#include "caf/stream_source_trait.hpp"
#include "caf/policy/arg.hpp"
namespace caf {
namespace detail {
template <class Driver, class Scatterer>
class stream_source_impl : public stream_manager {
class stream_source_impl : public Driver::source_type {
public:
// -- static asserts ---------------------------------------------------------
......@@ -43,6 +41,8 @@ public:
// -- member types -----------------------------------------------------------
using super = typename Driver::source_type;
using driver_type = Driver;
using scatterer_type = Scatterer;
......@@ -53,7 +53,7 @@ public:
template <class... Ts>
stream_source_impl(local_actor* self, Ts&&... xs)
: stream_manager(self),
: super(self),
at_end_(false),
driver_(std::forward<Ts>(xs)...),
out_(self) {
......@@ -98,7 +98,8 @@ private:
template <class Driver,
class Scatterer = broadcast_scatterer<typename Driver::output_type>,
class... Ts>
stream_manager_ptr make_stream_source(local_actor* self, Ts&&... xs) {
typename Driver::source_ptr_type make_stream_source(local_actor* self,
Ts&&... xs) {
using impl = stream_source_impl<Driver, Scatterer>;
return make_counted<impl>(self, std::forward<Ts>(xs)...);
}
......
......@@ -25,15 +25,14 @@
#include "caf/outbound_path.hpp"
#include "caf/sec.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_stage.hpp"
#include "caf/stream_stage_trait.hpp"
#include "caf/policy/arg.hpp"
namespace caf {
namespace detail {
template <class Driver, class Scatterer>
class stream_stage_impl : public stream_manager {
class stream_stage_impl : public Driver::stage_type {
public:
// -- static asserts ---------------------------------------------------------
......@@ -43,6 +42,8 @@ public:
// -- member types -----------------------------------------------------------
using super = typename Driver::stage_type;
using driver_type = Driver;
using scatterer_type = Scatterer;
......@@ -55,7 +56,7 @@ public:
template <class... Ts>
stream_stage_impl(local_actor* self, Ts&&... xs)
: stream_manager(self),
: super(self),
driver_(std::forward<Ts>(xs)...),
out_(self) {
// nop
......@@ -101,7 +102,8 @@ private:
template <class Driver,
class Scatterer = broadcast_scatterer<typename Driver::output_type>,
class... Ts>
stream_manager_ptr make_stream_stage(local_actor* self, Ts&&... xs) {
typename Driver::stage_ptr_type make_stream_stage(local_actor* self,
Ts&&... xs) {
using impl = stream_stage_impl<Driver, Scatterer>;
return make_counted<impl>(self, std::forward<Ts>(xs)...);
}
......
......@@ -58,6 +58,11 @@ template <class...> class typed_event_based_actor;
// -- variadic templates with 1 fixed argument ---------------------------------
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 ------------------------------------------------------------------
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_OUTPUT_STREAM_HPP
#define CAF_OUTPUT_STREAM_HPP
#include "caf/fwd.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_source.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 T, class... Ts>
class output_stream {
public:
// -- member types -----------------------------------------------------------
/// Dennotes the supertype.
using super = stream<T>;
/// Type of a single element.
using value_type = T;
/// Smart pointer to a source.
using source_pointer = stream_source_ptr<T, Ts...>;
// -- constructors, destructors, and assignment operators --------------------
output_stream(stream_slot id, source_pointer sptr)
: slot_(id),
ptr_(std::move(sptr)) {
// nop
}
// -- properties -------------------------------------------------------------
/// Returns the actor-specific stream slot ID.
inline stream_slot slot() const {
return slot_;
}
/// Returns the handler assigned to this stream on this actor.
inline source_pointer& ptr() noexcept {
return ptr_;
}
/// Returns the handler assigned to this stream on this actor.
inline const source_pointer& 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.slot_);
}
private:
// -- member variables -------------------------------------------------------
stream_slot slot_;
source_pointer ptr_;
};
} // namespace caf
#endif // CAF_OUTPUT_STREAM_HPP
......@@ -40,9 +40,11 @@
#include "caf/local_actor.hpp"
#include "caf/logger.hpp"
#include "caf/no_stages.hpp"
#include "caf/output_stream.hpp"
#include "caf/response_handle.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/sec.hpp"
#include "caf/stream.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_result.hpp"
#include "caf/stream_result_trait.hpp"
......
......@@ -20,22 +20,20 @@
#define CAF_STREAM_HPP
#include "caf/fwd.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_slot.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.
/// Identifies an unbound sequence of elements.
template <class T>
class stream {
public:
// -- member types -----------------------------------------------------------
/// Type of a single element.
using value_type = T;
// -- constructors and destructors -------------------------------------------
......@@ -49,10 +47,6 @@ public:
// nop
}
stream(invalid_stream_t) : stream(none) {
// nop
}
/// Convenience constructor for returning the result of `self->new_stream`
/// and similar functions.
stream(stream_slot id, stream_manager_ptr sptr = nullptr)
......@@ -69,7 +63,7 @@ public:
// nop
}
// -- accessors --------------------------------------------------------------
// -- properties -------------------------------------------------------------
/// Returns the actor-specific stream slot ID.
inline stream_slot slot() const {
......@@ -100,21 +94,6 @@ private:
stream_manager_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 output_stream final : public stream<T> {
public:
/// Dennotes the supertype.
using super = stream<T>;
// Import constructors.
using super::super;
};
} // namespace caf
#endif // CAF_STREAM_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_DETAIL_STREAM_SOURCE_HPP
#define CAF_DETAIL_STREAM_SOURCE_HPP
#include <tuple>
#include "caf/intrusive_ptr.hpp"
#include "caf/stream_manager.hpp"
namespace caf {
template <class Out, class... Ts>
class stream_source : public stream_manager {
public:
// -- member types -----------------------------------------------------------
using output_type = Out;
// -- constructors, destructors, and assignment operators --------------------
stream_source(local_actor* self) : stream_manager(self) {
// nop
}
};
template <class Out, class... HandshakeData>
using stream_source_ptr = intrusive_ptr<stream_source<Out, HandshakeData...>>;
} // namespace caf
#endif // CAF_DETAIL_STREAM_SOURCE_HPP
......@@ -40,6 +40,10 @@ public:
using handshake_tuple_type = std::tuple<stream_type, HandshakeData...>;
using source_type = stream_source<output_type, HandshakeData...>;
using source_ptr_type = intrusive_ptr<source_type>;
// -- constructors, destructors, and assignment operators --------------------
virtual ~stream_source_driver() {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_DETAIL_STREAM_STAGE_HPP
#define CAF_DETAIL_STREAM_STAGE_HPP
#include <tuple>
#include "caf/intrusive_ptr.hpp"
#include "caf/stream_source.hpp"
namespace caf {
template <class In, class Out, class... HandshakeData>
class stream_stage : public stream_source<Out, HandshakeData...> {
public:
// -- member types -----------------------------------------------------------
using super = stream_source<Out, HandshakeData...>;
using input_type = In;
// -- constructors, destructors, and assignment operators --------------------
stream_stage(local_actor* self) : super(self) {
// nop
}
};
template <class In, class Out, class... HandshakeData>
using stream_stage_ptr = intrusive_ptr<stream_stage<In, Out, HandshakeData...>>;
} // namespace caf
#endif // CAF_DETAIL_STREAM_STAGE_HPP
......@@ -43,6 +43,10 @@ public:
using handshake_tuple_type = std::tuple<stream_type, HandshakeData...>;
using stage_type = stream_stage<input_type, output_type, HandshakeData...>;
using stage_ptr_type = intrusive_ptr<stage_type>;
// -- constructors, destructors, and assignment operators --------------------
virtual ~stream_stage_driver() {
......
......@@ -150,7 +150,7 @@ TESTEE(broken_sink) {
TESTEE(filter) {
CAF_IGNORE_UNUSED(self);
return {
[=](stream<int>& in, std::string& fname) -> stream<int> {
[=](stream<int>& in, std::string& fname) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->make_stage(
// input stream
......
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