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

#include <stream_gatherer.hpp>

Inheritance diagram for caf::stream_gatherer:

Public Types
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
virtual 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)=0
	Adds a path to the edge and emits `ack_open` to the source. More...

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

virtual void	close (message result)=0
	Removes all paths gracefully.

virtual void	abort (error reason)=0
	Removes all paths with an error message.

virtual long	num_paths () const =0
	Returns the number of paths managed on this edge.

virtual bool	closed () const =0
	Returns `true` if no downstream exists, `false` otherwise.

virtual bool	continuous () const =0
	Returns whether this edge remains open after the last path is removed.

virtual void	continuous (bool value)=0
	Sets whether this edge remains open after the last path is removed.

virtual path_ptr	find (const stream_id &sid, const actor_addr &x)=0
	Returns the stored state for `x` if `x` is a known path and associated to `sid`, otherwise `nullptr`. More...

virtual path_ptr	path_at (size_t index)=0
	Returns the stored state for `x` if `x` is a known path and associated to `sid`, otherwise `nullptr`. More...

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

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

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

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

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

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

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))`.

Detailed Description

Type-erased policy for receiving data from sources.

Member Function Documentation

◆ add_path()

virtual path_ptr caf::stream_gatherer::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
	)

pure virtual

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.