Commit 1629ebc4 authored by Dominik Charousset's avatar Dominik Charousset

Support finalize callbacks with 1 and 2 args

parent 673c0fd3
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#define CAF_STREAM_SINK_DRIVER_IMPL_HPP #define CAF_STREAM_SINK_DRIVER_IMPL_HPP
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/stream_finalize_trait.hpp"
#include "caf/stream_sink_driver.hpp" #include "caf/stream_sink_driver.hpp"
#include "caf/stream_sink_trait.hpp" #include "caf/stream_sink_trait.hpp"
...@@ -44,7 +45,7 @@ public: ...@@ -44,7 +45,7 @@ public:
template <class Init> template <class Init>
stream_sink_driver_impl(Init init, Process f, Finalize fin) stream_sink_driver_impl(Init init, Process f, Finalize fin)
: process_(std::move(f)), : process_(std::move(f)),
finalize_(std::move(fin)) { fin_(std::move(fin)) {
init(state_); init(state_);
} }
...@@ -53,12 +54,12 @@ public: ...@@ -53,12 +54,12 @@ public:
} }
void finalize(const error& err) override { void finalize(const error& err) override {
finalize_(state_, err); stream_finalize_trait<Finalize, state_type>::invoke(fin_, state_, err);
} }
private: private:
Process process_; Process process_;
Finalize finalize_; Finalize fin_;
state_type state_; state_type state_;
}; };
......
...@@ -26,8 +26,6 @@ ...@@ -26,8 +26,6 @@
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_sink.hpp" #include "caf/stream_sink.hpp"
#include "caf/stream_sink_trait.hpp"
#include "caf/terminal_stream_scatterer.hpp"
#include "caf/policy/arg.hpp" #include "caf/policy/arg.hpp"
...@@ -47,15 +45,10 @@ public: ...@@ -47,15 +45,10 @@ public:
stream_sink_impl(scheduled_actor* self, Ts&&... xs) stream_sink_impl(scheduled_actor* self, Ts&&... xs)
: stream_manager(self), : stream_manager(self),
super(self), super(self),
driver_(std::forward<Ts>(xs)...), driver_(std::forward<Ts>(xs)...) {
out_(self) {
// nop // nop
} }
terminal_stream_scatterer& out() override {
return out_;
}
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>;
...@@ -73,7 +66,6 @@ protected: ...@@ -73,7 +66,6 @@ protected:
private: private:
driver_type driver_; driver_type driver_;
terminal_stream_scatterer out_;
}; };
template <class Driver, class... Ts> template <class Driver, class... Ts>
......
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#define CAF_STREAM_SOURCE_DRIVER_IMPL_HPP #define CAF_STREAM_SOURCE_DRIVER_IMPL_HPP
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/stream_finalize_trait.hpp"
#include "caf/stream_source_driver.hpp" #include "caf/stream_source_driver.hpp"
#include "caf/stream_source_trait.hpp" #include "caf/stream_source_trait.hpp"
...@@ -45,7 +46,7 @@ public: ...@@ -45,7 +46,7 @@ public:
stream_source_driver_impl(Init init, Pull f, Done pred, Finalize fin) stream_source_driver_impl(Init init, Pull f, Done pred, Finalize fin)
: pull_(std::move(f)), : pull_(std::move(f)),
done_(std::move(pred)), done_(std::move(pred)),
finalize_(std::move(fin)) { fin_(std::move(fin)) {
init(state_); init(state_);
} }
...@@ -58,14 +59,14 @@ public: ...@@ -58,14 +59,14 @@ public:
} }
void finalize(const error& err) override { void finalize(const error& err) override {
finalize_(state_, err); stream_finalize_trait<Finalize, state_type>::invoke(fin_, state_, err);
} }
private: private:
state_type state_; state_type state_;
Pull pull_; Pull pull_;
Done done_; Done done_;
Finalize finalize_; Finalize fin_;
}; };
} // namespace detail } // namespace detail
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#ifndef CAF_STREAM_STAGE_DRIVER_IMPL_HPP #ifndef CAF_STREAM_STAGE_DRIVER_IMPL_HPP
#define CAF_STREAM_STAGE_DRIVER_IMPL_HPP #define CAF_STREAM_STAGE_DRIVER_IMPL_HPP
#include "caf/stream_finalize_trait.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/stream_stage_driver.hpp" #include "caf/stream_stage_driver.hpp"
#include "caf/stream_stage_trait.hpp" #include "caf/stream_stage_trait.hpp"
...@@ -61,7 +62,7 @@ public: ...@@ -61,7 +62,7 @@ public:
} }
void finalize(const error& err) override { void finalize(const error& err) override {
return fin_(state_, err); stream_finalize_trait<Finalize, state_type>::invoke(fin_, state_, err);
} }
private: private:
......
...@@ -490,6 +490,20 @@ public: ...@@ -490,6 +490,20 @@ public:
static constexpr bool value = std::is_same<bool, result_type>::value; static constexpr bool value = std::is_same<bool, result_type>::value;
}; };
/// Checks wheter `F` is callable with arguments of types `Ts...`.
template <class F, class... Ts>
struct is_callable_with {
template <class U>
static auto sfinae(U*)
-> decltype((std::declval<U&>())(std::declval<Ts>()...), std::true_type());
template <class U>
static auto sfinae(...) -> std::false_type;
using type = decltype(sfinae<F>(nullptr));
static constexpr bool value = type::value;
};
/// Checks wheter `F` takes mutable references. /// Checks wheter `F` takes mutable references.
/// ///
/// A manipulator is a functor that manipulates its arguments via /// A manipulator is a functor that manipulates its arguments via
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_FINALIZE_TRAIT_HPP
#define CAF_STREAM_FINALIZE_TRAIT_HPP
#include "caf/make_message.hpp"
#include "caf/message.hpp"
#include "caf/stream_sink.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf {
/// Dispatches a finalize call to a function taking either one or two arguments.
template <class Fun, class State,
bool AcceptsTwoArgs =
detail::is_callable_with<Fun, State&, const error&>::value>
struct stream_finalize_trait;
/// Specializes the trait for callbacks that only take the state.
template <class Fun, class State>
struct stream_finalize_trait<Fun, State, false>{
static void invoke(Fun& f, State& st, const error&) {
static_assert(detail::is_callable_with<Fun, State&>::value,
"Finalize function neither accepts (State&, const error&) "
"nor (State&)");
f(st);
}
};
/// Specializes the trait for callbacks that take state and error.
template <class Fun, class State>
struct stream_finalize_trait<Fun, State, true>{
static void invoke(Fun& f, State& st, const error& err) {
f(st, err);
}
};
} // namespace caf
#endif // CAF_STREAM_FINALIZE_TRAIT_HPP
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_result.hpp" #include "caf/stream_result.hpp"
#include "caf/terminal_stream_scatterer.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
...@@ -38,7 +39,7 @@ public: ...@@ -38,7 +39,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_sink(scheduled_actor* self) : stream_manager(self) { stream_sink(scheduled_actor* self) : stream_manager(self), dummy_out_(self) {
// nop // nop
} }
...@@ -48,6 +49,10 @@ public: ...@@ -48,6 +49,10 @@ public:
return !this->continuous() && this->inbound_paths_.empty(); return !this->continuous() && this->inbound_paths_.empty();
} }
stream_scatterer& out() override {
return dummy_out_;
}
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Creates a new input path to the current sender. /// Creates a new input path to the current sender.
...@@ -55,6 +60,9 @@ public: ...@@ -55,6 +60,9 @@ public:
add_inbound_path(const stream<input_type>&) { add_inbound_path(const stream<input_type>&) {
return {add_unchecked_inbound_path_impl(), this}; return {add_unchecked_inbound_path_impl(), this};
} }
private:
terminal_stream_scatterer dummy_out_;
}; };
template <class In> template <class In>
......
...@@ -46,6 +46,10 @@ public: ...@@ -46,6 +46,10 @@ public:
right_super(self) { right_super(self) {
// nop // nop
} }
Scatterer& out() override {
return left_super::out();
}
}; };
template <class In, class Out, class Scatterer> template <class In, class Out, class Scatterer>
......
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