Commit 3ae049fe authored by Dominik Charousset's avatar Dominik Charousset

Remove obsolete files

parent afd89d5d
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#pragma once
#include <chrono>
#include <string>
#include <cstdint>
#include <cstddef> // size_t
#include <type_traits>
#include <iterator>
#include "caf/fwd.hpp"
#include "caf/atom.hpp"
#include "caf/error.hpp"
#include "caf/timestamp.hpp"
#include "caf/allowed_unsafe_message_type.hpp"
#include "caf/meta/annotation.hpp"
#include "caf/meta/save_callback.hpp"
#include "caf/meta/load_callback.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/apply_args.hpp"
#include "caf/detail/delegate_serialize.hpp"
#include "caf/detail/select_integer_type.hpp"
namespace caf {
/// A data processor translates an object into a format that can be
/// stored or vice versa. A data processor can be either in saving
/// or loading mode.
template <class Derived>
class data_processor {
public:
// -- member types -----------------------------------------------------------
/// Return type for `operator()`.
using result_type = error;
/// List of builtin types for data processors.
using builtin_t =
detail::type_list<
int8_t,
uint8_t,
int16_t,
uint16_t,
int32_t,
uint32_t,
int64_t,
uint64_t,
float,
double,
long double,
std::string,
std::u16string,
std::u32string
>;
// -- constructors, destructors, and assignment operators --------------------
data_processor(const data_processor&) = delete;
data_processor& operator=(const data_processor&) = delete;
data_processor(execution_unit* ctx = nullptr) : context_(ctx) {
// nop
}
virtual ~data_processor() {
// nop
}
// -- pure virtual functions -------------------------------------------------
/// Begins processing of an object. Saves the type information
/// to the underlying storage when in saving mode, otherwise
/// extracts them and sets both arguments accordingly.
virtual error begin_object(uint16_t& typenr, std::string& name) = 0;
/// Ends processing of an object.
virtual error end_object() = 0;
/// Begins processing of a sequence. Saves the size
/// to the underlying storage when in saving mode, otherwise
/// sets `num` accordingly.
virtual error begin_sequence(size_t& num) = 0;
/// Ends processing of a sequence.
virtual error end_sequence() = 0;
// -- getter -----------------------------------------------------------------
/// Returns the actor system associated to this data processor.
execution_unit* context() {
return context_;
}
// -- apply functions --------------------------------------------------------
/// Applies this processor to an arithmetic type.
template <class T>
typename std::enable_if<std::is_floating_point<T>::value, error>::type
apply(T& x) {
static constexpr auto tlindex = detail::tl_index_of<builtin_t, T>::value;
static_assert(tlindex >= 0, "T not recognized as builtin type");
return apply_impl(x);
}
template <class T>
typename std::enable_if<
std::is_integral<T>::value
&& !std::is_same<bool, T>::value,
error
>::type
apply(T& x) {
using type = detail::select_integer_type_t<sizeof(T),
std::is_signed<T>::value>;
return apply_impl(reinterpret_cast<type&>(x));
}
error apply(std::string& x) {
return apply_impl(x);
}
error apply(std::u16string& x) {
return apply_impl(x);
}
error apply(std::u32string& x) {
return apply_impl(x);
}
template <class D, atom_value V>
static error apply_atom_constant(D& self, atom_constant<V>&) {
static_assert(!D::writes_state, "cannot deserialize an atom_constant");
auto x = V;
return self.apply(x);
}
template <atom_value V>
error apply(atom_constant<V>& x) {
return apply_atom_constant(dref(), x);
}
/// Serializes enums using the underlying type
/// if no user-defined serialization is defined.
template <class T>
typename std::enable_if<
std::is_enum<T>::value
&& !detail::has_serialize<T>::value,
error
>::type
apply(T& x) {
using underlying = typename std::underlying_type<T>::type;
struct {
void operator()(T& lhs, underlying& rhs) const {
lhs = static_cast<T>(rhs);
}
void operator()(underlying& lhs, T& rhs) const {
lhs = static_cast<underlying>(rhs);
}
} assign;
underlying tmp = 0;
return convert_apply(dref(), x, tmp, assign);
}
/// Applies this processor to an empty type.
template <class T>
typename std::enable_if<
std::is_empty<T>::value && !detail::is_inspectable<Derived, T>::value,
error
>::type
apply(T&) {
return none;
}
error apply(bool& x) {
struct {
void operator()(bool& lhs, uint8_t& rhs) const {
lhs = rhs != 0;
}
void operator()(uint8_t& lhs, bool& rhs) const {
lhs = rhs ? 1 : 0;
}
} assign;
uint8_t tmp;
return convert_apply(dref(), x, tmp, assign);
}
// Special case to avoid using 1 byte per bool
error apply(std::vector<bool>& x) {
auto len = x.size();
auto err = begin_sequence(len);
if (err || len == 0)
return err;
struct {
size_t len;
void operator()(std::vector<bool>& lhs, std::vector<uint8_t>& rhs) const {
lhs.resize(len, false);
size_t cpt = 0;
for (auto v: rhs) {
for (int k = 0; k < 8; ++k) {
lhs[cpt] = ((v & (1 << k)) != 0);
if (++cpt >= len)
return;
}
}
}
void operator()(std::vector<uint8_t>& lhs, std::vector<bool>& rhs) const {
size_t k = 0;
lhs.resize((rhs.size() - 1) / 8 + 1, 0);
for (bool b: rhs) {
if (b)
lhs[k / 8] |= static_cast<uint8_t>(1 << (k % 8));
++k;
}
}
} assign;
assign.len = len;
std::vector<uint8_t> tmp;
return convert_apply(dref(), x, tmp, assign);
}
template <class T>
error consume_range(T& xs) {
for (auto& x : xs) {
using value_type = typename std::remove_reference<decltype(x)>::type;
using mutable_type = typename std::remove_const<value_type>::type;
if (auto err = dref().apply(const_cast<mutable_type&>(x)))
return err;
}
return none;
}
/// Converts each element in `xs` to `U` before calling `apply`.
template <class U, class T>
error consume_range_c(T& xs) {
for (U x : xs) {
if (auto err = dref().apply(x))
return err;
}
return none;
}
template <class T>
error fill_range(T& xs, size_t num_elements) {
xs.clear();
auto insert_iter = std::inserter(xs, xs.end());
for (size_t i = 0; i < num_elements; ++i) {
typename std::remove_const<typename T::value_type>::type x;
if (auto err = dref().apply(x))
return err;
*insert_iter++ = std::move(x);
}
return none;
}
/// Loads elements from type `U` before inserting to `xs`.
template <class U, class T>
error fill_range_c(T& xs, size_t num_elements) {
xs.clear();
auto insert_iter = std::inserter(xs, xs.end());
for (size_t i = 0; i < num_elements; ++i) {
U x;
if (auto err = dref().apply(x))
return err;
*insert_iter++ = std::move(x);
}
return none;
}
// Applies this processor as Derived to `xs` in saving mode.
template <class D, class T>
static typename std::enable_if<
D::reads_state && !detail::is_byte_sequence<T>::value,
error
>::type
apply_sequence(D& self, T& xs) {
auto s = xs.size();
return error::eval([&] { return self.begin_sequence(s); },
[&] { return self.consume_range(xs); },
[&] { return self.end_sequence(); });
}
// Applies this processor as Derived to `xs` in loading mode.
template <class D, class T>
static typename std::enable_if<
!D::reads_state && !detail::is_byte_sequence<T>::value,
error
>::type
apply_sequence(D& self, T& xs) {
size_t s;
return error::eval([&] { return self.begin_sequence(s); },
[&] { return self.fill_range(xs, s); },
[&] { return self.end_sequence(); });
}
// Optimized saving for contiguous byte sequences.
template <class D, class T>
static typename std::enable_if<
D::reads_state && detail::is_byte_sequence<T>::value,
error
>::type
apply_sequence(D& self, T& xs) {
auto s = xs.size();
return error::eval([&] { return self.begin_sequence(s); },
[&] { return s > 0 ? self.apply_raw(xs.size(), &xs[0])
: none; },
[&] { return self.end_sequence(); });
}
// Optimized loading for contiguous byte sequences.
template <class D, class T>
static typename std::enable_if<
!D::reads_state && detail::is_byte_sequence<T>::value,
error
>::type
apply_sequence(D& self, T& xs) {
size_t s;
return error::eval([&] { return self.begin_sequence(s); },
[&] { xs.resize(s);
return s > 0 ? self.apply_raw(s, &xs[0])
: none; },
[&] { return self.end_sequence(); });
}
/// Applies this processor to a sequence of values.
template <class T>
typename std::enable_if<
detail::is_iterable<T>::value
&& !detail::has_serialize<T>::value
&& !detail::is_inspectable<Derived, T>::value,
error
>::type
apply(T& xs) {
return apply_sequence(dref(), xs);
}
/// Applies this processor to an array.
template <class T, size_t S>
typename std::enable_if<detail::is_serializable<T>::value, error>::type
apply(std::array<T, S>& xs) {
return consume_range(xs);
}
/// Applies this processor to an array.
template <class T, size_t S>
typename std::enable_if<detail::is_serializable<T>::value, error>::type
apply(T (&xs) [S]) {
return consume_range(xs);
}
template <class F, class S>
typename std::enable_if<
detail::is_serializable<typename std::remove_const<F>::type>::value
&& detail::is_serializable<S>::value,
error
>::type
apply(std::pair<F, S>& xs) {
using t0 = typename std::remove_const<F>::type;
// This cast allows the data processor to cope with
// `pair<const K, V>` value types used by `std::map`.
if (auto err = dref().apply(const_cast<t0&>(xs.first)))
return err;
return dref().apply(xs.second);
}
template <class... Ts>
typename std::enable_if<
detail::conjunction<
detail::is_serializable<Ts>::value...
>::value,
error
>::type
apply(std::tuple<Ts...>& xs) {
//apply_helper f{*this};
return detail::apply_args(*this, detail::get_indices(xs), xs);
}
template <class T>
typename std::enable_if<
!std::is_empty<T>::value
&& detail::has_serialize<T>::value,
error
>::type
apply(T& x) {
// throws on error
detail::delegate_serialize(dref(), x);
return none;
}
template <class Rep, class Period>
typename std::enable_if<std::is_integral<Rep>::value, error>::type
apply(std::chrono::duration<Rep, Period>& x) {
using duration_type = std::chrono::duration<Rep, Period>;
// always save/store durations as int64_t to work around possibly
// different integer types on different platforms for standard typedefs
struct {
void operator()(duration_type& lhs, Rep& rhs) const {
duration_type tmp{rhs};
lhs = tmp;
}
void operator()(Rep& lhs, duration_type& rhs) const {
lhs = rhs.count();
}
} assign;
Rep tmp;
return convert_apply(dref(), x, tmp, assign);
}
template <class Rep, class Period>
typename std::enable_if<std::is_floating_point<Rep>::value, error>::type
apply(std::chrono::duration<Rep, Period>& x) {
using duration_type = std::chrono::duration<Rep, Period>;
// always save/store floating point durations
// as doubles for the same reason as above
struct {
void operator()(duration_type& lhs, double& rhs) const {
duration_type tmp{rhs};
lhs = tmp;
}
void operator()(double& lhs, duration_type& rhs) const {
lhs = rhs.count();
}
} assign;
double tmp;
return convert_apply(dref(), x, tmp, assign);
}
template <class Duration>
error apply(std::chrono::time_point<std::chrono::system_clock, Duration>& t) {
if (Derived::reads_state) {
auto dur = t.time_since_epoch();
return apply(dur);
}
if (Derived::writes_state) {
Duration dur{};
auto e = apply(dur);
t = std::chrono::time_point<std::chrono::system_clock, Duration>{dur};
return e;
}
}
/// Applies this processor to a raw block of data of size `num_bytes`.
virtual error apply_raw(size_t num_bytes, void* data) = 0;
template <class T>
typename std::enable_if<
detail::is_inspectable<Derived, T>::value
&& !detail::has_serialize<T>::value,
decltype(inspect(std::declval<Derived&>(), std::declval<T&>()))
>::type
apply(T& x) {
return inspect(dref(), x);
}
// -- operator() -------------------------------------------------------------
error operator()() {
return none;
}
#if defined(__cpp_fold_expressions) && defined(__cpp_if_constexpr)
template <class... Ts>
error operator()(Ts&&... xs) {
error result;
auto f = [&result, this](auto&& x) {
using type = detail::decay_t<decltype(x)>;
if constexpr (meta::is_save_callback<type>::value) {
if constexpr (Derived::reads_state)
if (auto err = x.fun()) {
result = std::move(err);
return false;
}
} else if constexpr (meta::is_load_callback<type>::value) {
if constexpr (Derived::writes_state)
if (auto err = x.fun()) {
result = std::move(err);
return false;
}
} else if constexpr (meta::is_annotation<type>::value
|| is_allowed_unsafe_message_type<type>::value) {
// skip element
} else {
if (auto err = dref().apply(deconst(x))) {
result = std::move(err);
return false;
}
}
return true;
};
if ((f(std::forward<Ts>(xs)) && ...))
return none;
return result;
}
#else // defined(__cpp_fold_expressions) && defined(__cpp_if_constexpr)
template <class F, class... Ts>
error operator()(meta::save_callback_t<F> x, Ts&&... xs) {
// TODO: use `if constexpr` when switching to C++17.
if (Derived::reads_state)
if (auto err = x.fun())
return err;
return dref()(std::forward<Ts>(xs)...);
}
template <class F, class... Ts>
error operator()(meta::load_callback_t<F> x, Ts&&... xs) {
// TODO: use `if constexpr` when switching to C++17.
if (Derived::writes_state)
if (auto err = x.fun())
return err;
return dref()(std::forward<Ts>(xs)...);
}
template <class... Ts>
error operator()(const meta::annotation&, Ts&&... xs) {
return dref()(std::forward<Ts>(xs)...);
}
template <class T, class... Ts>
typename std::enable_if<
is_allowed_unsafe_message_type<T>::value,
error
>::type
operator()(const T&, Ts&&... xs) {
return dref()(std::forward<Ts>(xs)...);
}
template <class T, class... Ts>
typename std::enable_if<
!meta::is_annotation<T>::value
&& !is_allowed_unsafe_message_type<T>::value,
error
>::type
operator()(T&& x, Ts&&... xs) {
static_assert(Derived::reads_state
|| (!std::is_rvalue_reference<T&&>::value
&& !std::is_const<
typename std::remove_reference<T>::type
>::value),
"a loading inspector requires mutable lvalue references");
if (auto err = apply(deconst(x)))
return err;
return dref()(std::forward<Ts>(xs)...);
}
#endif // defined(__cpp_fold_expressions) && defined(__cpp_if_constexpr)
protected:
virtual error apply_impl(int8_t&) = 0;
virtual error apply_impl(uint8_t&) = 0;
virtual error apply_impl(int16_t&) = 0;
virtual error apply_impl(uint16_t&) = 0;
virtual error apply_impl(int32_t&) = 0;
virtual error apply_impl(uint32_t&) = 0;
virtual error apply_impl(int64_t&) = 0;
virtual error apply_impl(uint64_t&) = 0;
virtual error apply_impl(float&) = 0;
virtual error apply_impl(double&) = 0;
virtual error apply_impl(long double&) = 0;
virtual error apply_impl(std::string&) = 0;
virtual error apply_impl(std::u16string&) = 0;
virtual error apply_impl(std::u32string&) = 0;
private:
template <class T>
T& deconst(const T& x) {
return const_cast<T&>(x);
}
template <class D, class T, class U, class F>
static typename std::enable_if<D::reads_state, error>::type
convert_apply(D& self, T& x, U& storage, F assign) {
assign(storage, x);
return self(storage);
}
template <class D, class T, class U, class F>
static typename std::enable_if<!D::reads_state, error>::type
convert_apply(D& self, T& x, U& storage, F assign) {
if (auto err = self(storage))
return err;
assign(x, storage);
return none;
}
// Returns a reference to the derived type.
Derived& dref() {
return *static_cast<Derived*>(this);
}
execution_unit* context_;
};
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#pragma once
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <iomanip>
#include <limits>
#include <sstream>
#include <stdexcept>
#include <string>
#include <type_traits>
#include "caf/deserializer.hpp"
#include "caf/detail/ieee_754.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
namespace caf {
/// Implements the deserializer interface with a binary serialization protocol.
template <class Streambuf>
class stream_deserializer : public deserializer {
using streambuf_type = typename std::remove_reference<Streambuf>::type;
using char_type = typename streambuf_type::char_type;
using streambuf_base = std::basic_streambuf<char_type>;
using streamsize = std::streamsize;
static_assert(std::is_base_of<streambuf_base, streambuf_type>::value,
"Streambuf must inherit from std::streambuf");
public:
template <class... Ts>
explicit stream_deserializer(actor_system& sys, Ts&&... xs)
: deserializer(sys), streambuf_(std::forward<Ts>(xs)...) {
}
template <class... Ts>
explicit stream_deserializer(execution_unit* ctx, Ts&&... xs)
: deserializer(ctx), streambuf_(std::forward<Ts>(xs)...) {
}
template <class S,
class = typename std::enable_if<std::is_same<
typename std::remove_reference<S>::type,
typename std::remove_reference<Streambuf>::type>::value>::type>
explicit stream_deserializer(S&& sb)
: deserializer(nullptr), streambuf_(std::forward<S>(sb)) {
}
error begin_object(uint16_t& typenr, std::string& name) override {
return error::eval([&] { return apply_int(typenr); },
[&] { return typenr == 0 ? apply(name) : error{}; });
}
error end_object() override {
return none;
}
error begin_sequence(size_t& num_elements) override {
// We serialize a `size_t` always in 32-bit, to guarantee compatibility
// with 32-bit nodes in the network.
// TODO: protect with `if constexpr (sizeof(size_t) > sizeof(uint32_t))`
// when switching to C++17 and pass `num_elements` directly to
// `varbyte_decode` in the `else` case
uint32_t x;
auto result = varbyte_decode(x);
if (!result)
num_elements = static_cast<size_t>(x);
return result;
}
error end_sequence() override {
return none;
}
error apply_raw(size_t num_bytes, void* data) override {
return range_check(streambuf_.sgetn(reinterpret_cast<char_type*>(data),
static_cast<streamsize>(num_bytes)),
num_bytes);
}
protected:
// Decode an unsigned integral type as variable-byte-encoded byte sequence.
template <class T>
error varbyte_decode(T& x) {
static_assert(std::is_unsigned<T>::value, "T must be an unsigned type");
auto n = 0;
x = 0;
uint8_t low7;
do {
auto c = streambuf_.sbumpc();
using traits = typename streambuf_type::traits_type;
if (traits::eq_int_type(c, traits::eof()))
return sec::end_of_stream;
low7 = static_cast<uint8_t>(traits::to_char_type(c));
x |= static_cast<T>((low7 & 0x7F)) << (7 * n);
++n;
} while (low7 & 0x80);
return none;
}
error apply_impl(int8_t& x) override {
return apply_raw(sizeof(int8_t), &x);
}
error apply_impl(uint8_t& x) override {
return apply_raw(sizeof(uint8_t), &x);
}
error apply_impl(int16_t& x) override {
return apply_int(x);
}
error apply_impl(uint16_t& x) override {
return apply_int(x);
}
error apply_impl(int32_t& x) override {
return apply_int(x);
}
error apply_impl(uint32_t& x) override {
return apply_int(x);
}
error apply_impl(int64_t& x) override {
return apply_int(x);
}
error apply_impl(uint64_t& x) override {
return apply_int(x);
}
error apply_impl(float& x) override {
return apply_float(x);
}
error apply_impl(double& x) override {
return apply_float(x);
}
error apply_impl(long double& x) override {
// The IEEE-754 conversion does not work for long double
// => fall back to string serialization (even though it sucks).
std::string tmp;
if (auto err = apply(tmp))
return err;
std::istringstream iss{std::move(tmp)};
iss >> x;
return none;
}
error apply_impl(std::string& x) override {
size_t str_size;
if (auto err = begin_sequence(str_size))
return err;
x.resize(str_size);
auto s = static_cast<streamsize>(str_size);
auto data = reinterpret_cast<char_type*>(&x[0]);
if (auto err = range_check(streambuf_.sgetn(data, s), str_size))
return err;
return end_sequence();
}
error apply_impl(std::u16string& x) override {
size_t str_size;
return error::eval([&] { return begin_sequence(str_size); },
[&] { return fill_range_c<uint16_t>(x, str_size); },
[&] { return end_sequence(); });
}
error apply_impl(std::u32string& x) override {
size_t str_size;
return error::eval([&] { return begin_sequence(str_size); },
[&] { return fill_range_c<uint32_t>(x, str_size); },
[&] { return end_sequence(); });
}
error range_check(std::streamsize got, size_t need) {
if (got >= 0 && static_cast<size_t>(got) == need)
return none;
CAF_LOG_ERROR("range_check failed");
return sec::end_of_stream;
}
template <class T>
error apply_int(T& x) {
typename std::make_unsigned<T>::type tmp = 0;
if (auto err = apply_raw(sizeof(T), &tmp))
return err;
x = static_cast<T>(detail::from_network_order(tmp));
return none;
}
template <class T>
error apply_float(T& x) {
typename detail::ieee_754_trait<T>::packed_type tmp = 0;
if (auto err = apply_int(tmp))
return err;
x = detail::unpack754(tmp);
return none;
}
private:
Streambuf streambuf_;
};
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#pragma once
#include <cstddef>
#include <cstdint>
#include <iomanip>
#include <limits>
#include <sstream>
#include <streambuf>
#include <string>
#include <type_traits>
#include "caf/config.hpp"
#include "caf/detail/ieee_754.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/sec.hpp"
#include "caf/serializer.hpp"
#include "caf/streambuf.hpp"
namespace caf {
/// Implements the serializer interface with a binary serialization protocol.
template <class Streambuf>
class stream_serializer : public serializer {
using streambuf_type = typename std::remove_reference<Streambuf>::type;
using char_type = typename streambuf_type::char_type;
using streambuf_base = std::basic_streambuf<char_type>;
static_assert(std::is_base_of<streambuf_base, streambuf_type>::value,
"Streambuf must inherit from std::streambuf");
public:
template <class... Ts>
explicit stream_serializer(actor_system& sys, Ts&&... xs)
: serializer(sys), streambuf_{std::forward<Ts>(xs)...} {
}
template <class... Ts>
explicit stream_serializer(execution_unit* ctx, Ts&&... xs)
: serializer(ctx), streambuf_{std::forward<Ts>(xs)...} {
}
template <class S,
class = typename std::enable_if<std::is_same<
typename std::remove_reference<S>::type,
typename std::remove_reference<Streambuf>::type>::value>::type>
explicit stream_serializer(S&& sb)
: serializer(nullptr), streambuf_(std::forward<S>(sb)) {
}
error begin_object(uint16_t& typenr, std::string& name) override {
return error::eval([&] { return apply(typenr); },
[&] { return typenr == 0 ? apply(name) : error{}; });
}
error end_object() override {
return none;
}
error begin_sequence(size_t& list_size) override {
// TODO: protect with `if constexpr (sizeof(size_t) > sizeof(uint32_t))`
// when switching to C++17
CAF_ASSERT(list_size <= std::numeric_limits<uint32_t>::max());
// Serialize a `size_t` always in 32-bit, to guarantee compatibility with
// 32-bit nodes in the network.
return varbyte_encode(static_cast<uint32_t>(list_size));
}
error end_sequence() override {
return none;
}
error apply_raw(size_t num_bytes, void* data) override {
auto ssize = static_cast<std::streamsize>(num_bytes);
auto n = streambuf_.sputn(reinterpret_cast<char_type*>(data), ssize);
if (n != ssize)
return sec::end_of_stream;
return none;
}
protected:
// Encode an unsigned integral type as variable-byte sequence.
template <class T>
error varbyte_encode(T x) {
static_assert(std::is_unsigned<T>::value, "T must be an unsigned type");
// For 64-bit values, the encoded representation cannot get larger than 10
// bytes. A scratch space of 16 bytes suffices as upper bound.
uint8_t buf[16];
auto i = buf;
while (x > 0x7f) {
*i++ = (static_cast<uint8_t>(x) & 0x7f) | 0x80;
x >>= 7;
}
*i++ = static_cast<uint8_t>(x) & 0x7f;
auto res = streambuf_.sputn(reinterpret_cast<char_type*>(buf),
static_cast<std::streamsize>(i - buf));
if (res != (i - buf))
return sec::end_of_stream;
return none;
}
error apply_impl(int8_t& x) override {
return apply_raw(sizeof(int8_t), &x);
}
error apply_impl(uint8_t& x) override {
return apply_raw(sizeof(uint8_t), &x);
}
error apply_impl(int16_t& x) override {
return apply_int(x);
}
error apply_impl(uint16_t& x) override {
return apply_int(x);
}
error apply_impl(int32_t& x) override {
return apply_int(x);
}
error apply_impl(uint32_t& x) override {
return apply_int(x);
}
error apply_impl(int64_t& x) override {
return apply_int(x);
}
error apply_impl(uint64_t& x) override {
return apply_int(x);
}
error apply_impl(float& x) override {
return apply_int(detail::pack754(x));
}
error apply_impl(double& x) override {
return apply_int(detail::pack754(x));
}
error apply_impl(long double& x) override {
// The IEEE-754 conversion does not work for long double
// => fall back to string serialization (event though it sucks).
std::ostringstream oss;
oss << std::setprecision(std::numeric_limits<long double>::digits) << x;
auto tmp = oss.str();
return apply_impl(tmp);
}
error apply_impl(std::string& x) override {
auto str_size = x.size();
auto data = const_cast<char*>(x.c_str());
return error::eval([&] { return begin_sequence(str_size); },
[&] { return apply_raw(x.size(), data); },
[&] { return end_sequence(); });
}
error apply_impl(std::u16string& x) override {
auto str_size = x.size();
return error::eval([&] { return begin_sequence(str_size); },
[&] { return consume_range_c<uint16_t>(x); },
[&] { return end_sequence(); });
}
error apply_impl(std::u32string& x) override {
auto str_size = x.size();
return error::eval([&] { return begin_sequence(str_size); },
[&] { return consume_range_c<uint32_t>(x); },
[&] { return end_sequence(); });
}
template <class T>
error apply_int(T x) {
using unsigned_type = typename std::make_unsigned<T>::type;
auto y = detail::to_network_order(static_cast<unsigned_type>(x));
return apply_raw(sizeof(T), &y);
}
private:
Streambuf streambuf_;
};
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#pragma once
#include <algorithm>
#include <cstddef>
#include <cstring>
#include <limits>
#include <streambuf>
#include <type_traits>
#include <vector>
#include "caf/config.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf {
/// The base class for all stream buffer implementations.
template <class CharT = char, class Traits = std::char_traits<CharT>>
class stream_buffer : public std::basic_streambuf<CharT, Traits> {
public:
using base = std::basic_streambuf<CharT, Traits>;
using pos_type = typename base::pos_type;
using off_type = typename base::off_type;
protected:
/// The standard only defines pbump(int), which can overflow on 64-bit
/// architectures. All stream buffer implementations should therefore use
/// these function instead. For a detailed discussion, see:
/// https://gcc.gnu.org/bugzilla/show_bug.cgi?id=47921
template <class T = int>
typename std::enable_if<sizeof(T) == 4>::type safe_pbump(std::streamsize n) {
while (n > std::numeric_limits<int>::max()) {
this->pbump(std::numeric_limits<int>::max());
n -= std::numeric_limits<int>::max();
}
this->pbump(static_cast<int>(n));
}
template <class T = int>
typename std::enable_if<sizeof(T) == 8>::type safe_pbump(std::streamsize n) {
this->pbump(static_cast<int>(n));
}
// As above, but for the get area.
template <class T = int>
typename std::enable_if<sizeof(T) == 4>::type safe_gbump(std::streamsize n) {
while (n > std::numeric_limits<int>::max()) {
this->gbump(std::numeric_limits<int>::max());
n -= std::numeric_limits<int>::max();
}
this->gbump(static_cast<int>(n));
}
template <class T = int>
typename std::enable_if<sizeof(T) == 8>::type safe_gbump(std::streamsize n) {
this->gbump(static_cast<int>(n));
}
pos_type default_seekoff(off_type off, std::ios_base::seekdir dir,
std::ios_base::openmode which) {
auto new_off = pos_type(off_type(-1));
auto get = (which & std::ios_base::in) == std::ios_base::in;
auto put = (which & std::ios_base::out) == std::ios_base::out;
if (!(get || put))
return new_off; // nothing to do
if (get) {
switch (dir) {
default:
return pos_type(off_type(-1));
case std::ios_base::beg:
new_off = 0;
break;
case std::ios_base::cur:
new_off = this->gptr() - this->eback();
break;
case std::ios_base::end:
new_off = this->egptr() - this->eback();
break;
}
new_off += off;
this->setg(this->eback(), this->eback() + new_off, this->egptr());
}
if (put) {
switch (dir) {
default:
return pos_type(off_type(-1));
case std::ios_base::beg:
new_off = 0;
break;
case std::ios_base::cur:
new_off = this->pptr() - this->pbase();
break;
case std::ios_base::end:
new_off = this->egptr() - this->pbase();
break;
}
new_off += off;
this->setp(this->pbase(), this->epptr());
safe_pbump(new_off);
}
return new_off;
}
pos_type default_seekpos(pos_type pos, std::ios_base::openmode which) {
auto get = (which & std::ios_base::in) == std::ios_base::in;
auto put = (which & std::ios_base::out) == std::ios_base::out;
if (!(get || put))
return pos_type(off_type(-1)); // nothing to do
if (get)
this->setg(this->eback(), this->eback() + pos, this->egptr());
if (put) {
this->setp(this->pbase(), this->epptr());
safe_pbump(pos);
}
return pos;
}
};
/// A streambuffer abstraction over a fixed array of bytes. This streambuffer
/// cannot overflow/underflow. Once it has reached its end, attempts to read
/// characters will return `trait_type::eof`.
template <class CharT = char, class Traits = std::char_traits<CharT>>
class arraybuf : public stream_buffer<CharT, Traits> {
public:
using base = std::basic_streambuf<CharT, Traits>;
using char_type = typename base::char_type;
using traits_type = typename base::traits_type;
using int_type = typename base::int_type;
using pos_type = typename base::pos_type;
using off_type = typename base::off_type;
/// Constructs an array streambuffer from a container.
/// @param c A contiguous container.
/// @pre `c.data()` must point to a contiguous sequence of characters having
/// length `c.size()`.
template <class Container,
class = typename std::enable_if<
detail::has_data_member<Container>::value
&& detail::has_size_member<Container>::value>::type>
arraybuf(Container& c)
: arraybuf(const_cast<char_type*>(c.data()), c.size()) {
// nop
}
/// Constructs an array streambuffer from a raw character sequence.
/// @param data A pointer to the first character.
/// @param size The length of the character sequence.
arraybuf(char_type* data, size_t size) {
setbuf(data, static_cast<std::streamsize>(size));
}
// There exists a bug in libstdc++ version < 5: the implementation does not
// provide the necessary move constructors, so we have to roll our own :-/.
// See https://gcc.gnu.org/bugzilla/show_bug.cgi?id=54316 for details.
// TODO: remove after having raised the minimum GCC version to 5.
arraybuf(arraybuf&& other) {
this->setg(other.eback(), other.gptr(), other.egptr());
this->setp(other.pptr(), other.epptr());
other.setg(nullptr, nullptr, nullptr);
other.setp(nullptr, nullptr);
}
// TODO: remove after having raised the minimum GCC version to 5.
arraybuf& operator=(arraybuf&& other) {
this->setg(other.eback(), other.gptr(), other.egptr());
this->setp(other.pptr(), other.epptr());
other.setg(nullptr, nullptr, nullptr);
other.setp(nullptr, nullptr);
return *this;
}
protected:
// -- positioning ----------------------------------------------------------
std::basic_streambuf<char_type, Traits>*
setbuf(char_type* s, std::streamsize n) override {
this->setg(s, s, s + n);
this->setp(s, s + n);
return this;
}
pos_type
seekpos(pos_type pos, std::ios_base::openmode which
= std::ios_base::in | std::ios_base::out) override {
return this->default_seekpos(pos, which);
}
pos_type seekoff(off_type off, std::ios_base::seekdir dir,
std::ios_base::openmode which) override {
return this->default_seekoff(off, dir, which);
}
// -- put area -------------------------------------------------------------
std::streamsize xsputn(const char_type* s, std::streamsize n) override {
auto available = this->epptr() - this->pptr();
auto actual = std::min(n, static_cast<std::streamsize>(available));
std::memcpy(this->pptr(), s,
static_cast<size_t>(actual) * sizeof(char_type));
this->safe_pbump(actual);
return actual;
}
// -- get area -------------------------------------------------------------
std::streamsize xsgetn(char_type* s, std::streamsize n) override {
auto available = this->egptr() - this->gptr();
auto actual = std::min(n, static_cast<std::streamsize>(available));
std::memcpy(s, this->gptr(),
static_cast<size_t>(actual) * sizeof(char_type));
this->safe_gbump(actual);
return actual;
}
};
/// A streambuffer abstraction over a contiguous container. It supports
/// reading in the same style as `arraybuf`, but is unbounded for output.
template <class Container>
class containerbuf
: public stream_buffer<typename Container::value_type,
std::char_traits<typename Container::value_type>> {
public:
using base
= std::basic_streambuf<typename Container::value_type,
std::char_traits<typename Container::value_type>>;
using char_type = typename base::char_type;
using traits_type = typename base::traits_type;
using int_type = typename base::int_type;
using pos_type = typename base::pos_type;
using off_type = typename base::off_type;
/// Constructs a container streambuf.
/// @param c A contiguous container.
template <class C = Container, class = typename std::enable_if<
detail::has_data_member<C>::value
&& detail::has_size_member<C>::value>::type>
containerbuf(Container& c) : container_(c) {
// We use a const_cast because C++11 doesn't provide a non-const data()
// overload. Using std::data(c) would be the right way to write this.
auto data = const_cast<char_type*>(c.data());
auto size = static_cast<std::streamsize>(c.size());
setbuf(data, size);
}
// See note in arraybuf(arraybuf&&).
// TODO: remove after having raised the minimum GCC version to 5.
containerbuf(containerbuf&& other) : container_(other.container_) {
this->setg(other.eback(), other.gptr(), other.egptr());
other.setg(nullptr, nullptr, nullptr);
}
// TODO: remove after having raised the minimum GCC version to 5.
containerbuf& operator=(containerbuf&& other) {
this->setg(other.eback(), other.gptr(), other.egptr());
other.setg(nullptr, nullptr, nullptr);
return *this;
}
// Hides base-class implementation to simplify single-character lookup.
int_type sgetc() {
if (this->gptr() == this->egptr())
return traits_type::eof();
return traits_type::to_int_type(*this->gptr());
}
// Hides base-class implementation to simplify single-character insert
// without a put area.
int_type sputc(char_type c) {
container_.push_back(c);
return c;
}
// Hides base-class implementation to simplify multi-character insert without
// a put area.
std::streamsize sputn(const char_type* s, std::streamsize n) {
return xsputn(s, n);
}
protected:
// -- positioning ----------------------------------------------------------
base* setbuf(char_type* s, std::streamsize n) override {
this->setg(s, s, s + n);
return this;
}
pos_type
seekpos(pos_type pos, std::ios_base::openmode which
= std::ios_base::in | std::ios_base::out) override {
// We only have a get area, so no put area (= out) operations.
if ((which & std::ios_base::out) == std::ios_base::out)
return pos_type(off_type(-1));
return this->default_seekpos(pos, which);
}
pos_type seekoff(off_type off, std::ios_base::seekdir dir,
std::ios_base::openmode which) override {
// We only have a get area, so no put area (= out) operations.
if ((which & std::ios_base::out) == std::ios_base::out)
return pos_type(off_type(-1));
return this->default_seekoff(off, dir, which);
}
// Synchronizes the get area with the underlying buffer.
int sync() override {
// See note in ctor for const_cast
auto data = const_cast<char_type*>(container_.data());
auto size = static_cast<std::streamsize>(container_.size());
setbuf(data, size);
return 0;
}
// -- get area -------------------------------------------------------------
// We can't get obtain more characters on underflow, so we only optimize
// multi-character sequential reads.
std::streamsize xsgetn(char_type* s, std::streamsize n) override {
auto available = this->egptr() - this->gptr();
auto actual = std::min(n, static_cast<std::streamsize>(available));
std::memcpy(s, this->gptr(),
static_cast<size_t>(actual) * sizeof(char_type));
this->safe_gbump(actual);
return actual;
}
// -- put area -------------------------------------------------------------
// Should never get called, because there is always space in the buffer.
// (But just in case, it does the same thing as sputc.)
int_type overflow(int_type c = traits_type::eof()) final {
if (!traits_type::eq_int_type(c, traits_type::eof()))
container_.push_back(traits_type::to_char_type(c));
return c;
}
std::streamsize xsputn(const char_type* s, std::streamsize n) override {
container_.insert(container_.end(), s, s + n);
return n;
}
private:
Container& container_;
};
using charbuf = arraybuf<char>;
using vectorbuf = containerbuf<std::vector<char>>;
} // namespace caf
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