Commit ef999854 authored by Dominik Charousset's avatar Dominik Charousset

Drop redundant template argument of stream sources

parent 25765733
......@@ -35,6 +35,7 @@ template <class> class optional;
template <class> class param;
template <class> class stream;
template <class> class stream_sink;
template <class> class stream_source;
template <class> class trivial_match_case;
template <class> class weak_intrusive_ptr;
......@@ -42,14 +43,13 @@ template <class> struct timeout_definition;
// -- 2 param templates --------------------------------------------------------
template <class, class> class stream_source;
template <class, class> class stream_stage;
// -- 3 param templates --------------------------------------------------------
template <class, class, int> class actor_cast_access;
template <class, class, class> class broadcast_downstream_manager;
template <class, class, class> class stream_stage;
// -- variadic templates -------------------------------------------------------
......
......@@ -35,7 +35,7 @@ struct make_source_result {
using output_type = typename DownstreamManager::output_type;
/// Fully typed stream manager as returned by `make_source`.
using source_type = stream_source<output_type, DownstreamManager>;
using source_type = stream_source<DownstreamManager>;
/// Pointer to a fully typed stream manager.
using source_ptr_type = intrusive_ptr<source_type>;
......
......@@ -37,7 +37,7 @@ public:
using output_type = typename DownstreamManager::output_type;
/// Fully typed stream manager as returned by `make_stage`.
using stage_type = stream_stage<In, output_type, DownstreamManager>;
using stage_type = stream_stage<In, DownstreamManager>;
/// Pointer to a fully typed stream manager.
using stage_ptr_type = intrusive_ptr<stage_type>;
......
......@@ -593,8 +593,7 @@ public:
template <class Init, class Fun, class Cleanup,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
stream_stage_ptr<typename Trait::input, typename Trait::output,
DownstreamManager>
stream_stage_ptr<typename Trait::input, DownstreamManager>
make_continuous_stage(Init init, Fun fun, Cleanup cleanup,
policy::arg<DownstreamManager> token = {}) {
CAF_IGNORE_UNUSED(token);
......
......@@ -30,12 +30,12 @@
namespace caf {
template <class Out, class DownstreamManager>
template <class DownstreamManager>
class stream_source : public virtual stream_manager {
public:
// -- member types -----------------------------------------------------------
using output_type = Out;
using output_type = typename DownstreamManager::output_type;
// -- constructors, destructors, and assignment operators --------------------
......@@ -48,36 +48,37 @@ public:
}
/// Creates a new output path to the current sender.
output_stream<Out, std::tuple<>, intrusive_ptr<stream_source>>
output_stream<output_type, std::tuple<>, intrusive_ptr<stream_source>>
add_outbound_path() {
CAF_LOG_TRACE("");
return this->add_unchecked_outbound_path<Out>().rebind(this);
return this->add_unchecked_outbound_path<output_type>().rebind(this);
}
/// Creates a new output path to the current sender with custom handshake.
template <class... Ts>
output_stream<Out, std::tuple<detail::strip_and_convert_t<Ts>...>,
output_stream<output_type, std::tuple<detail::strip_and_convert_t<Ts>...>,
intrusive_ptr<stream_source>>
add_outbound_path(std::tuple<Ts...> xs) {
CAF_LOG_TRACE(CAF_ARG(xs));
return this->add_unchecked_outbound_path<Out>(std::move(xs)).rebind(this);
return this->add_unchecked_outbound_path<output_type>(std::move(xs))
.rebind(this);
}
/// Creates a new output path to the current sender.
template <class Handle>
output_stream<Out, std::tuple<>, intrusive_ptr<stream_source>>
output_stream<output_type, std::tuple<>, intrusive_ptr<stream_source>>
add_outbound_path(const Handle& next) {
CAF_LOG_TRACE(CAF_ARG(next));
return this->add_unchecked_outbound_path<Out>(next).rebind(this);
return this->add_unchecked_outbound_path<output_type>(next).rebind(this);
}
/// Creates a new output path to the current sender with custom handshake.
template <class Handle, class... Ts>
output_stream<Out, std::tuple<detail::strip_and_convert_t<Ts>...>,
output_stream<output_type, std::tuple<detail::strip_and_convert_t<Ts>...>,
intrusive_ptr<stream_source>>
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<Out>(next, std::move(xs))
return this->add_unchecked_outbound_path<output_type>(next, std::move(xs))
.rebind(this);
}
......@@ -85,12 +86,8 @@ protected:
DownstreamManager out_;
};
template <class Out, class DownstreamManager>
using stream_source_ptr = intrusive_ptr<stream_source<Out, DownstreamManager>>;
template <class DownstreamManager>
using stream_source_ptr_t =
stream_source_ptr<typename DownstreamManager::output_type, DownstreamManager>;
using stream_source_ptr = intrusive_ptr<stream_source<DownstreamManager>>;
} // namespace caf
......
......@@ -43,7 +43,7 @@ public:
using stream_type = stream<output_type>;
/// Implemented `stream_source` interface.
using source_type = stream_source<output_type, downstream_manager_type>;
using source_type = stream_source<downstream_manager_type>;
/// Smart pointer to the interface type.
using source_ptr_type = intrusive_ptr<source_type>;
......
......@@ -28,13 +28,13 @@
namespace caf {
template <class In, class Out, class DownstreamManager>
class stream_stage : public stream_source<Out, DownstreamManager>,
template <class In, class DownstreamManager>
class stream_stage : public stream_source<DownstreamManager>,
public stream_sink<In> {
public:
// -- member types -----------------------------------------------------------
using left_super = stream_source<Out, DownstreamManager>;
using left_super = stream_source<DownstreamManager>;
using right_super = stream_sink<In>;
......@@ -52,9 +52,8 @@ public:
}
};
template <class In, class Out, class DownstreamManager>
using stream_stage_ptr =
intrusive_ptr<stream_stage<In, Out, DownstreamManager>>;
template <class In, class DownstreamManager>
using stream_stage_ptr = intrusive_ptr<stream_stage<In, DownstreamManager>>;
} // namespace caf
......
......@@ -47,7 +47,7 @@ public:
using stream_type = stream<output_type>;
/// Implemented `stream_stage` interface.
using stage_type = stream_stage<input_type, output_type, DownstreamManager>;
using stage_type = stream_stage<input_type, DownstreamManager>;
/// Smart pointer to the interface type.
using stage_ptr_type = intrusive_ptr<stage_type>;
......
......@@ -104,7 +104,7 @@ TESTEE(sum_up) {
}
TESTEE_STATE(stream_multiplexer) {
stream_stage_ptr<int, int, broadcast_downstream_manager<int>> stage;
stream_stage_ptr<int, broadcast_downstream_manager<int>> stage;
};
TESTEE(stream_multiplexer) {
......
......@@ -385,7 +385,7 @@ public:
mboxqueue mbox;
const char* name_;
vector<int> data; // Keeps track of all received data from all batches.
stream_stage_ptr<int, int, broadcast_downstream_manager<int>> forwarder;
stream_stage_ptr<int, broadcast_downstream_manager<int>> forwarder;
tick_type ticks_per_force_batches_interval;
tick_type ticks_per_credit_interval;
......
......@@ -113,7 +113,7 @@ TESTEE(log_producer) {
}
TESTEE_STATE(log_dispatcher) {
stream_stage_ptr<value_type, value_type, downstream_manager> stage;
stream_stage_ptr<value_type, downstream_manager> stage;
};
TESTEE(log_dispatcher) {
......
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