Commit f742ac1c authored by Hauke Goldhammer's avatar Hauke Goldhammer

add the changes requested in the comments of the last commits

parent 4d9f8aca
......@@ -39,14 +39,14 @@ struct stream_reader_state {
// nop
}
void init(std::unique_ptr<InputStream> src_stream) {
stream = std::move(src_stream);
void init(std::unique_ptr<InputStream> input) {
this->input = std::move(input);
}
// -- properties -------------------------------------------------------------
size_t at_end() const {
return !(*stream);
return !(*input);
}
// -- member variables -------------------------------------------------------
......@@ -54,9 +54,9 @@ struct stream_reader_state {
/// Gives this actor a useful name in CAF logs.
const char* name;
/// Stream we are about to stream
/// Input stream we are about to stream
// TODO: change after having raised the minimum GCC version to 5.
std::unique_ptr<InputStream> stream;
std::unique_ptr<InputStream> input;
/// Caches the line we are about to parse.
std::string line;
......@@ -69,25 +69,24 @@ using stream_source_type = stateful_actor<stream_reader_state<InputStream>>;
/// Streams the content of given 'src_stream' line by line using the given
/// policy to all given stream sinks.
template <class Policy, class InputStream, class Handle, class... Handles>
behavior stream_reader(stream_source_type<InputStream>* self,
std::unique_ptr<InputStream> src_stream, Handle sink,
Handles... sinks) {
void stream_reader(stream_source_type<InputStream>* self,
std::unique_ptr<InputStream> input, Handle sink,
Handles... sinks) {
using value_type = typename Policy::value_type;
self->state.init(std::move(src_stream));
self->state.init(std::move(input));
// Fail early if we got nothing to stream.
if (self->state.at_end())
return {};
self->quit();
// Spin up stream manager and connect the first sink.
auto src = self->make_source(
std::move(sink),
[&](unit_t&) {
[&](Policy& pol) {
// nop
},
[self](unit_t&, downstream<value_type>& out, size_t hint) {
[self](Policy& pol, downstream<value_type>& out, size_t hint) {
auto& st = self->state;
Policy pol;
size_t i = 0;
while (i < hint && getline(*(st.stream), st.line)) {
while (i < hint && getline(*(st.input), st.line)) {
if (auto count = pol(st.line, out)) {
i += *count;
} else {
......@@ -95,10 +94,10 @@ behavior stream_reader(stream_source_type<InputStream>* self,
}
}
},
[self](const unit_t&) { return self->state.at_end(); });
[self](const Policy& pol) { return self->state.at_end(); });
// Add the remaining sinks.
unit(src.ptr()->add_outbound_path(sinks)...);
return {};
self->quit();
}
} // namespace bb
......
......@@ -25,22 +25,22 @@
#include "caf/behavior.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/pec.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/unit.hpp"
#include "caf/pec.hpp"
namespace caf {
namespace bb {
namespace policy {
/// The Policy defines how the stream_reader pares a line of the given stream to
/// integers.
/// Parses whitespace-separated integers from input strings to 'ValueType' and
/// pushes generated integers to a downstream.
template <class ValueType = int32_t>
class tokenized_integer_reader {
public:
using value_type = ValueType;
/// Returns number of produced elements or an error.
/// Returns the number of parsed integers or an error.
expected<size_t> operator()(const std::string& line,
downstream<value_type> out) {
size_t count = 0;
......@@ -72,5 +72,5 @@ public:
}
};
} // namespace bb
} // namespace policy
} // namespace caf
......@@ -19,7 +19,7 @@
#define CAF_SUITE stream_reader
#include "caf/bb/stream_reader.hpp"
#include "caf/bb/tokenized_integer_reader.hpp"
#include "caf/policy/tokenized_integer_reader.hpp"
#include "caf/test/dsl.hpp"
......@@ -36,7 +36,7 @@ using namespace caf;
namespace {
using stream_type = std::istringstream;
using value_type = bb::tokenized_integer_reader<>::value_type;
using value_type = policy::tokenized_integer_reader<>::value_type;
TESTEE_SETUP();
......@@ -53,7 +53,7 @@ TESTEE(stream_reader_sink) {
[=](unit_t&) {
// nop
},
// Consumer
// consume values
[=](unit_t&, value_type val) {
self->state.vec.emplace_back(std::move(val));
},
......@@ -76,8 +76,7 @@ TESTEE(stream_monitor) {
self->set_down_handler([=](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.source, self->state.streamer);
if (dm.reason)
CAF_CHECK_EQUAL(dm.reason,
pec::unexpected_character);
CAF_CHECK_EQUAL(dm.reason, pec::unexpected_character);
});
return {[=](join_atom, actor streamer) {
......@@ -88,7 +87,7 @@ TESTEE(stream_monitor) {
struct config : actor_system_config {
config() {
add_message_type<value_type>("value_type");
// nop
}
};
......@@ -107,7 +106,7 @@ CAF_TEST(stream_to_sink) {
1254, 1, 20, 4, 56, 78, 95};
auto sink = sys.spawn(stream_reader_sink);
auto src
= sys.spawn(bb::stream_reader<bb::tokenized_integer_reader<value_type>,
= sys.spawn(bb::stream_reader<policy::tokenized_integer_reader<value_type>,
stream_type, actor>,
std::move(ptr_test_stream), sink);
auto mon = sys.spawn(stream_monitor);
......@@ -128,7 +127,7 @@ CAF_TEST(stream_to_sinks) {
auto snk2 = sys.spawn(stream_reader_sink);
auto snk3 = sys.spawn(stream_reader_sink);
auto src
= sys.spawn(bb::stream_reader<bb::tokenized_integer_reader<value_type>,
= sys.spawn(bb::stream_reader<policy::tokenized_integer_reader<value_type>,
stream_type, actor, actor, actor>,
std::move(ptr_test_stream), snk1, snk2, snk3);
auto mon = sys.spawn(stream_monitor);
......@@ -149,7 +148,7 @@ CAF_TEST(error_stream_to_sink) {
new stream_type(test_stringvalues)};
auto sink = sys.spawn(stream_reader_sink);
auto src
= sys.spawn(bb::stream_reader<bb::tokenized_integer_reader<value_type>,
= sys.spawn(bb::stream_reader<policy::tokenized_integer_reader<value_type>,
stream_type, actor>,
std::move(ptr_test_stream), sink);
auto mon = sys.spawn(stream_monitor);
......
......@@ -18,7 +18,7 @@
#define CAF_SUITE tokenized_integer_reader
#include "caf/bb/tokenized_integer_reader.hpp"
#include "caf/policy/tokenized_integer_reader.hpp"
#include "caf/test/dsl.hpp"
......@@ -29,7 +29,7 @@ using namespace caf;
namespace {
using value_type = bb::tokenized_integer_reader<>::value_type;
using value_type = policy::tokenized_integer_reader<>::value_type;
struct fixture {};
......@@ -38,7 +38,7 @@ struct fixture {};
CAF_TEST_FIXTURE_SCOPE(tokenized_integer_reader_tests, fixture)
CAF_TEST(int_policy) {
bb::tokenized_integer_reader<value_type> pol;
policy::tokenized_integer_reader<value_type> pol;
std::deque<value_type> q;
std::deque<value_type> test{1, 2, 3, 4};
downstream<value_type> out(q);
......@@ -48,13 +48,12 @@ CAF_TEST(int_policy) {
}
CAF_TEST(error_int_policy) {
bb::tokenized_integer_reader<value_type> pol;
policy::tokenized_integer_reader<value_type> pol;
std::deque<value_type> q;
downstream<value_type> out(q);
std::string test_line = "1 r 3 4";
auto count = pol(test_line, out);
CAF_CHECK_EQUAL(count.error(),
pec::unexpected_character);
CAF_CHECK_EQUAL(count.error(), pec::unexpected_character);
std::string test_line2 = "1 -2247483648 3 4";
count = pol(test_line2, out);
CAF_CHECK_EQUAL(count.error(), pec::exponent_underflow);
......
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