Unverified Commit 92141dc7 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #2

Add bb::stream_reader with an example policy for tokenized integer.
parents 3b33fe79 40c87e0c
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <string>
#include <vector>
#include "caf/behavior.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/unit.hpp"
namespace caf {
namespace bb {
/// @relates stream_reader
template <class InputStream>
struct stream_reader_state {
// -- constructors, destructors, and assignment operators --------------------
stream_reader_state() : name("stream-reader") {
// nop
}
void init(std::unique_ptr<InputStream> input) {
this->input = std::move(input);
}
// -- properties -------------------------------------------------------------
size_t at_end() const {
return !(*input);
}
// -- member variables -------------------------------------------------------
/// Gives this actor a useful name in CAF logs.
const char* name;
/// User-defined data source for reading ASCII or UTF-8 input line by line.
// TODO: change after having raised the minimum GCC version to 5.
std::unique_ptr<InputStream> input;
/// Caches the line we are about to parse.
std::string line;
};
/// @relates stream_reader
template <class InputStream>
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>
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(input));
// Fail early if we got nothing to stream.
if (self->state.at_end())
return;
// Spin up stream manager and connect the first sink.
auto src = self->make_source(
std::move(sink),
[&](Policy& pol) {
// nop
},
[self](Policy& pol, downstream<value_type>& out, size_t hint) {
auto& st = self->state;
size_t i = 0;
while (i < hint && getline(*(st.input), st.line)) {
if (auto count = pol(st.line, out)) {
i += *count;
} else {
self->quit(count.error());
}
}
},
[self](const Policy& pol) { return self->state.at_end(); });
// Add the remaining sinks.
unit(src.ptr()->add_outbound_path(sinks)...);
}
} // namespace bb
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <limits>
#include <string>
#include <vector>
#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"
namespace caf {
namespace policy {
/// 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 the number of parsed integers or an error.
expected<size_t> operator()(const std::string& line,
downstream<value_type> out) {
size_t count = 0;
auto i = line.c_str();
while (*i != '\0') {
// Parse next integer.
char* end = nullptr;
auto value = strtoll(i, &end, 10);
if (errno == ERANGE) {
if (value < 0) {
return make_error(pec::exponent_underflow);
}
return make_error(pec::exponent_overflow);
}
if (std::numeric_limits<value_type>::min() > value)
return make_error(pec::exponent_underflow);
if (value > std::numeric_limits<value_type>::max())
return make_error(pec::exponent_overflow);
if (value == 0 && !(*end == ' ' || *end == '\0'))
return make_error(pec::unexpected_character);
++count;
out.push(value);
// Advance iterator.
i = end;
while (isspace(*i))
++i;
}
return count;
}
};
} // namespace policy
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#define CAF_SUITE stream_reader
#include "caf/bb/stream_reader.hpp"
#include "caf/policy/tokenized_integer_reader.hpp"
#include "caf/test/dsl.hpp"
#include <memory>
#include <sstream>
#include <string>
#include <vector>
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
using namespace caf;
namespace {
using stream_type = std::istringstream;
using value_type = policy::tokenized_integer_reader<>::value_type;
TESTEE_SETUP();
TESTEE_STATE(stream_reader_sink) {
std::vector<value_type> vec;
};
TESTEE(stream_reader_sink) {
return {[=](stream<value_type> in) {
return self->make_sink(
// input stream
in,
// initialize state
[=](unit_t&) {
// nop
},
// consume values
[=](unit_t&, value_type val) {
self->state.vec.emplace_back(std::move(val));
},
// cleanup and produce result message
[=](unit_t&, const error& e) {
if (e) {
CAF_MESSAGE(self->name() << " " << e);
} else {
CAF_MESSAGE(self->name() << " is done");
}
});
}};
}
TESTEE_STATE(stream_monitor) {
actor streamer;
};
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);
});
return {[=](join_atom, actor streamer) {
self->state.streamer = streamer;
self->monitor(streamer);
}};
}
struct config : actor_system_config {
config() {
// nop
}
};
using fixture = test_coordinator_fixture<config>;
} // namespace
CAF_TEST_FIXTURE_SCOPE(stream_reader_tests, fixture)
CAF_TEST(stream_to_sink) {
scoped_actor self{sys};
std::string test_stringvalues = "1 2 3 4 5 6 7 78 1254 1 20\n4 56 78 95";
std::unique_ptr<stream_type> ptr_test_stream{
new stream_type(test_stringvalues)};
std::vector<value_type> test_container{1, 2, 3, 4, 5, 6, 7, 78,
1254, 1, 20, 4, 56, 78, 95};
auto sink = sys.spawn(stream_reader_sink);
auto src
= 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);
self->send(mon, join_atom::value, src);
run();
CAF_CHECK_EQUAL(deref<stream_reader_sink_actor>(sink).state.vec,
test_container);
}
CAF_TEST(stream_to_sinks) {
scoped_actor self{sys};
std::string test_stringvalues = "1 2 3 4 5 6 7 78 1254 1 20\n4 56 78 95";
std::unique_ptr<stream_type> ptr_test_stream{
new stream_type(test_stringvalues)};
std::vector<value_type> test_container{1, 2, 3, 4, 5, 6, 7, 78,
1254, 1, 20, 4, 56, 78, 95};
auto snk1 = sys.spawn(stream_reader_sink);
auto snk2 = sys.spawn(stream_reader_sink);
auto snk3 = sys.spawn(stream_reader_sink);
auto src
= 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);
self->send(mon, join_atom::value, src);
run();
CAF_CHECK_EQUAL(deref<stream_reader_sink_actor>(snk1).state.vec,
test_container);
CAF_CHECK_EQUAL(deref<stream_reader_sink_actor>(snk2).state.vec,
test_container);
CAF_CHECK_EQUAL(deref<stream_reader_sink_actor>(snk3).state.vec,
test_container);
}
CAF_TEST(error_stream_to_sink) {
scoped_actor self{sys};
std::string test_stringvalues = "1 2 3 4 5 6 7 rr 1254";
std::unique_ptr<stream_type> ptr_test_stream{
new stream_type(test_stringvalues)};
auto sink = sys.spawn(stream_reader_sink);
auto src
= 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);
self->send(mon, join_atom::value, src);
run();
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#define CAF_SUITE tokenized_integer_reader
#include "caf/policy/tokenized_integer_reader.hpp"
#include "caf/test/dsl.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
using namespace caf;
namespace {
using value_type = policy::tokenized_integer_reader<>::value_type;
struct fixture {};
} // namespace
CAF_TEST_FIXTURE_SCOPE(tokenized_integer_reader_tests, fixture)
CAF_TEST(int_policy) {
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);
std::string test_line = "1 2 3 4";
pol(test_line, out);
CAF_CHECK_EQUAL(q, test);
}
CAF_TEST(error_int_policy) {
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);
std::string test_line2 = "1 -2247483648 3 4";
count = pol(test_line2, out);
CAF_CHECK_EQUAL(count.error(), pec::exponent_underflow);
std::string test_line3 = "1 2147483648 3 4";
count = pol(test_line3, out);
CAF_CHECK_EQUAL(count.error(), pec::exponent_overflow);
}
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