Commit 1d9b8d02 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Pass the gatherer to the scatterer on construction

parent 11fbc9a2
...@@ -29,7 +29,7 @@ class pull5_gatherer : public random_gatherer { ...@@ -29,7 +29,7 @@ class pull5_gatherer : public random_gatherer {
public: public:
using super = random_gatherer; using super = random_gatherer;
pull5_gatherer(local_actor* selfptr); pull5_gatherer(local_actor* selfptr, stream_scatterer&);
void assign_credit(long available) override; void assign_credit(long available) override;
......
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#ifndef CAF_RANDOM_GATHERER_HPP #ifndef CAF_RANDOM_GATHERER_HPP
#define CAF_RANDOM_GATHERER_HPP #define CAF_RANDOM_GATHERER_HPP
#include "caf/fwd.hpp"
#include "caf/stream_gatherer_impl.hpp" #include "caf/stream_gatherer_impl.hpp"
namespace caf { namespace caf {
...@@ -28,7 +29,7 @@ class random_gatherer : public stream_gatherer_impl { ...@@ -28,7 +29,7 @@ class random_gatherer : public stream_gatherer_impl {
public: public:
using super = stream_gatherer_impl; using super = stream_gatherer_impl;
random_gatherer(local_actor* selfptr); random_gatherer(local_actor* selfptr, stream_scatterer&);
~random_gatherer() override; ~random_gatherer() override;
......
...@@ -43,7 +43,8 @@ public: ...@@ -43,7 +43,8 @@ public:
stream_sink_impl(local_actor* self, Fun fun, Finalize fin) stream_sink_impl(local_actor* self, Fun fun, Finalize fin)
: fun_(std::move(fun)), : fun_(std::move(fun)),
fin_(std::move(fin)), fin_(std::move(fin)),
in_(self) { out_(self),
in_(self, out_) {
// nop // nop
} }
...@@ -85,8 +86,8 @@ private: ...@@ -85,8 +86,8 @@ private:
state_type state_; state_type state_;
Fun fun_; Fun fun_;
Finalize fin_; Finalize fin_;
Gatherer in_;
Scatterer out_; Scatterer out_;
Gatherer in_;
}; };
} // namespace caf } // namespace caf
......
...@@ -44,8 +44,8 @@ public: ...@@ -44,8 +44,8 @@ public:
Fun fun, Cleanup cleanup) Fun fun, Cleanup cleanup)
: fun_(std::move(fun)), : fun_(std::move(fun)),
cleanup_(std::move(cleanup)), cleanup_(std::move(cleanup)),
in_(self), out_(self),
out_(self) { in_(self, out_) {
// nop // nop
} }
...@@ -113,8 +113,8 @@ private: ...@@ -113,8 +113,8 @@ private:
state_type state_; state_type state_;
Fun fun_; Fun fun_;
Cleanup cleanup_; Cleanup cleanup_;
UpstreamPolicy in_;
DownstreamPolicy out_; DownstreamPolicy out_;
UpstreamPolicy in_;
}; };
} // namespace caf } // namespace caf
......
...@@ -27,7 +27,7 @@ namespace caf { ...@@ -27,7 +27,7 @@ namespace caf {
/// stream scatterer generates credit without downstream actors. /// stream scatterer generates credit without downstream actors.
class terminal_stream_scatterer : public stream_scatterer { class terminal_stream_scatterer : public stream_scatterer {
public: public:
terminal_stream_scatterer() = default; terminal_stream_scatterer(local_actor* = nullptr);
~terminal_stream_scatterer() override; ~terminal_stream_scatterer() override;
......
...@@ -21,7 +21,8 @@ ...@@ -21,7 +21,8 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
pull5_gatherer::pull5_gatherer(local_actor* selfptr) : super(selfptr) { pull5_gatherer::pull5_gatherer(local_actor* selfptr, stream_scatterer& out)
: super(selfptr, out) {
// nop // nop
} }
......
...@@ -20,7 +20,8 @@ ...@@ -20,7 +20,8 @@
namespace caf { namespace caf {
random_gatherer::random_gatherer(local_actor* selfptr) : super(selfptr) { random_gatherer::random_gatherer(local_actor* selfptr, stream_scatterer&)
: super(selfptr) {
// nop // nop
} }
......
...@@ -22,6 +22,10 @@ ...@@ -22,6 +22,10 @@
namespace caf { namespace caf {
terminal_stream_scatterer::terminal_stream_scatterer(local_actor*) {
// nop
}
terminal_stream_scatterer::~terminal_stream_scatterer() { terminal_stream_scatterer::~terminal_stream_scatterer() {
// nop // nop
} }
......
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