Commit 2e80b6cb authored by Dominik Charousset's avatar Dominik Charousset

Add new publisher abstraction for async flows

parent 3d18029e
......@@ -18,6 +18,13 @@ is based on [Keep a Changelog](https://keepachangelog.com).
`--foo=bar` syntax.
- The functions `make_message` and `make_error` now support `std::string_view`
as input and automatically convert it to `std::string`.
- To make it easier to set up asynchronous flows, CAF now provides a new class:
`caf::async::publisher`. Any observable can be transformed into a publisher by
calling `to_publisher`. The publisher can then be used to subscribe to the
observable from other actors or threads. The publisher has only a single
member function: `observe_on`. It converts the publisher back into an
observable. This new abstraction allows users to set up asynchronous flows
without having to manually deal with SPSC buffers.
### Changed
......
......@@ -234,6 +234,7 @@ caf_add_component(
async.consumer_adapter
async.producer_adapter
async.promise
async.publisher
async.spsc_buffer
behavior
binary_deserializer
......
......@@ -18,10 +18,10 @@ class producer;
// -- template classes ---------------------------------------------------------
template <class T>
class spsc_buffer;
class consumer_resource;
template <class T>
class consumer_resource;
class future;
template <class T>
class producer_resource;
......@@ -30,7 +30,10 @@ template <class T>
class promise;
template <class T>
class future;
class publisher;
template <class T>
class spsc_buffer;
// -- smart pointer aliases ----------------------------------------------------
......
// 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/async/spsc_buffer.hpp"
#include "caf/flow/coordinator.hpp"
#include "caf/flow/observable_decl.hpp"
#include "caf/flow/op/fail.hpp"
namespace caf::async {
/// Provides an interface for accessing an asynchronous data flow. Unlike a @ref
/// future, a publisher produces multiple values over time. Subscribers will
/// only receive items that are emitted after they have subscribed to the
/// publisher.
template <class T>
class publisher {
public:
publisher() noexcept = default;
publisher(publisher&&) noexcept = default;
publisher(const publisher&) noexcept = default;
publisher& operator=(publisher&&) noexcept = default;
publisher& operator=(const publisher&) noexcept = default;
/// Creates a @ref flow::observable that reads and emits all values from this
/// publisher.
flow::observable<T> observe_on(flow::coordinator* ctx, size_t buffer_size,
size_t min_request_size) {
if (impl_)
return impl_->observe_on(ctx, buffer_size, min_request_size);
auto err = make_error(sec::invalid_observable,
"cannot subscribe to default-constructed observable");
// Note: cannot use ctx->make_observable() here because it would create a
// circular dependency between observable_builder and publisher.
return flow::make_observable<flow::op::fail<T>>(ctx, std::move(err));
}
/// Creates a @ref flow::observable that reads and emits all values from this
/// publisher.
flow::observable<T> observe_on(flow::coordinator* ctx) {
return observe_on(ctx, defaults::flow::buffer_size,
defaults::flow::min_demand);
}
/// Creates a new asynchronous observable by decorating a regular observable.
static publisher from(flow::observable<T> decorated) {
if (!decorated)
return {};
auto* ctx = decorated.ctx();
auto flag = disposable::make_flag();
ctx->watch(flag);
auto pimpl = make_counted<impl>(ctx, std::move(decorated), std::move(flag));
return publisher{std::move(pimpl)};
}
private:
class impl : public ref_counted {
public:
impl(async::execution_context_ptr source, flow::observable<T> decorated,
disposable flag)
: source_(std::move(source)),
decorated_(std::move(decorated)),
flag_(std::move(flag)) {
// nop
}
flow::observable<T> observe_on(flow::coordinator* ctx, size_t buffer_size,
size_t min_request_size) {
// Short circuit if we are already on the target coordinator.
if (ctx == source_.get())
return decorated_;
// Otherwise, create a new SPSC buffer and connect it to the source.
auto [pull, push] = async::make_spsc_buffer_resource<T>(buffer_size,
min_request_size);
source_->schedule_fn(
[push = std::move(push), decorated = decorated_]() mutable {
decorated.subscribe(std::move(push));
});
return pull.observe_on(ctx);
}
~impl() {
source_->schedule_fn([flag = std::move(flag_)]() mutable {
// The source called `watch` on the flag to keep the event loop alive as
// long as there are still async references to this observable. We need
// to dispose the flag in the event loop of the source in order to make
// sure that the source cleans up properly.
flag.dispose();
});
}
private:
async::execution_context_ptr source_;
flow::observable<T> decorated_;
disposable flag_;
};
using impl_ptr = caf::intrusive_ptr<impl>;
explicit publisher(impl_ptr pimpl) : impl_(std::move(pimpl)) {
// nop
}
impl_ptr impl_;
};
} // namespace caf::async
......@@ -69,6 +69,9 @@ public:
/// disposes all elements individually.
static disposable make_composite(std::vector<disposable> entries);
/// Creates a disposable that simply represents a flag.
static disposable make_flag();
// -- mutators ---------------------------------------------------------------
/// Disposes the resource. Calling `dispose()` on a disposed resource is a
......
......@@ -6,6 +6,7 @@
#include "caf/async/consumer.hpp"
#include "caf/async/producer.hpp"
#include "caf/async/publisher.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/cow_tuple.hpp"
#include "caf/cow_vector.hpp"
......@@ -377,6 +378,11 @@ public:
return materialize().to_resource(buffer_size, min_request_size);
}
/// @copydoc observable::to_resource
async::publisher<output_type> to_publisher() && {
return materialize().to_publisher();
}
/// @copydoc observable::observe_on
observable<output_type> observe_on(coordinator* other) && {
return materialize().observe_on(other);
......@@ -762,4 +768,9 @@ observable<T>::to_resource(size_t buffer_size, size_t min_request_size) {
return async::consumer_resource<T>{std::move(buf)};
}
template <class T>
async::publisher<T> observable<T>::to_publisher() {
return async::publisher<T>::from(*this);
}
} // namespace caf::flow
......@@ -4,6 +4,7 @@
#pragma once
#include "caf/async/fwd.hpp"
#include "caf/cow_vector.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/is_complete.hpp"
......@@ -250,6 +251,9 @@ public:
return to_resource(defaults::flow::buffer_size, defaults::flow::min_demand);
}
/// Creates a publisher that makes emitted items available asynchronously.
async::publisher<T> to_publisher();
const observable& as_observable() const& noexcept {
return *this;
}
......
......@@ -4,7 +4,7 @@
#pragma once
#include "caf/detail/plain_ref_counted.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/base.hpp"
#include "caf/flow/subscription.hpp"
......@@ -13,7 +13,7 @@ namespace caf::flow::op {
/// Convenience base type for *cold* observable types.
template <class T>
class cold : public detail::plain_ref_counted, public base<T> {
class cold : public detail::atomic_ref_counted, public base<T> {
public:
// -- member types -----------------------------------------------------------
......
......@@ -4,7 +4,7 @@
#pragma once
#include "caf/detail/plain_ref_counted.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/base.hpp"
#include "caf/flow/subscription.hpp"
......@@ -13,7 +13,7 @@ namespace caf::flow::op {
/// Convenience base type for *hot* observable types.
template <class T>
class hot : public detail::plain_ref_counted, public base<T> {
class hot : public detail::atomic_ref_counted, public base<T> {
public:
// -- member types -----------------------------------------------------------
......
......@@ -71,6 +71,48 @@ disposable disposable::make_composite(std::vector<disposable> entries) {
return disposable{make_counted<composite_impl>(std::move(entries))};
}
namespace {
class flag_impl : public ref_counted, public disposable::impl {
public:
flag_impl() : flag_(false) {
// nop
}
void dispose() {
flag_ = true;
}
bool disposed() const noexcept {
return flag_.load();
}
void ref_disposable() const noexcept {
ref();
}
void deref_disposable() const noexcept {
deref();
}
friend void intrusive_ptr_add_ref(const flag_impl* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const flag_impl* ptr) noexcept {
ptr->deref();
}
private:
std::atomic<bool> flag_;
};
} // namespace
disposable disposable::make_flag() {
return disposable{make_counted<flag_impl>()};
}
// -- utility ------------------------------------------------------------------
size_t disposable::erase_disposed(std::vector<disposable>& xs) {
......
// 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 async.publisher
#include "caf/async/publisher.hpp"
#include "core-test.hpp"
#include "caf/flow/scoped_coordinator.hpp"
#include "caf/scheduled_actor/flow.hpp"
using namespace caf;
BEGIN_FIXTURE_SCOPE(test_coordinator_fixture<>)
SCENARIO("actors can subscribe to their own publishers") {
GIVEN("an observable") {
WHEN("converting it to a publisher") {
THEN("the owning actor can subscribe to it") {
auto values = std::make_shared<std::vector<int>>();
sys.spawn([values](caf::event_based_actor* self) {
self->make_observable()
.iota(1)
.take(7)
.to_publisher()
.observe_on(self)
.do_on_error([](const error& what) { FAIL("error: " << what); })
.for_each([values](int x) { values->push_back(x); });
});
run();
CHECK_EQ(*values, std::vector({1, 2, 3, 4, 5, 6, 7}));
}
}
}
}
SCENARIO("default-constructed publishers are invalid") {
GIVEN("a default-constructed publisher") {
WHEN("an actor subscribes to it") {
THEN("the actor observes an invalid_observable error") {
auto err = std::make_shared<error>();
sys.spawn([err](caf::event_based_actor* self) {
auto items = async::publisher<int>{};
items.observe_on(self)
.do_on_error([err](const error& what) { *err = what; })
.for_each([](int) { FAIL("unexpected value"); });
});
run();
CHECK_EQ(*err, sec::invalid_observable);
}
}
}
}
SCENARIO("publishers from default-constructed observables are invalid") {
GIVEN("publisher with a default-constructed observable") {
WHEN("an actor subscribes to it") {
THEN("the actor observes an invalid_observable error") {
auto err = std::make_shared<error>();
sys.spawn([err](caf::event_based_actor* self) {
auto items = async::publisher<int>::from({});
items.observe_on(self)
.do_on_error([err](const error& what) { *err = what; })
.for_each([](int) { FAIL("unexpected value"); });
});
run();
CHECK_EQ(*err, sec::invalid_observable);
}
}
}
}
SCENARIO("actors can subscribe to publishers from other actors") {
GIVEN("three actors") {
WHEN("creating a publisher on one and subscribing on the others") {
THEN("the other actors receive the values") {
auto v1 = std::make_shared<std::vector<int>>();
auto v2 = std::make_shared<std::vector<int>>();
auto items = std::make_shared<async::publisher<int>>();
sys.spawn([items](caf::event_based_actor* self) {
*items = self->make_observable().iota(1).take(7).to_publisher();
});
run();
auto consumer = [items](caf::event_based_actor* self,
std::shared_ptr<std::vector<int>> values) {
items->observe_on(self)
.do_on_error([](const error& what) { FAIL("error: " << what); })
.for_each([values](int x) { values->push_back(x); });
};
sys.spawn(consumer, v1);
sys.spawn(consumer, v2);
run();
CHECK_EQ(*v1, std::vector({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(*v2, std::vector({1, 2, 3, 4, 5, 6, 7}));
}
}
}
}
SCENARIO("publishers from terminated actors produce errors") {
GIVEN("a publisher from a terminated actor") {
WHEN("another actors subscribes to it") {
THEN("the subscriber observe an error") {
auto items = std::make_shared<async::publisher<int>>();
sys.spawn([items](caf::event_based_actor* self) {
*items = self->make_observable().iota(1).take(7).to_publisher();
self->quit();
});
run();
auto err = std::make_shared<error>();
sys.spawn([items, err](caf::event_based_actor* self) {
items->observe_on(self)
.do_on_error([err](const error& what) { *err = what; })
.for_each([](int) { FAIL("unexpected value"); });
});
run();
CHECK_EQ(*err, sec::disposed);
}
}
}
}
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