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

Rename caf::flow::{item_publisher => multicaster}

Since we are using the term "publisher" to refer to objects that publish
items asynchronously now, the name `item_publisher` is likely to cause
confusion for new users. The new name also better reflects the purpose.
parent 2e80b6cb
...@@ -35,6 +35,9 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -35,6 +35,9 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- The output of `--dump-config` now only contains CAF options from loaded - The output of `--dump-config` now only contains CAF options from loaded
modules. Previously, it also included options from modules that were not modules. Previously, it also included options from modules that were not
loaded. loaded.
- We renamed `caf::flow::item_publisher` to `caf::flow::multicaster` to better
reflect its purpose and to avoid confusion with the new
`caf::async::publisher`.
### Fixed ### Fixed
......
...@@ -41,7 +41,7 @@ void worker_impl(caf::event_based_actor* self, ...@@ -41,7 +41,7 @@ void worker_impl(caf::event_based_actor* self,
// add this ID to the input to tag it. // add this ID to the input to tag it.
using message_t = std::pair<caf::uuid, lp::frame>; using message_t = std::pair<caf::uuid, lp::frame>;
// Allows us to push new flows into the central merge point. // Allows us to push new flows into the central merge point.
caf::flow::item_publisher<caf::flow::observable<message_t>> pub{self}; caf::flow::multicaster<caf::flow::observable<message_t>> pub{self};
// Our central merge point combines all inputs into a single, shared flow. // Our central merge point combines all inputs into a single, shared flow.
auto messages = pub.as_observable().merge().share(); auto messages = pub.as_observable().merge().share();
// Have one subscription for debug output. This also makes sure that the // Have one subscription for debug output. This also makes sure that the
......
...@@ -32,7 +32,7 @@ public: ...@@ -32,7 +32,7 @@ public:
using frame = caf::net::lp::frame; using frame = caf::net::lp::frame;
using publisher_type = caf::flow::item_publisher<QString>; using publisher_type = caf::flow::multicaster<QString>;
ChatWidget(QWidget* parent = nullptr); ChatWidget(QWidget* parent = nullptr);
......
...@@ -287,8 +287,8 @@ caf_add_component( ...@@ -287,8 +287,8 @@ caf_add_component(
flow.flat_map flow.flat_map
flow.for_each flow.for_each
flow.generation flow.generation
flow.item_publisher
flow.mixed flow.mixed
flow.multicaster
flow.observe_on flow.observe_on
flow.op.buffer flow.op.buffer
flow.op.cell flow.op.cell
......
...@@ -44,6 +44,9 @@ class observable_def; ...@@ -44,6 +44,9 @@ class observable_def;
template <class Generator> template <class Generator>
class generation_materializer; class generation_materializer;
template <class T>
class multicaster;
/// A blueprint for an @ref observer that generates items and applies any number /// A blueprint for an @ref observer that generates items and applies any number
/// of processing steps immediately before emitting them. /// of processing steps immediately before emitting them.
template <class Generator, class... Steps> template <class Generator, class... Steps>
......
...@@ -4,102 +4,11 @@ ...@@ -4,102 +4,11 @@
#pragma once #pragma once
#include "caf/flow/fwd.hpp" #include "caf/flow/multicaster.hpp"
#include "caf/flow/observable_decl.hpp"
#include "caf/flow/op/mcast.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include <cstdint>
namespace caf::flow { namespace caf::flow {
template <class T> template <class T>
class item_publisher { using item_publisher [[deprecated("use multicaster instead")]] = multicaster<T>;
public:
using impl_ptr = intrusive_ptr<op::mcast<T>>;
explicit item_publisher(coordinator* ctx) {
pimpl_ = make_counted<op::mcast<T>>(ctx);
}
explicit item_publisher(impl_ptr ptr) noexcept : pimpl_(std::move(ptr)) {
// nop
}
item_publisher(item_publisher&&) noexcept = default;
item_publisher& operator=(item_publisher&&) noexcept = default;
item_publisher(const item_publisher&) = delete;
item_publisher& operator=(const item_publisher&) = delete;
~item_publisher() {
if (pimpl_)
pimpl_->close();
}
/// Pushes an item to all subscribed observers. The publisher drops the item
/// if no subscriber exists.
void push(const T& item) {
pimpl_->push_all(item);
}
/// Pushes the items in range `[first, last)` to all subscribed observers. The
/// publisher drops the items if no subscriber exists.
template <class Iterator, class Sentinel>
void push(Iterator first, Sentinel last) {
while (first != last)
push(*first++);
}
/// Pushes the items from the initializer list to all subscribed observers.
/// The publisher drops the items if no subscriber exists.
void push(std::initializer_list<T> items) {
for (auto& item : items)
push(item);
}
/// Closes the publisher, eventually emitting on_complete on all observers.
void close() {
pimpl_->close();
}
/// Closes the publisher, eventually emitting on_error on all observers.
void abort(const error& reason) {
pimpl_->abort(reason);
}
/// Queries how many items the publisher may emit immediately to subscribed
/// observers.
size_t demand() const noexcept {
return pimpl_->min_demand();
}
/// Queries how many items are currently waiting in a buffer until the
/// observer requests additional items.
size_t buffered() const noexcept {
return pimpl_->max_buffered();
}
/// Queries whether there is at least one observer subscribed to the operator.
bool has_observers() const noexcept {
return pimpl_->has_observers();
}
/// Converts the publisher to an @ref observable.
observable<T> as_observable() const {
return observable<T>{pimpl_};
}
/// Subscribes a new @ref observer to the output of the publisher.
disposable subscribe(observer<T> out) {
return pimpl_->subscribe(out);
}
private:
impl_ptr pimpl_;
};
} // namespace caf::flow } // namespace caf::flow
// 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/flow/fwd.hpp"
#include "caf/flow/observable_decl.hpp"
#include "caf/flow/op/mcast.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include <cstdint>
namespace caf::flow {
/// A multicaster pushes items to any number of subscribers.
template <class T>
class multicaster {
public:
using impl_ptr = intrusive_ptr<op::mcast<T>>;
explicit multicaster(coordinator* ctx) {
pimpl_ = make_counted<op::mcast<T>>(ctx);
}
explicit multicaster(impl_ptr ptr) noexcept : pimpl_(std::move(ptr)) {
// nop
}
multicaster(multicaster&&) noexcept = default;
multicaster& operator=(multicaster&&) noexcept = default;
multicaster(const multicaster&) = delete;
multicaster& operator=(const multicaster&) = delete;
~multicaster() {
if (pimpl_)
pimpl_->close();
}
/// Pushes an item to all subscribed observers. The publisher drops the item
/// if no subscriber exists.
void push(const T& item) {
pimpl_->push_all(item);
}
/// Pushes the items in range `[first, last)` to all subscribed observers. The
/// publisher drops the items if no subscriber exists.
template <class Iterator, class Sentinel>
void push(Iterator first, Sentinel last) {
while (first != last)
push(*first++);
}
/// Pushes the items from the initializer list to all subscribed observers.
/// The publisher drops the items if no subscriber exists.
void push(std::initializer_list<T> items) {
for (auto& item : items)
push(item);
}
/// Closes the publisher, eventually emitting on_complete on all observers.
void close() {
pimpl_->close();
}
/// Closes the publisher, eventually emitting on_error on all observers.
void abort(const error& reason) {
pimpl_->abort(reason);
}
/// Queries how many items the publisher may emit immediately to subscribed
/// observers.
size_t demand() const noexcept {
return pimpl_->min_demand();
}
/// Queries how many items are currently waiting in a buffer until the
/// observer requests additional items.
size_t buffered() const noexcept {
return pimpl_->max_buffered();
}
/// Queries whether there is at least one observer subscribed to the operator.
bool has_observers() const noexcept {
return pimpl_->has_observers();
}
/// Converts the publisher to an @ref observable.
observable<T> as_observable() const {
return observable<T>{pimpl_};
}
/// Subscribes a new @ref observer to the output of the publisher.
disposable subscribe(observer<T> out) {
return pimpl_->subscribe(out);
}
private:
impl_ptr pimpl_;
};
} // namespace caf::flow
...@@ -27,7 +27,8 @@ ...@@ -27,7 +27,8 @@
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/fwd.hpp" #include "caf/flow/fwd.hpp"
#include "caf/flow/item_publisher.hpp" #include "caf/flow/item_publisher.hpp" // deprecated
#include "caf/flow/multicaster.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive/drr_cached_queue.hpp" #include "caf/intrusive/drr_cached_queue.hpp"
...@@ -177,7 +178,7 @@ public: ...@@ -177,7 +178,7 @@ public:
using exception_handler = std::function<error(pointer, std::exception_ptr&)>; using exception_handler = std::function<error(pointer, std::exception_ptr&)>;
#endif // CAF_ENABLE_EXCEPTIONS #endif // CAF_ENABLE_EXCEPTIONS
using batch_publisher = flow::item_publisher<async::batch>; using batch_publisher = flow::multicaster<async::batch>;
class batch_forwarder : public ref_counted { class batch_forwarder : public ref_counted {
public: public:
......
...@@ -2,9 +2,9 @@ ...@@ -2,9 +2,9 @@
// the main distribution directory for license terms and copyright or visit // the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE. // https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.item_publisher #define CAF_SUITE flow.multicaster
#include "caf/flow/item_publisher.hpp" #include "caf/flow/multicaster.hpp"
#include "core-test.hpp" #include "core-test.hpp"
...@@ -29,11 +29,11 @@ struct fixture : test_coordinator_fixture<> { ...@@ -29,11 +29,11 @@ struct fixture : test_coordinator_fixture<> {
BEGIN_FIXTURE_SCOPE(fixture) BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("item publishers discard items that arrive before a subscriber") { SCENARIO("a multicaster discards items that arrive before a subscriber") {
GIVEN("an item publisher") { GIVEN("an multicaster") {
WHEN("publishing items") { WHEN("pushing items") {
THEN("observers see only items that were published after subscribing") { THEN("observers see only items that were pushed after subscribing") {
auto uut = flow::item_publisher<int>{ctx.get()}; auto uut = flow::multicaster<int>{ctx.get()};
uut.push({1, 2, 3}); uut.push({1, 2, 3});
auto snk = flow::make_auto_observer<int>(); auto snk = flow::make_auto_observer<int>();
uut.subscribe(snk->as_observer()); uut.subscribe(snk->as_observer());
......
...@@ -9,7 +9,7 @@ ...@@ -9,7 +9,7 @@
#include "core-test.hpp" #include "core-test.hpp"
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/item_publisher.hpp" #include "caf/flow/multicaster.hpp"
#include "caf/flow/observable.hpp" #include "caf/flow/observable.hpp"
#include "caf/flow/observable_builder.hpp" #include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
...@@ -118,7 +118,7 @@ SCENARIO("the buffer operator forces items at regular intervals") { ...@@ -118,7 +118,7 @@ SCENARIO("the buffer operator forces items at regular intervals") {
cow_vector<int>{}, cow_vector<int>{64}, cow_vector<int>{}, cow_vector<int>{64},
cow_vector<int>{}, cow_vector<int>{128, 256, 512}, cow_vector<int>{}, cow_vector<int>{128, 256, 512},
}; };
auto pub = flow::item_publisher<int>{ctx.get()}; auto pub = flow::multicaster<int>{ctx.get()};
sys.spawn([&pub, outputs](caf::event_based_actor* self) { sys.spawn([&pub, outputs](caf::event_based_actor* self) {
pub.as_observable() pub.as_observable()
.observe_on(self) // .observe_on(self) //
......
...@@ -8,7 +8,7 @@ ...@@ -8,7 +8,7 @@
#include "core-test.hpp" #include "core-test.hpp"
#include "caf/flow/item_publisher.hpp" #include "caf/flow/multicaster.hpp"
#include "caf/flow/observable_builder.hpp" #include "caf/flow/observable_builder.hpp"
#include "caf/flow/scoped_coordinator.hpp" #include "caf/flow/scoped_coordinator.hpp"
...@@ -89,7 +89,7 @@ SCENARIO("mergers round-robin over their inputs") { ...@@ -89,7 +89,7 @@ SCENARIO("mergers round-robin over their inputs") {
} }
GIVEN("a merger with one input that completes") { GIVEN("a merger with one input that completes") {
WHEN("subscribing to the merger and requesting before the first push") { WHEN("subscribing to the merger and requesting before the first push") {
auto src = flow::item_publisher<int>{ctx.get()}; auto src = flow::multicaster<int>{ctx.get()};
auto uut = make_counted<flow::op::merge<int>>(ctx.get(), auto uut = make_counted<flow::op::merge<int>>(ctx.get(),
src.as_observable()); src.as_observable());
auto snk = flow::make_passive_observer<int>(); auto snk = flow::make_passive_observer<int>();
...@@ -122,7 +122,7 @@ SCENARIO("mergers round-robin over their inputs") { ...@@ -122,7 +122,7 @@ SCENARIO("mergers round-robin over their inputs") {
} }
} }
WHEN("subscribing to the merger pushing before the first request") { WHEN("subscribing to the merger pushing before the first request") {
auto src = flow::item_publisher<int>{ctx.get()}; auto src = flow::multicaster<int>{ctx.get()};
auto uut = make_counted<flow::op::merge<int>>(ctx.get(), auto uut = make_counted<flow::op::merge<int>>(ctx.get(),
src.as_observable()); src.as_observable());
ctx->run(); ctx->run();
...@@ -158,7 +158,7 @@ SCENARIO("mergers round-robin over their inputs") { ...@@ -158,7 +158,7 @@ SCENARIO("mergers round-robin over their inputs") {
} }
GIVEN("a merger with one input that aborts after some items") { GIVEN("a merger with one input that aborts after some items") {
WHEN("subscribing to the merger") { WHEN("subscribing to the merger") {
auto src = flow::item_publisher<int>{ctx.get()}; auto src = flow::multicaster<int>{ctx.get()};
auto uut = make_counted<flow::op::merge<int>>(ctx.get(), auto uut = make_counted<flow::op::merge<int>>(ctx.get(),
src.as_observable()); src.as_observable());
auto snk = flow::make_passive_observer<int>(); auto snk = flow::make_passive_observer<int>();
......
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