Commit a86b134b authored by Sebastian Wölke's avatar Sebastian Wölke

Extend make source by a finalize function

parent 290f7d8e
...@@ -27,13 +27,14 @@ ...@@ -27,13 +27,14 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
template <class Output, class Pull, class Done, class HandshakeData> template <class Output, class Pull, class Done, class Finalize,
class HandshakeData>
class stream_source_driver_impl; class stream_source_driver_impl;
/// Identifies an unbound sequence of messages. /// Identifies an unbound sequence of messages.
template <class Output, class Pull, class Done, class... Ts> template <class Output, class Pull, class Done, class Finalize, class... Ts>
class stream_source_driver_impl<Output, Pull, Done, std::tuple<Ts...>> final class stream_source_driver_impl<Output, Pull, Done, Finalize, std::tuple<Ts...>>
: public stream_source_driver<Output, Ts...> { final : public stream_source_driver<Output, Ts...> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -54,9 +55,11 @@ public: ...@@ -54,9 +55,11 @@ public:
using state_type = typename trait::state; using state_type = typename trait::state;
template <class Init, class Tuple> template <class Init, class Tuple>
stream_source_driver_impl(Init init, Pull f, Done pred, Tuple&& hs) stream_source_driver_impl(Init init, Pull f, Done pred, Finalize fin,
Tuple&& hs)
: pull_(std::move(f)), : pull_(std::move(f)),
done_(std::move(pred)), done_(std::move(pred)),
finalize_(std::move(fin)),
hs_(std::forward<Tuple>(hs)) { hs_(std::forward<Tuple>(hs)) {
init(state_); init(state_);
} }
...@@ -73,10 +76,15 @@ public: ...@@ -73,10 +76,15 @@ public:
return done_(state_); return done_(state_);
} }
void finalize(const error& err) override {
finalize_(state_, err);
}
private: private:
state_type state_; state_type state_;
Pull pull_; Pull pull_;
Done done_; Done done_;
Finalize finalize_;
tuple_type hs_; tuple_type hs_;
}; };
......
...@@ -456,18 +456,20 @@ public: ...@@ -456,18 +456,20 @@ public:
return {0, std::move(ptr)}; return {0, std::move(ptr)};
} }
template <class... Ts, class Init, class Pull, class Done, template <class... Ts, class Init, class Pull, class Done, class Finalize,
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>, broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>> class Trait = stream_source_trait_t<Pull>>
output_stream<typename Trait::output, detail::decay_t<Ts>...> output_stream<typename Trait::output, detail::decay_t<Ts>...>
make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done, make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done,
policy::arg<Scatterer> = {}) { Finalize finalize, policy::arg<Scatterer> = {}) {
using tuple_type = std::tuple<detail::decay_t<Ts>...>; using tuple_type = std::tuple<detail::decay_t<Ts>...>;
using driver = detail::stream_source_driver_impl<typename Trait::output, using driver =
Pull, Done, tuple_type>; detail::stream_source_driver_impl<typename Trait::output, Pull, Done,
Finalize, tuple_type>;
return make_source<driver, Scatterer>(std::move(init), std::move(pull), return make_source<driver, Scatterer>(std::move(init), std::move(pull),
std::move(done), std::move(xs)); std::move(done), std::move(finalize),
std::move(xs));
} }
template <class... Ts, class Init, class Pull, class Done, template <class... Ts, class Init, class Pull, class Done,
...@@ -477,7 +479,32 @@ public: ...@@ -477,7 +479,32 @@ public:
output_stream<typename Trait::output, detail::decay_t<Ts>...> output_stream<typename Trait::output, detail::decay_t<Ts>...>
make_source(Init init, Pull pull, Done done, make_source(Init init, Pull pull, Done done,
policy::arg<Scatterer> scatterer_type = {}) { policy::arg<Scatterer> scatterer_type = {}) {
return make_source(std::make_tuple(), init, pull, done, scatterer_type); auto finalize = [](typename Trait::state, const error&) {};
return make_source(std::make_tuple(), init, pull, done, finalize,
scatterer_type);
}
template <class... Ts, class Init, class Pull, class Done, class Finalize,
class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>>
output_stream<typename Trait::output, detail::decay_t<Ts>...>
make_source(Init init, Pull pull, Done done, Finalize finalize,
policy::arg<Scatterer> scatterer_type = {}) {
return make_source(std::make_tuple(), init, pull, done, finalize,
scatterer_type);
}
template <class... Ts, class Init, class Pull, class Done,
class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>>
output_stream<typename Trait::output, detail::decay_t<Ts>...>
make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done,
policy::arg<Scatterer> scatterer_type = {}) {
auto finalize = [](typename Trait::state, const error&) {};
return make_source(std::move(xs), init, pull, done, finalize,
scatterer_type);
} }
template <class Driver, class Input, class... Ts> template <class Driver, class Input, class... Ts>
......
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