Commit ea00044b authored by Dominik Charousset's avatar Dominik Charousset

Properly include sender slot in handshake

parent a2964318
...@@ -61,8 +61,8 @@ public: ...@@ -61,8 +61,8 @@ public:
init(state_); init(state_);
} }
handshake_tuple_type make_handshake() const override { handshake_tuple_type make_handshake(stream_slot slot) const override {
return std::tuple_cat(std::make_tuple(none), hs_); return std::tuple_cat(std::make_tuple(stream_type{slot}), hs_);
} }
void pull(downstream<output_type>& out, size_t num) override { void pull(downstream<output_type>& out, size_t num) override {
......
...@@ -85,8 +85,8 @@ public: ...@@ -85,8 +85,8 @@ public:
return hint != out_.capacity(); return hint != out_.capacity();
} }
message make_handshake() const override { message make_handshake(stream_slot slot) const override {
return make_message_from_tuple(driver_.make_handshake()); return make_message_from_tuple(driver_.make_handshake(slot));
} }
private: private:
......
...@@ -65,8 +65,8 @@ public: ...@@ -65,8 +65,8 @@ public:
init(state_); init(state_);
} }
handshake_tuple_type make_handshake() const override { handshake_tuple_type make_handshake(stream_slot slot) const override {
return std::tuple_cat(std::make_tuple(none), hs_); return std::tuple_cat(std::make_tuple(stream_type{slot}), hs_);
} }
void process(std::vector<input_type>&& batch, void process(std::vector<input_type>&& batch,
......
...@@ -85,8 +85,8 @@ public: ...@@ -85,8 +85,8 @@ public:
CAF_LOG_ERROR("received unexpected batch type (dropped)"); CAF_LOG_ERROR("received unexpected batch type (dropped)");
} }
message make_handshake() const override { message make_handshake(stream_slot slot) const override {
return make_message_from_tuple(driver_.make_handshake()); return make_message_from_tuple(driver_.make_handshake(slot));
} }
bool congested() const override { bool congested() const override {
......
...@@ -871,13 +871,6 @@ public: ...@@ -871,13 +871,6 @@ public:
return bhvr_stack_; return bhvr_stack_;
} }
template <class T, class... Ts>
static message make_handshake(std::tuple<Ts...>&& xs) {
stream<T> token;
auto ys = std::tuple_cat(std::forward_as_tuple(token), std::move(xs));
return make_message_from_tuple(std::move(ys));
}
/* /*
/// Tries to add a new sink to the stream manager `mgr`. /// Tries to add a new sink to the stream manager `mgr`.
/// @param mgr Pointer to the responsible stream manager. /// @param mgr Pointer to the responsible stream manager.
......
...@@ -43,13 +43,9 @@ public: ...@@ -43,13 +43,9 @@ public:
stream& operator=(stream&&) = default; stream& operator=(stream&&) = default;
stream& operator=(const stream&) = default; stream& operator=(const stream&) = default;
stream(none_t = none) : slot_(0) {
// nop
}
/// Convenience constructor for returning the result of `self->new_stream` /// Convenience constructor for returning the result of `self->new_stream`
/// and similar functions. /// and similar functions.
stream(stream_slot id, stream_manager_ptr sptr = nullptr) explicit stream(stream_slot id = 0, stream_manager_ptr sptr = nullptr)
: slot_(id), : slot_(id),
ptr_(std::move(sptr)) { ptr_(std::move(sptr)) {
// nop // nop
......
...@@ -183,7 +183,7 @@ protected: ...@@ -183,7 +183,7 @@ protected:
/// Returns a type-erased `stream<T>` as handshake token for downstream /// Returns a type-erased `stream<T>` as handshake token for downstream
/// actors. Returns an empty message for sinks. /// actors. Returns an empty message for sinks.
virtual message make_handshake() const; virtual message make_handshake(stream_slot slot) const;
/// Called whenever new credit becomes available. The default implementation /// Called whenever new credit becomes available. The default implementation
/// logs an error (sources are expected to override this hook). /// logs an error (sources are expected to override this hook).
......
...@@ -60,7 +60,7 @@ public: ...@@ -60,7 +60,7 @@ public:
// -- pure virtual functions ------------------------------------------------- // -- pure virtual functions -------------------------------------------------
/// Generates handshake data for the next actor in the pipeline. /// Generates handshake data for the next actor in the pipeline.
virtual handshake_tuple_type make_handshake() const = 0; virtual handshake_tuple_type make_handshake(stream_slot slot) const = 0;
/// Generates more stream elements. /// Generates more stream elements.
virtual void pull(downstream<output_type>& out, size_t num) = 0; virtual void pull(downstream<output_type>& out, size_t num) = 0;
......
...@@ -63,7 +63,7 @@ public: ...@@ -63,7 +63,7 @@ public:
// -- pure virtual functions ------------------------------------------------- // -- pure virtual functions -------------------------------------------------
/// Generates handshake data for the next actor in the pipeline. /// Generates handshake data for the next actor in the pipeline.
virtual handshake_tuple_type make_handshake() const = 0; virtual handshake_tuple_type make_handshake(stream_slot slot) const = 0;
/// Processes a single batch. /// Processes a single batch.
virtual void process(std::vector<input_type>&& batch, virtual void process(std::vector<input_type>&& batch,
......
...@@ -129,9 +129,9 @@ void stream_manager::send_handshake(strong_actor_ptr dest, stream_slot slot, ...@@ -129,9 +129,9 @@ void stream_manager::send_handshake(strong_actor_ptr dest, stream_slot slot,
in_flight_promises_.emplace( in_flight_promises_.emplace(
slot, response_promise{self()->ctrl(), client, fwd_stack, mid}); slot, response_promise{self()->ctrl(), client, fwd_stack, mid});
dest->enqueue( dest->enqueue(
make_mailbox_element( make_mailbox_element(std::move(client), mid, std::move(fwd_stack),
std::move(client), mid, std::move(fwd_stack), open_stream_msg{slot, make_handshake(slot),
open_stream_msg{slot, make_handshake(), self_->ctrl(), dest, priority_}), self_->ctrl(), dest, priority_}),
self_->context()); self_->context());
} }
...@@ -195,7 +195,7 @@ void stream_manager::output_closed(error) { ...@@ -195,7 +195,7 @@ void stream_manager::output_closed(error) {
// nop // nop
} }
message stream_manager::make_handshake() const { message stream_manager::make_handshake(stream_slot) const {
CAF_LOG_ERROR("stream_manager::make_handshake called"); CAF_LOG_ERROR("stream_manager::make_handshake called");
return none; return none;
} }
......
...@@ -245,8 +245,8 @@ public: ...@@ -245,8 +245,8 @@ public:
// nop // nop
} }
handshake_tuple_type make_handshake() const override { handshake_tuple_type make_handshake(stream_slot slot) const override {
return std::make_tuple(none); return std::make_tuple(stream_type{slot});
} }
void pull(downstream<int>& out, size_t hint) override { void pull(downstream<int>& out, size_t hint) override {
...@@ -278,8 +278,8 @@ public: ...@@ -278,8 +278,8 @@ public:
// nop // nop
} }
handshake_tuple_type make_handshake() const override { handshake_tuple_type make_handshake(stream_slot slot) const override {
return std::make_tuple(none); return std::make_tuple(stream_type{slot});
} }
void process(vector<int>&& batch, downstream<int>& out) override { void process(vector<int>&& batch, downstream<int>& out) override {
......
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