Commit 2e7790f4 authored by Dominik Charousset's avatar Dominik Charousset

Simplify make_(source|sink|stage) implementations

parent 73f9eb8b
...@@ -427,11 +427,7 @@ public: ...@@ -427,11 +427,7 @@ public:
template <class In, class Out, class... Ts> template <class In, class Out, class... Ts>
output_stream<Out, Ts...> output_stream<Out, Ts...>
add_input_path(const stream<In>&, stream_stage_ptr<In, Out, Ts...> mgr) { add_input_path(const stream<In>&, stream_stage_ptr<In, Out, Ts...> mgr) {
auto slot = next_slot(); auto slot = assign_new_slot(mgr);
if (!add_stream_manager(slot, mgr)) {
CAF_LOG_WARNING("unable to assign a manager to its slot");
return {0, nullptr};
}
return {slot, std::move(mgr)}; return {slot, std::move(mgr)};
} }
...@@ -480,13 +476,9 @@ public: ...@@ -480,13 +476,9 @@ public:
template <class Driver, class Input, class... Ts> template <class Driver, class Input, class... Ts>
stream_result<typename Driver::output_type> make_sink(const stream<Input>&, stream_result<typename Driver::output_type> make_sink(const stream<Input>&,
Ts&&... xs) { Ts&&... xs) {
auto slot = next_slot(); auto mgr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...);
auto ptr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...); auto slot = assign_new_slot(mgr);
if (!add_stream_manager(slot, ptr)) { return {slot, std::move(mgr)};
CAF_LOG_WARNING("unable to add a stream manager for a sink");
return {0, nullptr};
}
return {slot, std::move(ptr)};
} }
template <class Input, class Init, class Fun, class Finalize, template <class Input, class Init, class Fun, class Finalize,
...@@ -504,14 +496,11 @@ public: ...@@ -504,14 +496,11 @@ public:
class Input = int, class... Ts> class Input = int, class... Ts>
typename Driver::output_stream_type make_stage(const stream<Input>&, typename Driver::output_stream_type make_stage(const stream<Input>&,
Ts&&... xs) { Ts&&... xs) {
auto slot = next_slot(); using detail::make_stream_stage;
auto ptr = detail::make_stream_stage<Driver, Scatterer>( auto mgr = make_stream_stage<Driver, Scatterer>(this,
this, std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
if (!add_stream_manager(slot, ptr)) { auto slot = assign_new_slot(mgr);
CAF_LOG_WARNING("unable to add a stream manager for a stage"); return {slot, std::move(mgr)};
return {0, nullptr};
}
return {slot, std::move(ptr)};
} }
template <class In, class... Ts, class Init, class Fun, class Cleanup, template <class In, class... Ts, class Init, class Fun, class Cleanup,
...@@ -1195,6 +1184,10 @@ public: ...@@ -1195,6 +1184,10 @@ public:
/// Returns a currently unused slot. /// Returns a currently unused slot.
stream_slot next_slot(); stream_slot next_slot();
/// Assigns a new slot to `ptr`, adds a new entry to `stream_managers_`, and
/// returns the slot ID.
stream_slot assign_new_slot(stream_manager_ptr ptr);
/// Adds a new stream manager to the actor and starts cycle management if /// Adds a new stream manager to the actor and starts cycle management if
/// needed. /// needed.
bool add_stream_manager(stream_slot id, stream_manager_ptr ptr); bool add_stream_manager(stream_slot id, stream_manager_ptr ptr);
......
...@@ -920,6 +920,16 @@ stream_slot scheduled_actor::next_slot() { ...@@ -920,6 +920,16 @@ stream_slot scheduled_actor::next_slot() {
return result; return result;
} }
stream_slot scheduled_actor::assign_new_slot(stream_manager_ptr ptr) {
CAF_LOG_TRACE("");
if (stream_managers_.empty())
stream_ticks_.start(clock().now());
auto slot = next_slot();
CAF_ASSERT(stream_managers_.count(slot) == 0);
stream_managers_.emplace(slot, std::move(ptr));
return slot;
}
bool scheduled_actor::add_stream_manager(stream_slot id, bool scheduled_actor::add_stream_manager(stream_slot id,
stream_manager_ptr ptr) { stream_manager_ptr ptr) {
CAF_LOG_TRACE(CAF_ARG(id)); CAF_LOG_TRACE(CAF_ARG(id));
......
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