Commit e1513383 authored by Dominik Charousset's avatar Dominik Charousset

Implement statically-typed stream handles

parent 6cdf7973
...@@ -362,6 +362,7 @@ caf_add_component( ...@@ -362,6 +362,7 @@ caf_add_component(
typed_message_view typed_message_view
typed_response_promise typed_response_promise
typed_spawn typed_spawn
typed_stream
unit unit
unordered_flat_map unordered_flat_map
uri uri
......
...@@ -20,6 +20,7 @@ namespace caf { ...@@ -20,6 +20,7 @@ namespace caf {
// -- 1 param templates -------------------------------------------------------- // -- 1 param templates --------------------------------------------------------
template <class> class [[deprecated ("use std::optional instead")]] optional;
template <class> class [[nodiscard]] error_code; template <class> class [[nodiscard]] error_code;
template <class> class basic_cow_string; template <class> class basic_cow_string;
template <class> class behavior_type_of; template <class> class behavior_type_of;
...@@ -29,9 +30,9 @@ template <class> class dictionary; ...@@ -29,9 +30,9 @@ template <class> class dictionary;
template <class> class expected; template <class> class expected;
template <class> class intrusive_cow_ptr; template <class> class intrusive_cow_ptr;
template <class> class intrusive_ptr; template <class> class intrusive_ptr;
template <class> class [[deprecated ("use std::optional instead")]] optional;
template <class> class param; template <class> class param;
template <class> class span; template <class> class span;
template <class> class typed_stream;
template <class> class weak_intrusive_ptr; template <class> class weak_intrusive_ptr;
template <class> struct inspector_access; template <class> struct inspector_access;
......
...@@ -468,16 +468,26 @@ public: ...@@ -468,16 +468,26 @@ public:
void watch(disposable what) override; void watch(disposable what) override;
/// Converts an @ref observable into a @ref stream to allow other actors to
/// consume its items.
/// @param name The human-readable name for this stream.
/// @param max_delay The maximum delay between emitting two batches.
/// @param max_items_per_batch The maximum amount of items per batch.
/// @param obs An observable sequence of items.
/// @returns a @ref stream that makes @p obs available to other actors.
template <class Observable> template <class Observable>
flow::assert_scheduled_actor_hdr_t<Observable, stream> flow::assert_scheduled_actor_hdr_t<Observable, stream>
to_stream(std::string name, timespan max_delay, size_t max_items_per_batch, to_stream(std::string name, timespan max_delay, size_t max_items_per_batch,
Observable&&); Observable&& obs);
/// @copydoc to_stream
template <class Observable> template <class Observable>
flow::assert_scheduled_actor_hdr_t<Observable, stream> flow::assert_scheduled_actor_hdr_t<Observable, stream>
to_stream(cow_string name, timespan max_delay, size_t max_items_per_batch, to_stream(cow_string name, timespan max_delay, size_t max_items_per_batch,
Observable&&); Observable&& obs);
/// Utility class that converts an @ref observable to a @ref stream when
/// passed to @c compose.
struct to_stream_t { struct to_stream_t {
scheduled_actor* self; scheduled_actor* self;
cow_string name; cow_string name;
...@@ -496,11 +506,70 @@ public: ...@@ -496,11 +506,70 @@ public:
return {this, cow_string{std::move(name)}, max_delay, max_items_per_batch}; return {this, cow_string{std::move(name)}, max_delay, max_items_per_batch};
} }
/// Returns a function object for passing it to @c compose.
to_stream_t to_stream(cow_string name, timespan max_delay, to_stream_t to_stream(cow_string name, timespan max_delay,
size_t max_items_per_batch) { size_t max_items_per_batch) {
return {this, std::move(name), max_delay, max_items_per_batch}; return {this, std::move(name), max_delay, max_items_per_batch};
} }
/// Converts an @ref observable into a @ref stream to allow other actors to
/// consume its items.
/// @param name The human-readable name for this stream.
/// @param max_delay The maximum delay between emitting two batches.
/// @param max_items_per_batch The maximum amount of items per batch.
/// @param obs An observable sequence of items.
/// @returns a @ref stream that makes @p obs available to other actors.
template <class Observable>
flow::assert_scheduled_actor_hdr_t<
Observable, typed_stream<typename Observable::output_type>>
to_typed_stream(std::string name, timespan max_delay,
size_t max_items_per_batch, Observable obs);
/// @copydoc to_stream
template <class Observable>
flow::assert_scheduled_actor_hdr_t<
Observable, typed_stream<typename Observable::output_type>>
to_typed_stream(cow_string name, timespan max_delay,
size_t max_items_per_batch, Observable obs);
/// Utility class that converts an @ref observable to a @ref stream when
/// passed to @c compose.
struct to_typed_stream_t {
scheduled_actor* self;
cow_string name;
timespan max_delay;
size_t max_items_per_batch;
template <class Observable>
auto operator()(Observable&& what) {
return self->to_typed_stream(name, max_delay, max_items_per_batch,
std::forward<Observable>(what));
}
};
/// Returns a function object for passing it to @c compose.
to_typed_stream_t to_typed_stream(std::string name, timespan max_delay,
size_t max_items_per_batch) {
return {this, cow_string{std::move(name)}, max_delay, max_items_per_batch};
}
/// Returns a function object for passing it to @c compose.
to_typed_stream_t to_typed_stream(cow_string name, timespan max_delay,
size_t max_items_per_batch) {
return {this, std::move(name), max_delay, max_items_per_batch};
}
/// Lifts a statically typed stream into an @ref observable.
/// @param what The input stream.
/// @param buf_capacity Upper bound for caching inputs from the stream.
/// @param demand_threshold Minimal free buffer capacity before signaling
/// demand upstream.
/// @note Both @p buf_capacity and @p demand_threshold are considered hints.
/// The actor may increase (or decrease) the effective settings
/// depending on the amount of messages per batch or other factors.
template <class T>
flow::assert_scheduled_actor_hdr_t<flow::observable<T>>
observe(typed_stream<T> what, size_t buf_capacity, size_t demand_threshold);
/// Lifts a stream into an @ref observable. /// Lifts a stream into an @ref observable.
/// @param what The input stream. /// @param what The input stream.
/// @param buf_capacity Upper bound for caching inputs from the stream. /// @param buf_capacity Upper bound for caching inputs from the stream.
......
...@@ -15,6 +15,7 @@ ...@@ -15,6 +15,7 @@
#include "caf/flow/single.hpp" #include "caf/flow/single.hpp"
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/stream.hpp" #include "caf/stream.hpp"
#include "caf/typed_stream.hpp"
namespace caf::flow { namespace caf::flow {
...@@ -108,6 +109,35 @@ scheduled_actor::to_stream(cow_string name, timespan max_delay, ...@@ -108,6 +109,35 @@ scheduled_actor::to_stream(cow_string name, timespan max_delay,
max_items_per_batch); max_items_per_batch);
} }
template <class Observable>
flow::assert_scheduled_actor_hdr_t<
Observable, typed_stream<typename Observable::output_type>>
scheduled_actor::to_typed_stream(std::string name, timespan max_delay,
size_t max_items_per_batch, Observable obs) {
auto res = to_stream(std::move(name), max_delay, max_items_per_batch,
std::move(obs));
return {res.source(), res.name(), res.id()};
}
template <class Observable>
flow::assert_scheduled_actor_hdr_t<
Observable, typed_stream<typename Observable::output_type>>
scheduled_actor::to_typed_stream(cow_string name, timespan max_delay,
size_t max_items_per_batch, Observable obs) {
auto res = to_stream(std::move(name), max_delay, max_items_per_batch,
std::move(obs));
return {res.source(), res.name(), res.id()};
}
template <class T>
flow::assert_scheduled_actor_hdr_t<flow::observable<T>>
scheduled_actor::observe(typed_stream<T> what, size_t buf_capacity,
size_t demand_threshold) {
return do_observe(what.dynamically_typed(), buf_capacity, demand_threshold)
.transform(detail::unbatch<T>{})
.as_observable();
}
template <class T> template <class T>
flow::assert_scheduled_actor_hdr_t<flow::observable<T>> flow::assert_scheduled_actor_hdr_t<flow::observable<T>>
scheduled_actor::observe_as(stream what, size_t buf_capacity, scheduled_actor::observe_as(stream what, size_t buf_capacity,
......
...@@ -55,7 +55,7 @@ public: ...@@ -55,7 +55,7 @@ public:
/// Queries the source of this stream. Default-constructed streams return a /// Queries the source of this stream. Default-constructed streams return a
/// @c null pointer. /// @c null pointer.
const strong_actor_ptr& source() { const strong_actor_ptr& source() const noexcept {
return source_; return source_;
} }
......
// 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/async/batch.hpp"
#include "caf/cow_string.hpp"
#include "caf/detail/comparable.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
#include "caf/stream.hpp"
#include <cstddef>
#include <cstdint>
#include <string>
namespace caf {
/// Provides access to a statically typed, potentially unbound sequence of items
/// emitted by an actor. Each stream is uniquely identified by the address of
/// the hosting actor plus an integer value. Further, streams have
/// human-readable names attached to them in order to make help with
/// observability and logging.
template <class T>
class typed_stream : private detail::comparable<typed_stream<T>>,
private detail::comparable<stream> {
public:
// -- constructors, destructors, and assignment operators --------------------
typed_stream() = default;
typed_stream(typed_stream&&) noexcept = default;
typed_stream(const typed_stream&) noexcept = default;
typed_stream& operator=(typed_stream&&) noexcept = default;
typed_stream& operator=(const typed_stream&) noexcept = default;
typed_stream(strong_actor_ptr source, std::string name, uint64_t id)
: source_(std::move(source)), name_(std::move(name)), id_(id) {
// nop
}
typed_stream(strong_actor_ptr source, cow_string name, uint64_t id)
: source_(std::move(source)), name_(std::move(name)), id_(id) {
// nop
}
// -- properties -------------------------------------------------------------
/// Queries the source of this stream. Default-constructed streams return a
/// @c null pointer.
const strong_actor_ptr& source() const noexcept {
return source_;
}
/// Returns the human-readable name for this stream, as announced by the
/// source.
const std::string& name() const noexcept {
return name_.str();
}
/// Returns the source-specific identifier for this stream.
uint64_t id() const noexcept {
return id_;
}
// -- conversion -------------------------------------------------------------
/// Returns a dynamically typed version of this stream.
stream dynamically_typed() const noexcept {
return {source_, type_id_v<T>, name_, id_};
}
// -- comparison -------------------------------------------------------------
ptrdiff_t compare(const stream& other) const noexcept {
return compare_impl(other);
}
ptrdiff_t compare(const typed_stream& other) const noexcept {
return compare_impl(other);
}
// -- serialization ----------------------------------------------------------
template <class Inspector>
friend bool inspect(Inspector& f, typed_stream& obj) {
return f.object(obj).fields(f.field("source", obj.source_),
f.field("name", obj.name_),
f.field("id", obj.id_));
}
private:
template <class OtherStream>
ptrdiff_t compare_impl(const OtherStream& other) const noexcept {
if (source_ < other.source())
return -1;
if (source_ == other.source()) {
if (id_ < other.id())
return -1;
if (id_ == other.id())
return 0;
}
return 1;
}
strong_actor_ptr source_;
cow_string name_;
uint64_t id_ = 0;
};
} // namespace caf
#pragma once
#include "caf/cow_vector.hpp" #include "caf/cow_vector.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/result.hpp" #include "caf/result.hpp"
#include "caf/test/bdd_dsl.hpp" #include "caf/test/bdd_dsl.hpp"
#include "caf/type_id.hpp" #include "caf/type_id.hpp"
#include "caf/typed_actor.hpp" #include "caf/typed_actor.hpp"
#include "caf/typed_stream.hpp"
#include <cstdint> #include <cstdint>
#include <numeric> #include <numeric>
...@@ -420,6 +423,7 @@ bool inspect(Inspector& f, phone_book& x) { ...@@ -420,6 +423,7 @@ bool inspect(Inspector& f, phone_book& x) {
CAF_BEGIN_TYPE_ID_BLOCK(core_test, caf::first_custom_type_id) CAF_BEGIN_TYPE_ID_BLOCK(core_test, caf::first_custom_type_id)
ADD_TYPE_ID((caf::cow_vector<int>) ) ADD_TYPE_ID((caf::cow_vector<int>) )
ADD_TYPE_ID((caf::typed_stream<int32_t>) )
ADD_TYPE_ID((circle)) ADD_TYPE_ID((circle))
ADD_TYPE_ID((dummy_enum)) ADD_TYPE_ID((dummy_enum))
ADD_TYPE_ID((dummy_enum_class)) ADD_TYPE_ID((dummy_enum_class))
......
// 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 typed_stream
#include "caf/typed_stream.hpp"
#include "core-test.hpp"
#include "caf/scheduled_actor/flow.hpp"
using namespace caf;
using namespace std::literals;
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;
}
};
using ivec = std::vector<int32_t>;
using istream = typed_stream<int32_t>;
behavior int_sink(event_based_actor* self, std::shared_ptr<ivec> results) {
return {
[self, results](istream input) {
self //
->observe(input, 30, 10)
.for_each([results](int x) { results->push_back(x); });
},
};
}
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
TEST_CASE("default-constructed") {
auto uut = istream{};
CHECK_EQ(uut.id(), 0u);
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 = istream{actor_cast<strong_actor_ptr>(dummy), "foo", 42};
CHECK_EQ(uut.id(), 42u);
CHECK_EQ(uut.name(), "foo");
CHECK_EQ(uut.source(), dummy);
CHECK_EQ(uut, deep_copy(uut));
}
TEST_CASE("streams allow actors to transmit flow items to others") {
auto res = ivec{};
res.resize(256);
std::iota(res.begin(), res.end(), 1);
auto r1 = std::make_shared<ivec>();
auto s1 = sys.spawn(int_sink, r1);
auto r2 = std::make_shared<ivec>();
auto s2 = sys.spawn(int_sink, r2);
run();
auto src = sys.spawn([s1, s2](event_based_actor* self) {
auto vals = self //
->make_observable()
.iota(int32_t{1})
.take(256)
.compose(self->to_typed_stream("foo", 10ms, 10));
self->send(s1, vals);
self->send(s2, vals);
});
run_once();
expect((istream), from(src).to(s1));
expect((istream), from(src).to(s2));
expect((stream_open_msg), from(s1).to(src));
expect((stream_open_msg), from(s2).to(src));
expect((stream_ack_msg), from(src).to(s1));
expect((stream_ack_msg), from(src).to(s2));
run();
CHECK_EQ(*r1, res);
CHECK_EQ(*r2, res);
}
END_FIXTURE_SCOPE()
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