Commit f094d85b authored by Dominik Charousset's avatar Dominik Charousset

Add new class for actor-hosted streams

parent c04371ed
...@@ -189,6 +189,7 @@ caf_add_component( ...@@ -189,6 +189,7 @@ caf_add_component(
src/serializer.cpp src/serializer.cpp
src/settings.cpp src/settings.cpp
src/skip.cpp src/skip.cpp
src/stream.cpp
src/string_algorithms.cpp src/string_algorithms.cpp
src/string_view.cpp src/string_view.cpp
src/telemetry/collector/prometheus.cpp src/telemetry/collector/prometheus.cpp
...@@ -343,6 +344,7 @@ caf_add_component( ...@@ -343,6 +344,7 @@ caf_add_component(
simple_timeout simple_timeout
span span
stateful_actor stateful_actor
stream
string_algorithms string_algorithms
string_view string_view
telemetry.collector.prometheus telemetry.collector.prometheus
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/actor_control_block.hpp"
#include "caf/cow_string.hpp"
#include "caf/detail/comparable.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/type_id.hpp"
#include <cstddef>
#include <cstdint>
#include <string>
namespace caf {
class CAF_CORE_EXPORT stream : private detail::comparable<stream> {
public:
// -- constructors, destructors, and assignment operators --------------------
stream() = default;
stream(stream&&) noexcept = default;
stream(const stream&) noexcept = default;
stream& operator=(stream&&) noexcept = default;
stream& operator=(const stream&) noexcept = default;
stream(caf::strong_actor_ptr source, type_id_t type, std::string name,
uint64_t id)
: source_(std::move(source)), type_(type), name_(std::move(name)), id_(id) {
// nop
}
// -- properties -------------------------------------------------------------
template <class T>
bool has_element_type() const noexcept {
return type_id_v<T> == type_;
}
const caf::strong_actor_ptr& source() {
return source_;
}
type_id_t type() const noexcept {
return type_;
}
const std::string& name() const noexcept {
return name_.str();
}
uint64_t id() const noexcept {
return id_;
}
// -- comparison -------------------------------------------------------------
ptrdiff_t compare(const stream& other) const noexcept;
// -- serialization ----------------------------------------------------------
template <class Inspector>
friend bool inspect(Inspector& f, stream& obj) {
return f.object(obj).fields(f.field("source", obj.source_),
f.field("type", obj.type_),
f.field("name", obj.name_),
f.field("id", obj.id_));
}
private:
caf::strong_actor_ptr source_;
type_id_t type_ = invalid_type_id;
cow_string name_;
uint64_t id_ = 0;
};
} // namespace caf
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/stream.hpp"
#include "caf/config.hpp"
namespace caf {
ptrdiff_t stream::compare(const stream& other) const noexcept {
if (source_ < other.source_)
return -1;
if (source_ == other.source_) {
if (id_ < other.id_)
return -1;
if (id_ == other.id_) {
CAF_ASSERT(type_ == other.type_);
CAF_ASSERT(name_ == other.name_);
return 0;
}
}
return 1;
}
} // namespace caf
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE stream
#include "caf/stream.hpp"
#include "core-test.hpp"
using namespace caf;
namespace {
struct fixture : test_coordinator_fixture<> {
template <class T>
caf::expected<T> deep_copy(const T& obj) {
caf::byte_buffer buf;
{
caf::binary_serializer sink{sys, buf};
if (!sink.apply(obj))
return {sink.get_error()};
}
auto result = T{};
{
caf::binary_deserializer source{sys, buf};
if (!source.apply(result))
return {source.get_error()};
}
return result;
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
TEST_CASE("default-constructed") {
auto uut = stream{};
CHECK(!uut.has_element_type<int32_t>());
CHECK_EQ(uut.id(), 0u);
CHECK_EQ(uut.type(), invalid_type_id);
CHECK_EQ(uut.name(), "");
CHECK_EQ(uut.source(), nullptr);
CHECK_EQ(uut, deep_copy(uut));
}
TEST_CASE("value-constructed") {
auto dummy = sys.spawn([] {});
auto uut = stream{actor_cast<strong_actor_ptr>(dummy), type_id_v<int32_t>,
"foo", 42};
CHECK(uut.has_element_type<int32_t>());
CHECK_EQ(uut.id(), 42u);
CHECK_EQ(uut.type(), type_id_v<int32_t>);
CHECK_EQ(uut.name(), "foo");
CHECK_EQ(uut.source(), dummy);
CHECK_EQ(uut, deep_copy(uut));
}
END_FIXTURE_SCOPE()
...@@ -13,6 +13,16 @@ ...@@ -13,6 +13,16 @@
} \ } \
void CAF_UNIQUE(test)::run_test_impl() void CAF_UNIQUE(test)::run_test_impl()
#define TEST_CASE(description) \
namespace { \
struct CAF_UNIQUE(test) : caf_test_case_auto_fixture { \
void run_test_impl(); \
}; \
::caf::test::detail::adder<::caf::test::test_impl<CAF_UNIQUE(test)>> \
CAF_UNIQUE(a){CAF_XSTR(CAF_SUITE), description, false}; \
} \
void CAF_UNIQUE(test)::run_test_impl()
#define GIVEN(description) \ #define GIVEN(description) \
CAF_MESSAGE("GIVEN " description); \ CAF_MESSAGE("GIVEN " description); \
if (true) if (true)
......
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