Commit 024f037a authored by Matthias Vallentin's avatar Matthias Vallentin

Make (de)serializer streambuffer based

First, this change moves the context to the parameter to the end in
order to make it optional. The context is only necessary when
serializing messages. Since CAF's serialization framework also serves
also other application's need, this CAF-specific detail now no longer
constitutes a mandatory part of the API.

Second, this change switches the binary (de)serializer to a streambuffer
based API. This enables much more flexible forms of interfacing, e.g.,
by serializing into/from a file, socket, and more.
parent 21670aa2
......@@ -8,8 +8,6 @@
#include <iostream>
#include "caf/all.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp"
using std::cout;
using std::cerr;
......
......@@ -33,8 +33,6 @@ set (LIBCAF_CORE_SRCS
src/behavior.cpp
src/behavior_stack.cpp
src/behavior_impl.cpp
src/binary_deserializer.cpp
src/binary_serializer.cpp
src/blocking_actor.cpp
src/concatenated_tuple.cpp
src/continue_helper.cpp
......
......@@ -39,6 +39,7 @@
#include "caf/duration.hpp"
#include "caf/exception.hpp"
#include "caf/resumable.hpp"
#include "caf/streambuf.hpp"
#include "caf/actor_addr.hpp"
#include "caf/actor_pool.hpp"
#include "caf/attachable.hpp"
......
......@@ -20,57 +20,14 @@
#ifndef CAF_BINARY_DESERIALIZER_HPP
#define CAF_BINARY_DESERIALIZER_HPP
#include <string>
#include <cstddef>
#include <utility>
#include "caf/deserializer.hpp"
#include "caf/stream_deserializer.hpp"
#include "caf/streambuf.hpp"
namespace caf {
/// Implements the deserializer interface with a binary serialization protocol.
class binary_deserializer : public deserializer {
public:
binary_deserializer(actor_system& sys, const void* buf, size_t buf_size);
binary_deserializer(execution_unit* ctx, const void* buf, size_t buf_size);
binary_deserializer(actor_system& ctx, const void* first, const void* last);
binary_deserializer(execution_unit* ctx, const void* first, const void* last);
template <class C, class T>
binary_deserializer(C&& ctx, const T& xs)
: binary_deserializer(std::forward<C>(ctx), xs.data(), xs.size()) {
// nop
}
/// Replaces the current read buffer.
void set_rdbuf(const void* buf, size_t buf_size);
/// Replaces the current read buffer.
void set_rdbuf(const void* first, const void* last);
/// Returns whether this deserializer has reached the end of its buffer.
bool at_end() const;
/// Compares the next `num_bytes` from the underlying buffer to `buf`
/// with same semantics as `strncmp(this->pos_, buf, num_bytes) == 0`.
bool buf_equals(const void* buf, size_t num_bytes);
/// Moves the current read position in the buffer by `num_bytes`.
binary_deserializer& advance(ptrdiff_t num_bytes);
void begin_object(uint16_t& nr, std::string& name) override;
void end_object() override;
void begin_sequence(size_t& num_elements) override;
void end_sequence() override;
void apply_raw(size_t num_bytes, void* data) override;
protected:
void apply_builtin(builtin in_out_type, void* in_out) override;
private:
const void* pos_;
const void* end_;
};
/// A stream serializer that writes into an unbounded contiguous character
/// sequence.
using binary_deserializer = stream_deserializer<charbuf>;
} // namespace caf
......
......@@ -20,98 +20,14 @@
#ifndef CAF_BINARY_SERIALIZER_HPP
#define CAF_BINARY_SERIALIZER_HPP
#include <utility>
#include <sstream>
#include <iomanip>
#include <iterator>
#include <functional>
#include <type_traits>
#include "caf/fwd.hpp"
#include "caf/serializer.hpp"
#include "caf/primitive_variant.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/stream_serializer.hpp"
#include "caf/streambuf.hpp"
namespace caf {
/// Implements the serializer interface with a binary serialization protocol.
class binary_serializer : public serializer {
public:
/// Creates a binary serializer writing to given iterator position.
template <class C,
class Out,
class E =
typename std::enable_if<
std::is_same<
typename std::iterator_traits<Out>::iterator_category,
std::output_iterator_tag
>::value
|| std::is_same<
typename std::iterator_traits<Out>::iterator_category,
std::random_access_iterator_tag
>::value
>::type>
binary_serializer(C&& ctx, Out it) : serializer(std::forward<C>(ctx)) {
reset_iter(it);
}
template <class C,
class T,
class E =
typename std::enable_if<
detail::has_char_insert<T>::value
>::type>
binary_serializer(C&& ctx, T& out) : serializer(std::forward<C>(ctx)) {
reset_container(out);
}
using write_fun = std::function<void(const char*, size_t)>;
void begin_object(uint16_t& typenr, std::string& name) override;
void end_object() override;
void begin_sequence(size_t& list_size) override;
void end_sequence() override;
void apply_raw(size_t num_bytes, void* data) override;
void apply_builtin(builtin in_out_type, void* in_out) override;
template <class OutIter>
void reset_iter(OutIter iter) {
struct fun {
fun(OutIter pos) : pos_(pos) {
// nop
}
void operator()(const char* first, size_t num_bytes) {
pos_ = std::copy(first, first + num_bytes, pos_);
}
OutIter pos_;
};
out_ = fun{iter};
}
template <class T>
void reset_container(T& x) {
struct fun {
fun(T& container) : x_(container) {
// nop
}
void operator()(const char* first, size_t num_bytes) {
auto last = first + num_bytes;
x_.insert(x_.end(), first, last);
}
T& x_;
};
out_ = fun{x};
}
private:
write_fun out_;
};
/// A stream serializer that writes into an unbounded contiguous character
/// sequence.
using binary_serializer = stream_serializer<vectorbuf>;
} // namespace caf
......
......@@ -44,7 +44,7 @@ public:
data_processor(const data_processor&) = delete;
data_processor& operator=(const data_processor&) = delete;
data_processor(execution_unit* ctx) : context_(ctx) {
data_processor(execution_unit* ctx = nullptr) : context_(ctx) {
// nop
}
......
......@@ -42,7 +42,7 @@ public:
explicit deserializer(actor_system& sys);
explicit deserializer(execution_unit* ctx);
explicit deserializer(execution_unit* ctx = nullptr);
};
/// Reads `x` from `source`.
......
......@@ -613,6 +613,30 @@ public:
static constexpr bool value = false;
};
template <class T>
struct has_data_member {
template <class U>
static auto test(U* x) -> decltype(x->data(), std::true_type());
template <class U>
static auto test(...) -> std::false_type;
using type = decltype(test<T>(nullptr));
static constexpr bool value = type::value;
};
template <class T>
struct has_size_member {
template <class U>
static auto test(U* x) -> decltype(x->size(), std::true_type());
template <class U>
static auto test(...) -> std::false_type;
using type = decltype(test<T>(nullptr));
static constexpr bool value = type::value;
};
/// Checks whether T is convertible to either `std::function<void (T&)>`
/// or `std::function<void (const T&)>`.
template <class F, class T>
......
......@@ -74,12 +74,10 @@ class continue_helper;
class mailbox_element;
class message_handler;
class response_promise;
class binary_serializer;
class event_based_actor;
class type_erased_tuple;
class type_erased_value;
class actor_control_block;
class binary_deserializer;
class actor_system_config;
class uniform_type_info_map;
class forwarding_actor_proxy;
......
......@@ -41,7 +41,7 @@ public:
explicit serializer(actor_system& sys);
explicit serializer(execution_unit* ctx);
explicit serializer(execution_unit* ctx = nullptr);
virtual ~serializer();
};
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_DESERIALIZER_HPP
#define CAF_STREAM_DESERIALIZER_HPP
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <exception>
#include <iomanip>
#include <limits>
#include <sstream>
#include <stdexcept>
#include <string>
#include <type_traits>
#include "caf/deserializer.hpp"
#include "caf/logger.hpp"
#include "caf/detail/ieee_754.hpp"
#include "caf/detail/network_order.hpp"
namespace caf {
/// Implements the deserializer interface with a binary serialization protocol.
template <class Streambuf>
class stream_deserializer : public deserializer {
static_assert(
std::is_base_of<
std::streambuf,
typename std::remove_reference<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)} {
}
void begin_object(uint16_t& typenr, std::string& name) override {
apply_int(typenr);
if (typenr == 0)
apply(name);
}
void end_object() override {
// nop
}
void begin_sequence(size_t& num_elements) override {
static_assert(sizeof(size_t) >= sizeof(uint32_t),
"sizeof(size_t) < sizeof(uint32_t)");
uint32_t tmp;
apply_int(tmp);
num_elements = static_cast<size_t>(tmp);
}
void end_sequence() override {
// nop
}
void apply_raw(size_t num_bytes, void* data) override {
range_check(num_bytes);
streambuf_.sgetn(reinterpret_cast<char*>(data), num_bytes);
}
protected:
void apply_builtin(builtin type, void* val) override {
CAF_ASSERT(val != nullptr);
switch (type) {
case i8_v:
case u8_v:
apply_raw(sizeof(uint8_t), val);
break;
case i16_v:
case u16_v:
apply_int(*reinterpret_cast<uint16_t*>(val));
break;
case i32_v:
case u32_v:
apply_int(*reinterpret_cast<uint32_t*>(val));
break;
case i64_v:
case u64_v:
apply_int(*reinterpret_cast<uint64_t*>(val));
break;
case float_v:
apply_float(*reinterpret_cast<float*>(val));
break;
case double_v:
apply_float(*reinterpret_cast<double*>(val));
break;
case ldouble_v: {
// the IEEE-754 conversion does not work for long double
// => fall back to string serialization (even though it sucks)
std::string tmp;
apply(tmp);
std::istringstream iss{std::move(tmp)};
iss >> *reinterpret_cast<long double*>(val);
break;
}
case string8_v: {
auto& str = *reinterpret_cast<std::string*>(val);
uint32_t str_size;
apply_int(str_size);
range_check(str_size);
str.resize(str_size);
// TODO: When using C++14, switch to str.data(), which then has a
// non-const overload.
streambuf_.sgetn(reinterpret_cast<char*>(&str[0]), str_size);
break;
}
case string16_v: {
auto& str = *reinterpret_cast<std::u16string*>(val);
uint32_t str_size;
apply_int(str_size);
auto bytes = str_size * sizeof(std::u16string::value_type);
range_check(bytes);
str.resize(str_size);
// TODO: When using C++14, switch to str.data(), which then has a
// non-const overload.
streambuf_.sgetn(reinterpret_cast<char*>(&str[0]), bytes);
break;
}
case string32_v: {
auto& str = *reinterpret_cast<std::u32string*>(val);
uint32_t str_size;
apply_int(str_size);
auto bytes = str_size * sizeof(std::u32string::value_type);
range_check(bytes);
str.resize(str_size);
// TODO: When using C++14, switch to str.data(), which then has a
// non-const overload.
streambuf_.sgetn(reinterpret_cast<char*>(&str[0]), bytes);
break;
}
}
}
private:
void range_check(size_t read_size) {
static_assert(sizeof(std::streamsize) <= sizeof(size_t),
"std::streamsize > std::size_t");
if (static_cast<size_t>(streambuf_.in_avail()) < read_size) {
CAF_LOG_ERROR("range_check failed");
throw std::out_of_range("stream_deserializer<T>::read_range()");
}
}
template <class T>
void apply_int(T& x) {
T tmp;
apply_raw(sizeof(T), &tmp);
x = detail::from_network_order(tmp);
}
template <class T>
void apply_float(T& x) {
typename detail::ieee_754_trait<T>::packed_type tmp;
apply_int(tmp);
x = detail::unpack754(tmp);
}
Streambuf streambuf_;
};
} // namespace caf
#endif // CAF_STREAM_DESERIALIZER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SERIALIZER_HPP
#define CAF_STREAM_SERIALIZER_HPP
#include <cstddef>
#include <cstdint>
#include <iomanip>
#include <limits>
#include <streambuf>
#include <sstream>
#include <string>
#include <type_traits>
#include "caf/serializer.hpp"
#include "caf/streambuf.hpp"
#include "caf/detail/ieee_754.hpp"
#include "caf/detail/network_order.hpp"
namespace caf {
/// Implements the serializer interface with a binary serialization protocol.
template <class Streambuf>
class stream_serializer : public serializer {
static_assert(
std::is_base_of<
std::streambuf,
typename std::remove_reference<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)} {
}
void begin_object(uint16_t& typenr, std::string& name) override {
apply(typenr);
if (typenr == 0)
apply(name);
}
void end_object() override {
// nop
}
void begin_sequence(size_t& list_size) override {
auto s = static_cast<uint32_t>(list_size);
apply(s);
}
void end_sequence() override {
// nop
}
void apply_raw(size_t num_bytes, void* data) override {
streambuf_.sputn(reinterpret_cast<char*>(data), num_bytes);
}
protected:
void apply_builtin(builtin type, void* val) override {
CAF_ASSERT(val != nullptr);
switch (type) {
case i8_v:
case u8_v:
apply_raw(sizeof(uint8_t), val);
break;
case i16_v:
case u16_v:
apply_int(*reinterpret_cast<uint16_t*>(val));
break;
case i32_v:
case u32_v:
apply_int(*reinterpret_cast<uint32_t*>(val));
break;
case i64_v:
case u64_v:
apply_int(*reinterpret_cast<uint64_t*>(val));
break;
case float_v:
apply_int(detail::pack754(*reinterpret_cast<float*>(val)));
break;
case double_v:
apply_int(detail::pack754(*reinterpret_cast<double*>(val)));
break;
case ldouble_v: {
// 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)
<< *reinterpret_cast<long double*>(val);
auto tmp = oss.str();
auto s = static_cast<uint32_t>(tmp.size());
apply(s);
apply_raw(tmp.size(), const_cast<char*>(tmp.c_str()));
break;
}
case string8_v: {
auto str = reinterpret_cast<std::string*>(val);
auto s = static_cast<uint32_t>(str->size());
apply(s);
apply_raw(str->size(), const_cast<char*>(str->c_str()));
break;
}
case string16_v: {
auto& str = *reinterpret_cast<std::u16string*>(val);
auto s = static_cast<uint32_t>(str.size());
apply(s);
for (auto c : str) {
// the standard does not guarantee that char16_t is exactly 16 bits...
auto tmp = static_cast<uint16_t>(c);
apply_raw(sizeof(tmp), &tmp);
}
break;
}
case string32_v: {
auto& str = *reinterpret_cast<std::u32string*>(val);
auto s = static_cast<uint32_t>(str.size());
apply(s);
for (auto c : str) {
// the standard does not guarantee that char32_t is exactly 32 bits...
auto tmp = static_cast<uint32_t>(c);
apply_raw(sizeof(tmp), &tmp);
}
break;
}
}
}
private:
template <class T>
void apply_int(T x) {
auto y = detail::to_network_order(x);
apply_raw(sizeof(T), &y);
}
Streambuf streambuf_;
};
} // namespace caf
#endif // CAF_STREAM_SERIALIZER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAMBUF_HPP
#define CAF_STREAMBUF_HPP
#include <algorithm>
#include <cstddef>
#include <cstring>
#include <streambuf>
#include <vector>
#include "caf/detail/type_traits.hpp"
namespace caf {
/// 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 std::basic_streambuf<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<CharT*>(c.data()), c.size()} {
}
/// 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(CharT* 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;
}
/// Replaces the internal character sequence with a new one.
/// @param s A pointer to the first character.
/// @param n The length of the character sequence.
std::basic_streambuf<CharT, Traits>*
setbuf(CharT* s, std::streamsize n) override {
this->setg(s, s, s + n);
this->setp(s, s + n);
return this;
}
protected:
std::streamsize xsputn(const char_type* s, std::streamsize n) override {
auto available = this->epptr() - this->pptr();
auto actual = std::min(n, available);
std::memcpy(this->pptr(), s, actual * sizeof(char_type));
this->pbump(actual);
return actual;
}
std::streamsize xsgetn(char_type* s, std::streamsize n) override {
auto available = this->egptr() - this->gptr();
auto actual = std::min(n, available);
std::memcpy(s, this->gptr(), actual * sizeof(char_type));
this->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 std::basic_streambuf<
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} {
this->setg(const_cast<char_type*>(c.data()),
const_cast<char_type*>(c.data()),
const_cast<char_type*>(c.data()) + c.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.
int_type sputc(char_type c) {
container_.push_back(c);
return c;
}
protected:
// 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, available);
std::memcpy(s, this->gptr(), actual * sizeof(char_type));
this->gbump(actual);
return actual;
}
// 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 {
// TODO: Do a performance analysis whether the current implementation based
// on copy_n is faster than these two statements:
// (1) container_.resize(container_.size() + n);
// (2) std::memcpy(this->pptr(), s, n * sizeof(char_type));
std::copy_n(s, n, std::back_inserter(container_));
return n;
}
private:
Container& container_;
};
using charbuf = arraybuf<char>;
using vectorbuf = containerbuf<std::vector<char>>;
} // namespace caf
#endif // CAF_STREAMBUF_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include <limits>
#include <string>
#include <cstdint>
#include <cstring>
#include <sstream>
#include <iomanip>
#include <iterator>
#include <exception>
#include <stdexcept>
#include <type_traits>
#include "caf/logger.hpp"
#include "caf/actor_system.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/detail/ieee_754.hpp"
#include "caf/detail/network_order.hpp"
namespace caf {
namespace {
using pointer = const void*;
const char* as_char_pointer(pointer ptr) {
return reinterpret_cast<const char*>(ptr);
}
template <class Distance>
pointer advanced(pointer ptr, Distance num_bytes) {
return reinterpret_cast<const char*>(ptr) + num_bytes;
}
inline void range_check(pointer begin, pointer end, size_t read_size) {
if (advanced(begin, read_size) > end) {
CAF_LOG_ERROR("range_check failed");
throw std::out_of_range("binary_deserializer::read_range()");
}
}
template <class T>
void apply_int(binary_deserializer& bd, T& x) {
T tmp;
bd.apply_raw(sizeof(T), &tmp);
x = detail::from_network_order(tmp);
}
template <class T>
void apply_float(binary_deserializer& bd, T& x) {
typename detail::ieee_754_trait<T>::packed_type tmp;
apply_int(bd, tmp);
x = detail::unpack754(tmp);
}
} // namespace <anonmyous>
binary_deserializer::binary_deserializer(actor_system& sys,
const void* buf, size_t buf_size)
: binary_deserializer(sys, buf, advanced(buf, buf_size)) {
// nop
}
binary_deserializer::binary_deserializer(execution_unit* ctx,
const void* buf,
size_t buf_size)
: binary_deserializer(ctx, buf, advanced(buf, buf_size)) {
// nop
}
binary_deserializer::binary_deserializer(actor_system& sys,
const void* first, const void* last)
: binary_deserializer(sys.dummy_execution_unit(), first, last) {
// nop
}
binary_deserializer::binary_deserializer(execution_unit* ctx,
const void* first, const void* last)
: deserializer(ctx),
pos_(first),
end_(last) {
// nop
}
bool binary_deserializer::at_end() const {
return pos_ == end_;
}
/// with same semantics as `strncmp(this->pos_, buf, num_bytes)`.
bool binary_deserializer::buf_equals(const void* buf, size_t num_bytes) {
auto bytes_left = static_cast<size_t>(std::distance(as_char_pointer(pos_),
as_char_pointer(end_)));
if (bytes_left < num_bytes)
return false;
return strncmp(as_char_pointer(pos_), as_char_pointer(buf), num_bytes) == 0;
}
binary_deserializer& binary_deserializer::advance(ptrdiff_t num_bytes) {
pos_ = advanced(pos_, num_bytes);
return *this;
}
void binary_deserializer::begin_object(uint16_t& nr, std::string& name) {
apply_int(*this, nr);
if (nr == 0)
apply(name);
}
void binary_deserializer::end_object() {
// nop
}
void binary_deserializer::begin_sequence(size_t& result) {
CAF_LOG_TRACE("");
static_assert(sizeof(size_t) >= sizeof(uint32_t),
"sizeof(size_t) < sizeof(uint32_t)");
uint32_t tmp;
apply_int(*this, tmp);
result = static_cast<size_t>(tmp);
}
void binary_deserializer::end_sequence() {
// nop
}
void binary_deserializer::apply_raw(size_t num_bytes, void* storage) {
range_check(pos_, end_, num_bytes);
memcpy(storage, pos_, num_bytes);
pos_ = advanced(pos_, num_bytes);
}
void binary_deserializer::apply_builtin(builtin type, void* val) {
CAF_ASSERT(val != nullptr);
switch (type) {
case i8_v:
case u8_v:
apply_raw(sizeof(uint8_t), val);
break;
case i16_v:
case u16_v:
apply_int(*this, *reinterpret_cast<uint16_t*>(val));
break;
case i32_v:
case u32_v:
apply_int(*this, *reinterpret_cast<uint32_t*>(val));
break;
case i64_v:
case u64_v:
apply_int(*this, *reinterpret_cast<uint64_t*>(val));
break;
case float_v:
apply_float(*this, *reinterpret_cast<float*>(val));
break;
case double_v:
apply_float(*this, *reinterpret_cast<double*>(val));
break;
case ldouble_v: {
// the IEEE-754 conversion does not work for long double
// => fall back to string serialization (event though it sucks)
std::string tmp;
apply(tmp);
std::istringstream iss{std::move(tmp)};
iss >> *reinterpret_cast<long double*>(val);
break;
}
case string8_v: {
auto& str = *reinterpret_cast<std::string*>(val);
uint32_t str_size;
apply_int(*this, str_size);
range_check(pos_, end_, str_size);
str.clear();
str.reserve(str_size);
auto last = advanced(pos_, str_size);
std::copy(reinterpret_cast<const char*>(pos_),
reinterpret_cast<const char*>(last),
std::back_inserter(str));
pos_ = last;
break;
}
case string16_v: {
auto& str = *reinterpret_cast<std::string*>(val);
uint32_t str_size;
apply_int(*this, str_size);
range_check(pos_, end_, str_size * sizeof(uint16_t));
str.clear();
str.reserve(str_size);
auto last = advanced(pos_, str_size);
std::copy(reinterpret_cast<const uint16_t*>(pos_),
reinterpret_cast<const uint16_t*>(last),
std::back_inserter(str));
pos_ = last;
break;
}
case string32_v: {
auto& str = *reinterpret_cast<std::string*>(val);
uint32_t str_size;
apply_int(*this, str_size);
range_check(pos_, end_, str_size * sizeof(uint32_t));
str.clear();
str.reserve(str_size);
auto last = advanced(pos_, str_size);
std::copy(reinterpret_cast<const uint32_t*>(pos_),
reinterpret_cast<const uint32_t*>(last),
std::back_inserter(str));
pos_ = last;
break;
}
}
}
} // namespace caf
......@@ -253,7 +253,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
return;
}
std::vector<char> buf;
binary_serializer bs{self->context(), std::back_inserter(buf)};
binary_serializer bs{std::back_inserter(buf), self->context()};
self->save_state(bs, 0);
auto sender = node.sender;
// request(...)
......@@ -299,7 +299,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
self->bhvr_stack().pop_back();
self->is_migrated_from(false);
}
binary_deserializer bd{self->context(), buf.data(), buf.size()};
binary_deserializer bd{buf.data(), buf.size(), self->context()};
self->load_state(bd, 0);
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
......
......@@ -46,6 +46,7 @@
#include <type_traits>
#include "caf/message.hpp"
#include "caf/streambuf.hpp"
#include "caf/serializer.hpp"
#include "caf/ref_counted.hpp"
#include "caf/deserializer.hpp"
......@@ -160,14 +161,14 @@ struct fixture {
template <class T, class... Ts>
vector<char> serialize(T& x, Ts&... xs) {
vector<char> buf;
binary_serializer bs{&context, std::back_inserter(buf)};
binary_serializer bs{&context, buf};
apply(bs, x, xs...);
return buf;
}
template <class T, class... Ts>
void deserialize(const vector<char>& buf, T& x, Ts&... xs) {
binary_deserializer bd{&context, buf.data(), buf.size()};
binary_deserializer bd{&context, buf};
apply(bd, x, xs...);
}
......@@ -381,6 +382,7 @@ CAF_TEST(multiple_messages) {
CAF_CHECK(is_message(m2).equal(i32, te, str, rs));
}
CAF_TEST(type_erased_value) {
auto buf = serialize(str);
type_erased_value_ptr ptr{new type_erased_value_impl<std::string>};
......@@ -411,4 +413,28 @@ CAF_TEST(type_erased_tuple) {
CAF_CHECK_EQUAL(to_string(tview), deep_to_string(std::make_tuple(str, i32)));
}
CAF_TEST(streambuf_serialization) {
auto data = std::string{"The quick brown fox jumps over the lazy dog"};
std::vector<char> buf;
// First, we check the standard use case in CAF where stream serializers own
// their stream buffers.
stream_serializer<vectorbuf> bs{vectorbuf{buf}};
bs << data;
stream_deserializer<charbuf> bd{charbuf{buf}};
std::string target;
bd >> target;
CAF_CHECK(data == target);
// Second, we test another use case where the serializers only keep
// references of the stream buffers.
buf.clear();
target.clear();
vectorbuf vb{buf};
stream_serializer<vectorbuf&> vs{vb};
vs << data;
charbuf cb{buf};
stream_deserializer<charbuf&> vd{cb};
vd >> target;
CAF_CHECK(data == target);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -17,115 +17,94 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/binary_serializer.hpp"
#include "caf/config.hpp"
#include <limits>
#define CAF_SUITE streambuf
#include "caf/test/unit_test.hpp"
#include "caf/detail/ieee_754.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/streambuf.hpp"
namespace caf {
using namespace caf;
namespace {
template <class T>
void apply_int(binary_serializer& bs, T x) {
auto y = detail::to_network_order(x);
bs.apply_raw(sizeof(T), &y);
}
} // namespace <anonmyous>
void binary_serializer::begin_object(uint16_t& nr, std::string& name) {
apply(nr);
if (nr == 0)
apply(name);
}
void binary_serializer::end_object() {
// nop
CAF_TEST(signed_arraybuf) {
auto data = std::string{"The quick brown fox jumps over the lazy dog"};
arraybuf<char> ab{data};
// Let's read some.
CAF_CHECK_EQUAL(static_cast<size_t>(ab.in_avail()), data.size());
CAF_CHECK_EQUAL(ab.sgetc(), 'T');
std::string buf;
buf.resize(3);
size_t got = ab.sgetn(&buf[0], 3);
CAF_CHECK_EQUAL(got, 3u);
CAF_CHECK_EQUAL(buf, "The");
CAF_CHECK_EQUAL(ab.sgetc(), ' ');
// Exhaust the stream.
buf.resize(data.size());
got = ab.sgetn(&buf[0] + 3, data.size() - 3);
CAF_CHECK_EQUAL(got, data.size() - 3);
CAF_CHECK_EQUAL(data, buf);
CAF_CHECK_EQUAL(ab.in_avail(), 0);
// No more.
auto c = ab.sgetc();
CAF_CHECK_EQUAL(c, charbuf::traits_type::eof());
// Reset the stream and write into it.
ab.setbuf(&data[0], data.size());
CAF_CHECK_EQUAL(static_cast<size_t>(ab.in_avail()), data.size());
auto put = ab.sputn("One", 3);
CAF_CHECK_EQUAL(put, 3);
CAF_CHECK(data.compare(0, 3, "One") == 0);
}
void binary_serializer::begin_sequence(size_t& list_size) {
auto s = static_cast<uint32_t>(list_size);
apply(s);
CAF_TEST(unsigned_arraybuf) {
std::vector<uint8_t> data = {1, 2, 3, 4};
arraybuf<uint8_t> ab{data};
decltype(data) buf;
std::copy(std::istreambuf_iterator<uint8_t>{&ab},
std::istreambuf_iterator<uint8_t>{},
std::back_inserter(buf));
CAF_CHECK_EQUAL(data, buf);
}
void binary_serializer::end_sequence() {
// nop
CAF_TEST(containerbuf) {
std::string data{
"Habe nun, ach! Philosophie,\n"
"Juristerei und Medizin,\n"
"Und leider auch Theologie\n"
"Durchaus studiert, mit heißem Bemühn.\n"
"Da steh ich nun, ich armer Tor!\n"
"Und bin so klug als wie zuvor"
};
// Write some data.
std::vector<char> buf;
vectorbuf vb{buf};
auto put = vb.sputn(data.data(), data.size());
CAF_CHECK_EQUAL(static_cast<size_t>(put), data.size());
put = vb.sputn(";", 1);
CAF_CHECK_EQUAL(put, 1);
std::string target;
std::copy(buf.begin(), buf.end(), std::back_inserter(target));
CAF_CHECK_EQUAL(data + ';', target);
// Check "overflow" on a new stream.
buf.clear();
vectorbuf vb2{buf};
auto chr = vb2.sputc('x');
CAF_CHECK_EQUAL(chr, 'x');
// Let's read some data into a stream.
buf.clear();
containerbuf<std::string> scb{data};
std::copy(std::istreambuf_iterator<char>{&scb},
std::istreambuf_iterator<char>{},
std::back_inserter(buf));
CAF_CHECK_EQUAL(buf.size(), data.size());
CAF_CHECK(std::equal(buf.begin(), buf.end(), data.begin() /*, data.end() */));
// We're done, nothing to see here, please move along.
CAF_CHECK_EQUAL(scb.sgetc(), containerbuf<std::string>::traits_type::eof());
// Let's read again, but now in one big block.
buf.clear();
containerbuf<std::string> sib2{data};
buf.resize(data.size());
size_t got = sib2.sgetn(&buf[0], buf.size());
CAF_CHECK_EQUAL(got, data.size());
CAF_CHECK_EQUAL(buf.size(), data.size());
CAF_CHECK(std::equal(buf.begin(), buf.end(), data.begin() /*, data.end() */));
}
void binary_serializer::apply_raw(size_t num_bytes, void* data) {
out_(reinterpret_cast<char*>(data), num_bytes);
}
void binary_serializer::apply_builtin(builtin type, void* val) {
CAF_ASSERT(val != nullptr);
switch (type) {
case i8_v:
case u8_v:
apply_raw(sizeof(uint8_t), val);
break;
case i16_v:
case u16_v:
apply_int(*this, *reinterpret_cast<uint16_t*>(val));
break;
case i32_v:
case u32_v:
apply_int(*this, *reinterpret_cast<uint32_t*>(val));
break;
case i64_v:
case u64_v:
apply_int(*this, *reinterpret_cast<uint64_t*>(val));
break;
case float_v:
apply_int(*this, detail::pack754(*reinterpret_cast<float*>(val)));
break;
case double_v:
apply_int(*this, detail::pack754(*reinterpret_cast<double*>(val)));
break;
case ldouble_v: {
// 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)
<< *reinterpret_cast<long double*>(val);
auto tmp = oss.str();
auto s = static_cast<uint32_t>(tmp.size());
apply(s);
apply_raw(tmp.size(), const_cast<char*>(tmp.c_str()));
break;
}
case string8_v: {
auto str = reinterpret_cast<std::string*>(val);
auto s = static_cast<uint32_t>(str->size());
apply(s);
apply_raw(str->size(), const_cast<char*>(str->c_str()));
break;
}
case string16_v: {
auto& str = *reinterpret_cast<std::u16string*>(val);
auto s = static_cast<uint32_t>(str.size());
apply(s);
for (auto c : str) {
// the standard does not guarantee that char16_t is exactly 16 bits...
auto tmp = static_cast<uint16_t>(c);
apply_raw(sizeof(tmp), &tmp);
}
break;
}
case string32_v: {
auto& str = *reinterpret_cast<std::u32string*>(val);
auto s = static_cast<uint32_t>(str.size());
apply(s);
for (auto c : str) {
// the standard does not guarantee that char32_t is exactly 32 bits...
auto tmp = static_cast<uint32_t>(c);
apply_raw(sizeof(tmp), &tmp);
}
break;
}
}
}
} // namespace caf
......@@ -19,6 +19,7 @@
#include "caf/io/basp/instance.hpp"
#include "caf/streambuf.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp"
......@@ -65,7 +66,7 @@ connection_state instance::handle(execution_unit* ctx,
return err();
}
} else {
binary_deserializer bd{ctx, dm.buf.data(), dm.buf.size()};
binary_deserializer bd{ctx, dm.buf};
bd >> hdr;
CAF_LOG_DEBUG(CAF_ARG(hdr));
if (! valid(hdr)) {
......@@ -81,7 +82,7 @@ connection_state instance::handle(execution_unit* ctx,
&& hdr.dest_node != this_node_) {
auto path = lookup(hdr.dest_node);
if (path) {
binary_serializer bs{ctx, std::back_inserter(path->wr_buf)};
binary_serializer bs{ctx, path->wr_buf};
bs << hdr;
if (payload)
bs.apply_raw(payload->size(), payload->data());
......@@ -119,7 +120,7 @@ connection_state instance::handle(execution_unit* ctx,
actor_id aid = invalid_actor_id;
std::set<std::string> sigs;
if (payload_valid()) {
binary_deserializer bd{ctx, payload->data(), payload->size()};
binary_deserializer bd{ctx, *payload};
bd >> aid >> sigs;
}
// close self connection after handshake is done
......@@ -176,7 +177,7 @@ connection_state instance::handle(execution_unit* ctx,
&& tbl_.lookup_direct(hdr.source_node) == invalid_connection_handle
&& tbl_.add_indirect(last_hop, hdr.source_node))
callee_.learned_new_node_indirectly(hdr.source_node);
binary_deserializer bd{ctx, payload->data(), payload->size()};
binary_deserializer bd{ctx, *payload};
std::vector<strong_actor_ptr> forwarding_stack;
message msg;
bd >> forwarding_stack >> msg;
......@@ -323,7 +324,7 @@ void instance::write(execution_unit* ctx,
payload_writer* pw) {
if (! pw) {
uint32_t zero = 0;
binary_serializer bs{ctx, std::back_inserter(buf)};
binary_serializer bs{ctx, buf};
bs << source_node
<< dest_node
<< source_actor
......@@ -333,16 +334,16 @@ void instance::write(execution_unit* ctx,
<< operation_data;
} else {
// reserve space in the buffer to write the payload later on
auto wr_pos = static_cast<ptrdiff_t>(buf.size());
auto wr_pos = buf.size();
char placeholder[basp::header_size];
buf.insert(buf.end(), std::begin(placeholder), std::end(placeholder));
auto pl_pos = buf.size();
{ // lifetime scope of first serializer (write payload)
binary_serializer bs{ctx, std::back_inserter(buf)};
binary_serializer bs{ctx, buf};
(*pw)(bs);
}
// write broker message to the reserved space
binary_serializer bs2{ctx, buf.begin() + wr_pos};
stream_serializer<charbuf> bs2{ctx, buf.data() + wr_pos, pl_pos - wr_pos};
auto plen = static_cast<uint32_t>(buf.size() - pl_pos);
bs2 << source_node
<< dest_node
......
......@@ -140,7 +140,7 @@ public:
uint32_t serialized_size(const message& msg) {
buffer buf;
binary_serializer bs{mpx_, back_inserter(buf)};
binary_serializer bs{mpx_, buf};
bs << msg;
return static_cast<uint32_t>(buf.size());
}
......@@ -233,7 +233,7 @@ public:
template <class... Ts>
void to_payload(buffer& buf, const Ts&... xs) {
binary_serializer bs{mpx_, std::back_inserter(buf)};
binary_serializer bs{mpx_, buf};
to_payload(bs, xs...);
}
......@@ -254,7 +254,7 @@ public:
std::pair<basp::header, buffer> from_buf(const buffer& buf) {
basp::header hdr;
binary_deserializer bd{mpx_, buf.data(), buf.size()};
binary_deserializer bd{mpx_, buf};
bd >> hdr;
buffer payload;
if (hdr.payload_len > 0) {
......@@ -327,7 +327,7 @@ public:
std::tie(hdr, buf) = read_from_out_buf(hdl);
CAF_MESSAGE("dispatch output buffer for connection " << hdl.id());
CAF_REQUIRE(hdr.operation == basp::message_type::dispatch_message);
binary_deserializer source{mpx_, buf.data(), buf.size()};
binary_deserializer source{mpx_, buf};
std::vector<strong_actor_ptr> stages;
message msg;
source >> stages >> msg;
......@@ -372,7 +372,7 @@ public:
CAF_MESSAGE("output buffer has " << ob.size() << " bytes");
basp::header hdr;
{ // lifetime scope of source
binary_deserializer source{this_->mpx(), ob.data(), ob.size()};
binary_deserializer source{this_->mpx(), ob};
source >> hdr;
}
buffer payload;
......
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