Commit 8d110349 authored by Dominik Charousset's avatar Dominik Charousset

Implement self->new_stream, relates #531

parent 62f0e627
...@@ -25,6 +25,8 @@ ...@@ -25,6 +25,8 @@
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/deep_to_string.hpp" #include "caf/deep_to_string.hpp"
#include "caf/meta/type_name.hpp"
namespace caf { namespace caf {
/// Stores a flow-control configuration. /// Stores a flow-control configuration.
...@@ -34,16 +36,10 @@ struct named_actor_config { ...@@ -34,16 +36,10 @@ struct named_actor_config {
size_t max_pending; size_t max_pending;
}; };
template <class Processor> template <class Inspector>
void serialize(Processor& proc, named_actor_config& x, unsigned int) { typename Inspector::result_type inspect(Inspector& f, named_actor_config& x) {
proc & x.strategy; return f(meta::type_name("named_actor_config"), x.strategy, x.low_watermark,
proc & x.low_watermark; x.max_pending);
proc & x.max_pending;
}
inline std::string to_string(const named_actor_config& x) {
return "named_actor_config"
+ deep_to_string_as_tuple(x.strategy, x.low_watermark, x.max_pending);
} }
} // namespace caf } // namespace caf
......
...@@ -38,6 +38,7 @@ ...@@ -38,6 +38,7 @@
#include "caf/stream_sink_impl.hpp" #include "caf/stream_sink_impl.hpp"
#include "caf/stream_stage_impl.hpp" #include "caf/stream_stage_impl.hpp"
#include "caf/stream_source_impl.hpp" #include "caf/stream_source_impl.hpp"
#include "caf/stream_result_trait.hpp"
#include "caf/policy/greedy.hpp" #include "caf/policy/greedy.hpp"
#include "caf/policy/anycast.hpp" #include "caf/policy/anycast.hpp"
...@@ -294,6 +295,57 @@ public: ...@@ -294,6 +295,57 @@ public:
// -- stream management ------------------------------------------------------ // -- stream management ------------------------------------------------------
// Starts a new stream.
template <class Handle, class Init, class Getter, class ClosedPredicate,
class ResHandler>
stream<typename stream_source_trait_t<Getter>::output>
new_stream(const Handle& dest, Init init, Getter getter, ClosedPredicate pred,
ResHandler res_handler) {
using type = typename stream_source_trait_t<Getter>::output;
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<
void (state_type&),
typename detail::get_callable_trait<Init>::fun_sig
>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
bool (const state_type&),
typename detail::get_callable_trait<ClosedPredicate>::fun_sig
>::value,
"Expected signature `bool (const State&)` for "
"closed_predicate function");
if (!dest) {
CAF_LOG_ERROR("cannot stream to an invalid actor handle");
return stream_id{nullptr, 0};
}
// generate new stream ID
stream_id sid{ctrl(),
new_request_id(message_priority::normal).integer_value()};
stream<type> token{sid};
// generate new ID for the final response message and send handshake
auto res_id = new_request_id(message_priority::normal);
dest->enqueue(make_mailbox_element(
ctrl(), res_id, {},
make<stream_msg::open>(sid, make_message(std::move(token)),
ctrl(), stream_priority::normal,
std::vector<atom_value>{}, false)),
context());
// install response handler
this->add_multiplexed_response_handler(
res_id.response_id(),
behavior{[=](result_type& res) { res_handler(std::move(res)); },
[=](error& err) { res_handler(std::move(err)); }});
// install stream handler
using impl = stream_source_impl<Getter, ClosedPredicate>;
std::unique_ptr<downstream_policy> p{new policy::anycast};
auto ptr = make_counted<impl>(this, sid, std::move(p), std::move(getter),
std::move(pred));
init(ptr->state());
streams_.emplace(std::move(sid), std::move(ptr));
return sid;
}
/// Adds a stream source to this actor. /// Adds a stream source to this actor.
template <class Init, class Getter, class ClosedPredicate> template <class Init, class Getter, class ClosedPredicate>
stream<typename stream_source_trait_t<Getter>::output> stream<typename stream_source_trait_t<Getter>::output>
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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://openresult.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_RESULT_TRAIT_HPP
#define CAF_STREAM_RESULT_TRAIT_HPP
#include "caf/detail/type_traits.hpp"
namespace caf {
template <class F>
struct stream_result_trait;
template <class T>
struct stream_result_trait<void (expected<T>)> {
using type = T;
};
template <class F>
using stream_result_trait_t =
stream_result_trait<typename detail::get_callable_trait<F>::fun_sig>;
} // namespace caf
#endif // CAF_STREAM_RESULT_TRAIT_HPP
...@@ -49,7 +49,7 @@ public: ...@@ -49,7 +49,7 @@ public:
} }
error consume(message& msg) final { error consume(message& msg) final {
using vec_type = std::vector<output_type>; using vec_type = std::vector<input_type>;
if (msg.match_elements<vec_type>()) { if (msg.match_elements<vec_type>()) {
auto& xs = msg.get_as<vec_type>(0); auto& xs = msg.get_as<vec_type>(0);
for (auto& x : xs) for (auto& x : xs)
......
...@@ -41,9 +41,9 @@ auto stream_msg_visitor::operator()(stream_msg::open& x) -> result_type { ...@@ -41,9 +41,9 @@ auto stream_msg_visitor::operator()(stream_msg::open& x) -> result_type {
return {sec::downstream_already_exists, e_}; return {sec::downstream_already_exists, e_};
auto& predecessor = x.prev_stage; auto& predecessor = x.prev_stage;
auto fail = [&](error reason) -> result_type { auto fail = [&](error reason) -> result_type {
unsafe_send_as(self_, predecessor, make<stream_msg::abort>(sid_, reason));
auto rp = self_->make_response_promise(); auto rp = self_->make_response_promise();
rp.deliver(reason); rp.deliver(reason);
unsafe_send_as(self_, predecessor, make<stream_msg::abort>(sid_, reason));
return {std::move(reason), e_}; return {std::move(reason), e_};
}; };
if (!predecessor) { if (!predecessor) {
......
...@@ -59,6 +59,33 @@ behavior file_reader(event_based_actor* self) { ...@@ -59,6 +59,33 @@ behavior file_reader(event_based_actor* self) {
}; };
} }
void streamer(event_based_actor* self, const actor& dest) {
using buf = std::deque<int>;
self->new_stream(
// destination of the stream
dest,
// initialize state
[&](buf& xs) {
xs = buf{1, 2, 3, 4, 5, 6, 7, 8, 9};
},
// get next element
[=](buf& xs, downstream<int>& out, size_t num) {
auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i)
out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
},
// check whether we reached the end
[=](const buf& xs) {
return xs.empty();
},
// handle result of the stream
[=](expected<int>) {
// nop
}
);
}
behavior filter(event_based_actor* self) { behavior filter(event_based_actor* self) {
return { return {
[=](stream<int>& in) -> stream<int> { [=](stream<int>& in) -> stream<int> {
...@@ -114,6 +141,56 @@ behavior sum_up(event_based_actor* self) { ...@@ -114,6 +141,56 @@ behavior sum_up(event_based_actor* self) {
}; };
} }
behavior drop_all(event_based_actor* self) {
return {
[=](stream<int>& in) {
return self->add_sink(
// input stream
in,
// initialize state
[](unit_t&) {
// nop
},
// processing step
[](unit_t&, int) {
// nop
},
// cleanup and produce void "result"
[](unit_t&) -> unit_t {
return unit;
}
);
}
};
}
void streamer_without_result(event_based_actor* self, const actor& dest) {
using buf = std::deque<int>;
self->new_stream(
// destination of the stream
dest,
// initialize state
[&](buf& xs) {
xs = buf{1, 2, 3, 4, 5, 6, 7, 8, 9};
},
// get next element
[=](buf& xs, downstream<int>& out, size_t num) {
auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i)
out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
},
// check whether we reached the end
[=](const buf& xs) {
return xs.empty();
},
// handle result of the stream
[=](expected<void>) {
// nop
}
);
}
using fixture = test_coordinator_fixture<>; using fixture = test_coordinator_fixture<>;
} // namespace <anonymous> } // namespace <anonymous>
...@@ -263,4 +340,52 @@ CAF_TEST(depth3_pipeline) { ...@@ -263,4 +340,52 @@ CAF_TEST(depth3_pipeline) {
//CAF_CHECK_EQUAL(res, 25); //CAF_CHECK_EQUAL(res, 25);
} }
CAF_TEST(broken_pipeline_stramer) {
CAF_MESSAGE("streams must abort if a stage fails to initialize its state");
auto stage = sys.spawn(broken_filter);
// run initialization code
sched.run();
auto source = sys.spawn(streamer, stage);
// run initialization code
sched.run_once();
// source --(stream_msg::open)--> stage
expect((stream_msg::open), from(source).to(stage).with(_, source, _, _, false));
CAF_CHECK(!deref(source).streams().empty());
CAF_CHECK(deref(stage).streams().empty());
// stage --(stream_msg::abort)--> source
expect((stream_msg::abort),
from(stage).to(source).with(sec::stream_init_failed));
CAF_CHECK(deref(source).streams().empty());
CAF_CHECK(deref(stage).streams().empty());
// sink ----(error)---> source
expect((error), from(stage).to(source).with(_));
}
CAF_TEST(depth2_pipeline_streamer) {
auto sink = sys.spawn(sum_up);
// run initialization code
sched.run();
auto source = sys.spawn(streamer, 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: 25)---> source
expect((int), from(sink).to(source).with(45));
}
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