libcaf  0.15.5
Public Types | Public Member Functions | Protected Member Functions | Protected Attributes | List of all members
caf::stream_gatherer_impl Class Reference

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

#include <stream_gatherer_impl.hpp>

Inheritance diagram for caf::stream_gatherer_impl:
caf::stream_edge_impl< stream_gatherer > caf::stream_gatherer caf::random_gatherer

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_vecpaths () 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_promiselisteners_
 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

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
sidStream ID used by the source.
xHandle to the source.
original_stageActor that received the stream handshake initially.
prioPriority of data on this path.
available_creditMaximum credit for granting to the source.
redeployableStores whether the source can re-appear after aborts.
result_cbCallback 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.
Returns
The added path on success, nullptr otherwise.

Implements caf::stream_gatherer.

long caf::stream_gatherer_impl::high_watermark ( ) const
overridevirtual

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

Implements caf::stream_gatherer.

void caf::stream_gatherer_impl::high_watermark ( long  x)
overridevirtual

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

Implements caf::stream_gatherer.


The documentation for this class was generated from the following file: