Commit 1161eeff authored by Dominik Charousset's avatar Dominik Charousset

Fix void result of streams

parent 8d110349
...@@ -303,7 +303,6 @@ public: ...@@ -303,7 +303,6 @@ public:
ResHandler res_handler) { ResHandler res_handler) {
using type = typename stream_source_trait_t<Getter>::output; using type = typename stream_source_trait_t<Getter>::output;
using state_type = typename stream_source_trait_t<Getter>::state; using state_type = typename stream_source_trait_t<Getter>::state;
using result_type = typename stream_result_trait_t<ResHandler>::type;
static_assert(std::is_same< static_assert(std::is_same<
void (state_type&), void (state_type&),
typename detail::get_callable_trait<Init>::fun_sig typename detail::get_callable_trait<Init>::fun_sig
...@@ -334,8 +333,7 @@ public: ...@@ -334,8 +333,7 @@ public:
// install response handler // install response handler
this->add_multiplexed_response_handler( this->add_multiplexed_response_handler(
res_id.response_id(), res_id.response_id(),
behavior{[=](result_type& res) { res_handler(std::move(res)); }, stream_result_trait_t<ResHandler>::make_result_handler(res_handler));
[=](error& err) { res_handler(std::move(err)); }});
// install stream handler // install stream handler
using impl = stream_source_impl<Getter, ClosedPredicate>; using impl = stream_source_impl<Getter, ClosedPredicate>;
std::unique_ptr<downstream_policy> p{new policy::anycast}; std::unique_ptr<downstream_policy> p{new policy::anycast};
......
...@@ -20,6 +20,9 @@ ...@@ -20,6 +20,9 @@
#ifndef CAF_STREAM_RESULT_TRAIT_HPP #ifndef CAF_STREAM_RESULT_TRAIT_HPP
#define CAF_STREAM_RESULT_TRAIT_HPP #define CAF_STREAM_RESULT_TRAIT_HPP
#include "caf/unit.hpp"
#include "caf/behavior.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
...@@ -30,6 +33,33 @@ struct stream_result_trait; ...@@ -30,6 +33,33 @@ struct stream_result_trait;
template <class T> template <class T>
struct stream_result_trait<void (expected<T>)> { struct stream_result_trait<void (expected<T>)> {
using type = T; using type = T;
template <class OnResult>
static behavior make_result_handler(OnResult f) {
return {
[=](T& res) {
f(std::move(res));
},
[=](error& err) {
f(std::move(err));
}
};
}
};
template <>
struct stream_result_trait<void (expected<void>)> {
using type = void;
template <class OnResult>
static behavior make_result_handler(OnResult f) {
return {
[=]() {
f(unit);
},
[=](error& err) {
f(std::move(err));
}
};
}
}; };
template <class F> template <class F>
......
...@@ -60,7 +60,7 @@ public: ...@@ -60,7 +60,7 @@ public:
} }
message finalize() final { message finalize() final {
return make_message(fin_(state_)); return trait::make_result(state_, fin_);
} }
state_type& state() { state_type& state() {
......
...@@ -20,6 +20,9 @@ ...@@ -20,6 +20,9 @@
#ifndef CAF_STREAM_SINK_TRAIT_HPP #ifndef CAF_STREAM_SINK_TRAIT_HPP
#define CAF_STREAM_SINK_TRAIT_HPP #define CAF_STREAM_SINK_TRAIT_HPP
#include "caf/message.hpp"
#include "caf/make_message.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
...@@ -32,6 +35,22 @@ struct stream_sink_trait<void (State&, In), Out (State&)> { ...@@ -32,6 +35,22 @@ struct stream_sink_trait<void (State&, In), Out (State&)> {
using state = State; using state = State;
using input = In; using input = In;
using output = Out; using output = Out;
template <class F>
static message make_result(state& st, F f) {
return make_message(f(st));
}
};
template <class State, class In>
struct stream_sink_trait<void (State&, In), void (State&)> {
using state = State;
using input = In;
using output = void;
template <class F>
static message make_result(state& st, F& f) {
f(st);
return make_message();
}
}; };
template <class Fun, class Fin> template <class Fun, class Fin>
......
...@@ -156,8 +156,8 @@ behavior drop_all(event_based_actor* self) { ...@@ -156,8 +156,8 @@ behavior drop_all(event_based_actor* self) {
// nop // nop
}, },
// cleanup and produce void "result" // cleanup and produce void "result"
[](unit_t&) -> unit_t { [](unit_t&) {
return unit; // nop
} }
); );
} }
...@@ -388,4 +388,31 @@ CAF_TEST(depth2_pipeline_streamer) { ...@@ -388,4 +388,31 @@ CAF_TEST(depth2_pipeline_streamer) {
expect((int), from(sink).to(source).with(45)); expect((int), from(sink).to(source).with(45));
} }
CAF_TEST(stream_without_result) {
auto sink = sys.spawn(drop_all);
// run initialization code
sched.run();
auto source = sys.spawn(streamer_without_result, sink);
// run initialization code
sched.run_once();
// source ----(stream_msgreturn ::open)----> sink
expect((stream_msg::open), from(source).to(sink).with(_, source, _, _, false));
// source <----(stream_msg::ack_open)------ sink
expect((stream_msg::ack_open), from(sink).to(source).with(5, _, false));
// source ----(stream_msg::batch)---> sink
expect((stream_msg::batch),
from(source).to(sink).with(5, std::vector<int>{1, 2, 3, 4, 5}, 0));
// source <--(stream_msg::ack_batch)---- sink
expect((stream_msg::ack_batch), from(sink).to(source).with(5, 0));
// source ----(stream_msg::batch)---> sink
expect((stream_msg::batch),
from(source).to(sink).with(4, std::vector<int>{6, 7, 8, 9}, 1));
// source <--(stream_msg::ack_batch)---- sink
expect((stream_msg::ack_batch), from(sink).to(source).with(4, 1));
// source ----(stream_msg::close)---> sink
expect((stream_msg::close), from(source).to(sink).with());
// sink ----(result: <empty>)---> source
expect((void), from(sink).to(source).with());
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
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