Commit b0deb598 authored by Dominik Charousset's avatar Dominik Charousset

Rename add_{unsafe => unchecked}_(...)_path

parent 80ad81ec
...@@ -56,10 +56,6 @@ public: ...@@ -56,10 +56,6 @@ public:
return out_; return out_;
} }
bool done() const override {
return !this->continuous() && this->inbound_paths_.empty();
}
void handle(inbound_path*, downstream_msg::batch& x) override { void handle(inbound_path*, downstream_msg::batch& x) override {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
using vec_type = std::vector<input_type>; using vec_type = std::vector<input_type>;
......
...@@ -485,7 +485,7 @@ public: ...@@ -485,7 +485,7 @@ public:
std::move(fin)); std::move(fin));
auto tk = std::make_tuple(stream<output_type>{}); auto tk = std::make_tuple(stream<output_type>{});
auto handshake = make_message_from_tuple(std::tuple_cat(tk, std::move(xs))); auto handshake = make_message_from_tuple(std::tuple_cat(tk, std::move(xs)));
auto slot = mgr->add_unsafe_outbound_path_impl( auto slot = mgr->add_unchecked_outbound_path_impl(
actor_cast<strong_actor_ptr>(dest), std::move(handshake)); actor_cast<strong_actor_ptr>(dest), std::move(handshake));
if (slot == invalid_stream_slot) { if (slot == invalid_stream_slot) {
CAF_LOG_WARNING("unable to assign a slot in make_source"); CAF_LOG_WARNING("unable to assign a slot in make_source");
......
...@@ -152,9 +152,9 @@ public: ...@@ -152,9 +152,9 @@ public:
/// @pre `out().terminal() == false` /// @pre `out().terminal() == false`
/// @private /// @private
template <class Out> template <class Out>
output_stream_t<Out> add_unsafe_outbound_path() { output_stream_t<Out> add_unchecked_outbound_path() {
auto handshake = make_message(stream<Out>{}); auto handshake = make_message(stream<Out>{});
return {0, add_unsafe_outbound_path_impl(std::move(handshake)), this}; return {0, add_unchecked_outbound_path_impl(std::move(handshake)), this};
} }
/// Creates an outbound path to the current sender without any type checking. /// Creates an outbound path to the current sender without any type checking.
...@@ -162,10 +162,10 @@ public: ...@@ -162,10 +162,10 @@ public:
/// @private /// @private
template <class Out, class... Ts> template <class Out, class... Ts>
output_stream_t<Out, detail::strip_and_convert_t<Ts>...> output_stream_t<Out, detail::strip_and_convert_t<Ts>...>
add_unsafe_outbound_path(std::tuple<Ts...> xs) { add_unchecked_outbound_path(std::tuple<Ts...> xs) {
auto tk = std::make_tuple(stream<Out>{}); auto tk = std::make_tuple(stream<Out>{});
auto handshake = make_message_from_tuple(std::tuple_cat(tk, std::move(xs))); auto handshake = make_message_from_tuple(std::tuple_cat(tk, std::move(xs)));
return {0, add_unsafe_outbound_path_impl(std::move(handshake)), this}; return {0, add_unchecked_outbound_path_impl(std::move(handshake)), this};
} }
/// Creates an outbound path to `next`, only checking whether the interface /// Creates an outbound path to `next`, only checking whether the interface
...@@ -175,11 +175,11 @@ public: ...@@ -175,11 +175,11 @@ public:
/// @pre `out().terminal() == false` /// @pre `out().terminal() == false`
/// @private /// @private
template <class Out, class Handle> template <class Out, class Handle>
output_stream_t<Out> add_unsafe_outbound_path(Handle next) { output_stream_t<Out> add_unchecked_outbound_path(Handle next) {
// TODO: type checking // TODO: type checking
auto handshake = make_message(stream<Out>{}); auto handshake = make_message(stream<Out>{});
auto hdl = actor_cast<strong_actor_ptr>(std::move(next)); auto hdl = actor_cast<strong_actor_ptr>(std::move(next));
auto slot = add_unsafe_outbound_path_impl(std::move(hdl), auto slot = add_unchecked_outbound_path_impl(std::move(hdl),
std::move(handshake)); std::move(handshake));
return {0, slot, this}; return {0, slot, this};
} }
...@@ -192,12 +192,12 @@ public: ...@@ -192,12 +192,12 @@ public:
/// @private /// @private
template <class Out, class Handle, class... Ts> template <class Out, class Handle, class... Ts>
output_stream_t<Out, detail::strip_and_convert_t<Ts>...> output_stream_t<Out, detail::strip_and_convert_t<Ts>...>
add_unsafe_outbound_path(const Handle& next, std::tuple<Ts...> xs) { add_unchecked_outbound_path(const Handle& next, std::tuple<Ts...> xs) {
// TODO: type checking // TODO: type checking
auto tk = std::make_tuple(stream<Out>{}); auto tk = std::make_tuple(stream<Out>{});
auto handshake = make_message_from_tuple(std::tuple_cat(tk, std::move(xs))); auto handshake = make_message_from_tuple(std::tuple_cat(tk, std::move(xs)));
auto hdl = actor_cast<strong_actor_ptr>(std::move(next)); auto hdl = actor_cast<strong_actor_ptr>(std::move(next));
auto slot = add_unsafe_outbound_path_impl(std::move(hdl), auto slot = add_unchecked_outbound_path_impl(std::move(hdl),
std::move(handshake)); std::move(handshake));
return {0, slot, this}; return {0, slot, this};
} }
...@@ -207,27 +207,27 @@ public: ...@@ -207,27 +207,27 @@ public:
/// @pre `out().terminal() == false` /// @pre `out().terminal() == false`
/// @private /// @private
template <class In> template <class In>
stream_result<> add_unsafe_inbound_path(const stream<In>&) { stream_result<> add_unchecked_inbound_path(const stream<In>&) {
return {add_unsafe_inbound_path_impl(), this}; return {add_unchecked_inbound_path_impl(), this};
} }
/// Adds a new outbound path to `rp.next()`. /// Adds a new outbound path to `rp.next()`.
/// @private /// @private
stream_slot add_unsafe_outbound_path_impl(response_promise& rp, stream_slot add_unchecked_outbound_path_impl(response_promise& rp,
message handshake); message handshake);
/// Adds a new outbound path to `next`. /// Adds a new outbound path to `next`.
/// @private /// @private
stream_slot add_unsafe_outbound_path_impl(strong_actor_ptr next, stream_slot add_unchecked_outbound_path_impl(strong_actor_ptr next,
message handshake); message handshake);
/// Calls `add_unsafe_outbound_path_impl(make_response_promise(), handshake)`. /// Calls `add_unchecked_outbound_path_impl(make_response_promise(), handshake)`.
/// @private /// @private
stream_slot add_unsafe_outbound_path_impl(message handshake); stream_slot add_unchecked_outbound_path_impl(message handshake);
/// Adds the current sender as an inbound path. /// Adds the current sender as an inbound path.
/// @pre Current message is an `open_stream_msg`. /// @pre Current message is an `open_stream_msg`.
stream_slot add_unsafe_inbound_path_impl(); stream_slot add_unchecked_inbound_path_impl();
protected: protected:
// -- modifiers for self ----------------------------------------------------- // -- modifiers for self -----------------------------------------------------
......
...@@ -42,12 +42,18 @@ public: ...@@ -42,12 +42,18 @@ public:
// nop // nop
} }
// -- overridden member functions --------------------------------------------
bool done() const override {
return !this->continuous() && this->inbound_paths_.empty();
}
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Creates a new input path to the current sender. /// Creates a new input path to the current sender.
stream_result<intrusive_ptr<stream_sink>> stream_result<intrusive_ptr<stream_sink>>
add_inbound_path(const stream<input_type>&) { add_inbound_path(const stream<input_type>&) {
return {add_unsafe_inbound_path_impl(), this}; return {add_unchecked_inbound_path_impl(), this};
} }
}; };
......
...@@ -51,7 +51,7 @@ public: ...@@ -51,7 +51,7 @@ public:
output_stream<Out, std::tuple<>, intrusive_ptr<stream_source>> output_stream<Out, std::tuple<>, intrusive_ptr<stream_source>>
add_outbound_path() { add_outbound_path() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
return this->add_unsafe_outbound_path<Out>().rebind(this); return this->add_unchecked_outbound_path<Out>().rebind(this);
} }
/// Creates a new output path to the current sender with custom handshake. /// Creates a new output path to the current sender with custom handshake.
...@@ -60,7 +60,7 @@ public: ...@@ -60,7 +60,7 @@ public:
intrusive_ptr<stream_source>> intrusive_ptr<stream_source>>
add_outbound_path(std::tuple<Ts...> xs) { add_outbound_path(std::tuple<Ts...> xs) {
CAF_LOG_TRACE(CAF_ARG(xs)); CAF_LOG_TRACE(CAF_ARG(xs));
return this->add_unsafe_outbound_path<Out>(std::move(xs)).rebind(this); return this->add_unchecked_outbound_path<Out>(std::move(xs)).rebind(this);
} }
/// Creates a new output path to the current sender. /// Creates a new output path to the current sender.
...@@ -68,7 +68,7 @@ public: ...@@ -68,7 +68,7 @@ public:
output_stream<Out, std::tuple<>, intrusive_ptr<stream_source>> output_stream<Out, std::tuple<>, intrusive_ptr<stream_source>>
add_outbound_path(const Handle& next) { add_outbound_path(const Handle& next) {
CAF_LOG_TRACE(CAF_ARG(next)); CAF_LOG_TRACE(CAF_ARG(next));
return this->add_unsafe_outbound_path<Out>(next).rebind(this); return this->add_unchecked_outbound_path<Out>(next).rebind(this);
} }
/// Creates a new output path to the current sender with custom handshake. /// Creates a new output path to the current sender with custom handshake.
...@@ -77,7 +77,7 @@ public: ...@@ -77,7 +77,7 @@ public:
intrusive_ptr<stream_source>> intrusive_ptr<stream_source>>
add_outbound_path(const Handle& next, std::tuple<Ts...> xs) { add_outbound_path(const Handle& next, std::tuple<Ts...> xs) {
CAF_LOG_TRACE(CAF_ARG(next) << CAF_ARG(xs)); CAF_LOG_TRACE(CAF_ARG(next) << CAF_ARG(xs));
return this->add_unsafe_outbound_path<Out>(next, std::move(xs)) return this->add_unchecked_outbound_path<Out>(next, std::move(xs))
.rebind(this); .rebind(this);
} }
......
...@@ -176,7 +176,7 @@ void stream_manager::remove_input_path(stream_slot slot, error reason, ...@@ -176,7 +176,7 @@ void stream_manager::remove_input_path(stream_slot slot, error reason,
self_->erase_inbound_path_later(slot, std::move(reason)); self_->erase_inbound_path_later(slot, std::move(reason));
} }
stream_slot stream_manager::add_unsafe_outbound_path_impl(response_promise& rp, stream_slot stream_manager::add_unchecked_outbound_path_impl(response_promise& rp,
message handshake) { message handshake) {
CAF_LOG_TRACE(CAF_ARG(rp) << CAF_ARG(handshake)); CAF_LOG_TRACE(CAF_ARG(rp) << CAF_ARG(handshake));
CAF_ASSERT(out().terminal() == false); CAF_ASSERT(out().terminal() == false);
...@@ -195,31 +195,31 @@ stream_slot stream_manager::add_unsafe_outbound_path_impl(response_promise& rp, ...@@ -195,31 +195,31 @@ stream_slot stream_manager::add_unsafe_outbound_path_impl(response_promise& rp,
return slot; return slot;
} }
stream_slot stream_manager::add_unsafe_outbound_path_impl(strong_actor_ptr next, stream_slot stream_manager::add_unchecked_outbound_path_impl(strong_actor_ptr next,
message handshake) { message handshake) {
CAF_LOG_TRACE(CAF_ARG(next) << CAF_ARG(handshake)); CAF_LOG_TRACE(CAF_ARG(next) << CAF_ARG(handshake));
response_promise rp{self_->ctrl(), nullptr, {next}, make_message_id()}; response_promise rp{self_->ctrl(), nullptr, {next}, make_message_id()};
return add_unsafe_outbound_path_impl(rp, std::move(handshake)); return add_unchecked_outbound_path_impl(rp, std::move(handshake));
} }
stream_slot stream_manager::add_unsafe_outbound_path_impl(message handshake) { stream_slot stream_manager::add_unchecked_outbound_path_impl(message handshake) {
CAF_LOG_TRACE(CAF_ARG(handshake)); CAF_LOG_TRACE(CAF_ARG(handshake));
auto rp = self_->make_response_promise(); auto rp = self_->make_response_promise();
return add_unsafe_outbound_path_impl(rp, std::move(handshake)); return add_unchecked_outbound_path_impl(rp, std::move(handshake));
} }
stream_slot stream_manager::add_unsafe_inbound_path_impl() { stream_slot stream_manager::add_unchecked_inbound_path_impl() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto x = self_->current_mailbox_element(); auto x = self_->current_mailbox_element();
if (x == nullptr || !x->content().match_elements<open_stream_msg>()) { if (x == nullptr || !x->content().match_elements<open_stream_msg>()) {
CAF_LOG_ERROR("add_unsafe_inbound_path called, but current message " CAF_LOG_ERROR("add_unchecked_inbound_path called, but current message "
"is not an open_stream_msg"); "is not an open_stream_msg");
return invalid_stream_slot; return invalid_stream_slot;
} }
auto& osm = x->content().get_mutable_as<open_stream_msg>(0); auto& osm = x->content().get_mutable_as<open_stream_msg>(0);
if (out().terminal() && !self_->current_forwarding_stack().empty()) { if (out().terminal() && !self_->current_forwarding_stack().empty()) {
// Sinks must always terminate the stream. // Sinks must always terminate the stream.
CAF_LOG_WARNING("add_unsafe_inbound_path called in a sink, but the " CAF_LOG_WARNING("add_unchecked_inbound_path called in a sink, but the "
"handshake has further stages"); "handshake has further stages");
stream_slots path_id{osm.slot, 0}; stream_slots path_id{osm.slot, 0};
auto code = sec::cannot_add_downstream; auto code = sec::cannot_add_downstream;
......
...@@ -225,24 +225,24 @@ TESTEE(stream_multiplexer) { ...@@ -225,24 +225,24 @@ TESTEE(stream_multiplexer) {
[=](join_atom, ints_atom) { [=](join_atom, ints_atom) {
auto& stg = self->state.stage; auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request for integers"); CAF_MESSAGE("received 'join' request for integers");
auto result = stg->add_unsafe_outbound_path<int>(); auto result = stg->add_unchecked_outbound_path<int>();
stg->out().assign<int_scatterer>(result.out()); stg->out().assign<int_scatterer>(result.out());
return result; return result;
}, },
[=](join_atom, strings_atom) { [=](join_atom, strings_atom) {
auto& stg = self->state.stage; auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request for strings"); CAF_MESSAGE("received 'join' request for strings");
auto result = stg->add_unsafe_outbound_path<string>(); auto result = stg->add_unchecked_outbound_path<string>();
stg->out().assign<string_scatterer>(result.out()); stg->out().assign<string_scatterer>(result.out());
return result; return result;
}, },
[=](const stream<int>& in) { [=](const stream<int>& in) {
CAF_MESSAGE("received handshake for integers"); CAF_MESSAGE("received handshake for integers");
return self->state.stage->add_unsafe_inbound_path(in); return self->state.stage->add_unchecked_inbound_path(in);
}, },
[=](const stream<string>& in) { [=](const stream<string>& in) {
CAF_MESSAGE("received handshake for strings"); CAF_MESSAGE("received handshake for strings");
return self->state.stage->add_unsafe_inbound_path(in); return self->state.stage->add_unchecked_inbound_path(in);
} }
}; };
} }
......
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