Type-erased policy for receiving data from sources. More...

#include <stream_gatherer_impl.hpp>

Inheritance diagram for caf::stream_gatherer_impl:

Public Types
using	super = stream_edge_impl< stream_gatherer >

using	assignment_pair = std::pair< path_type *, long >

Public Types inherited from caf::stream_edge_impl< stream_gatherer >
using	super = stream_gatherer

using	path_type = typename super::path_type
	Either `inbound_path` or `outbound_path`.

using	path_ptr = path_type *
	A raw pointer to a path.

using	path_ptr_vec = std::vector< path_ptr >
	Vector of raw pointers (views) of paths.

using	path_ptr_iter = typename path_ptr_vec::iterator
	Iterator to a vector of raw pointers.

using	path_uptr = std::unique_ptr< path_type >
	A unique pointer to a path.

using	path_uptr_vec = std::vector< path_uptr >
	Vector of owning pointers of paths.

using	path_uptr_iter = typename path_uptr_vec::iterator
	Iterator to a vector of owning pointers.

using	regular_shutdown = typename path_type::regular_shutdown
	Message type for sending graceful shutdowns along the path (either `stream_msg::drop` or `stream_msg::close`). More...

using	irregular_shutdown = typename path_type::irregular_shutdown
	Message type for sending errors along the path (either `stream_msg::forced_drop` or `stream_msg::forced_close`). More...

Public Types inherited from caf::stream_gatherer
using	path_type = inbound_path
	Type of a single path to a data source.

using	path_ptr = path_type *
	Pointer to a single path to a data source.

Public Member Functions
	stream_gatherer_impl (local_actor *selfptr)

path_ptr	add_path (const stream_id &sid, strong_actor_ptr x, strong_actor_ptr original_stage, stream_priority prio, long available_credit, bool redeployable, response_promise result_cb) override
	Adds a path to the edge and emits `ack_open` to the source. More...

bool	remove_path (const stream_id &sid, const actor_addr &x, error reason, bool silent) override
	Removes a path from the gatherer.

void	close (message result) override
	Removes all paths gracefully.

void	abort (error reason) override
	Removes all paths with an error message.

long	high_watermark () const override
	Returns the point at which an actor stops sending out demand immediately (waiting for the available credit to first drop below the watermark). More...

long	min_credit_assignment () const override
	Returns the minimum amount of credit required to send a `demand` message.

long	max_credit () const override
	Returns the maximum credit assigned to a single upstream actors.

void	high_watermark (long x) override
	Sets the point at which an actor stops sending out demand immediately (waiting for the available credit to first drop below the watermark). More...

void	min_credit_assignment (long x) override
	Sets the minimum amount of credit required to send a `demand` message.

void	max_credit (long x) override
	Sets the maximum credit assigned to a single upstream actors.

Public Member Functions inherited from caf::stream_edge_impl< stream_gatherer >
	stream_edge_impl (local_actor *selfptr)

path_ptr	find (const stream_id &sid, const actor_addr &x) override

const path_uptr_vec &	paths () const
	Returns all available paths.

local_actor *	self () const
	Returns a pointer to the parent actor.

bool	remove_path (path_uptr_iter i, error reason, bool silent)

bool	remove_path (const stream_id &sid, const actor_addr &x, error reason, bool silent) override

void	abort (error reason) override

long	num_paths () const override

bool	closed () const override

bool	continuous () const override

void	continuous (bool value) override

path_ptr	path_at (size_t index) override

Public Member Functions inherited from caf::stream_gatherer
virtual void	assign_credit (long downstream_capacity)=0
	Assigns new credit to all sources.

virtual long	initial_credit (long downstream_capacity, path_ptr x)=0
	Calculates initial credit for `x` after adding it to the gatherer.

bool	remove_path (const stream_id &sid, const strong_actor_ptr &x, error reason, bool silent)
	Removes a path from the gatherer.

path_ptr	find (const stream_id &sid, const strong_actor_ptr &x)
	Convenience function for calling `find(x, actor_cast<actor_addr>(x))`.

Protected Member Functions
void	emit_credits ()

Protected Member Functions inherited from caf::stream_edge_impl< stream_gatherer >
path_ptr	add_path_impl (const stream_id &sid, strong_actor_ptr x)
	Adds a path to the edge without emitting messages.

void	close_impl (F f)

Protected Attributes
long	high_watermark_

long	min_credit_assignment_

long	max_credit_

std::vector< assignment_pair >	assignment_vec_

std::vector< response_promise >	listeners_
	Listeners for the final result.

Protected Attributes inherited from caf::stream_edge_impl< stream_gatherer >
local_actor *	self_

path_uptr_vec	paths_

bool	continuous_

Additional Inherited Members
Static Public Member Functions inherited from caf::stream_edge_impl< stream_gatherer >
static void	sort_by_credit (PathContainer &xs)
	Sorts `xs` in descending order by available credit.

static T	fold (PathContainer &xs, T init, F f)

static path_ptr	find (PathContainer &xs, const stream_id &sid, const Handle &x)
	Finds the path for `ptr` and returns a pointer to it.

static PathContainer::iterator	iter_find (PathContainer &xs, const stream_id &sid, const Handle &x)
	Finds the path for `ptr` and returns an iterator to it.

Static Public Attributes inherited from caf::stream_edge_impl< stream_gatherer >
static constexpr const auto	aborter_type
	Stream aborter flag to monitor paths.

Detailed Description

Type-erased policy for receiving data from sources.

Member Function Documentation

◆ add_path()

path_ptr caf::stream_gatherer_impl::add_path	(	const stream_id &	sid,
		strong_actor_ptr	x,
		strong_actor_ptr	original_stage,
		stream_priority	prio,
		long	available_credit,
		bool	redeployable,
		response_promise	result_cb
	)

overridevirtual

Adds a path to the edge and emits ack_open to the source.

Parameters

sid	Stream ID used by the source.
x	Handle to the source.
original_stage	Actor that received the stream handshake initially.
prio	Priority of data on this path.
available_credit	Maximum credit for granting to the source.
redeployable	Stores whether the source can re-appear after aborts.
result_cb	Callback for the listener of the final stream result. The gatherer must ignore the promise when returning `nullptr`, because the previous stage is responsible for it until the gatherer acknowledges the handshake. The callback is invalid if the stream has a next stage.