Unverified Commit cfb49df3 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #665

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