abstract class Concurrent::Stream::Base(T)

Overview

#map, #select, #run and #tee run in a fiber pool. #batch runs in a single fiber. All other methods "join" in the calling fiber.

Exceptions are raised in #each when joined.

TODO better error handling.

Included Modules

Direct Known Subclasses

Defined in:

concurrent/stream.cr

Constructors

Instance Method Summary

Instance methods inherited from module Concurrent::Stream::Receive

handle_error(ex, src_vch, src_ech, dst_ech) handle_error

Constructor Detail

def self.new(*, fibers : Int32, dst_vch : Channel(T), dst_ech : Channel(Exception)? = nil) #

[View source]

Instance Method Detail

def batch(size : Int32) #

Parallel batch. Runs in a single fiber. Multiple fibers would delay further stream processing.


[View source]
def close : Nil #

[View source]
def map(*, fibers : Int32? = nil, &block : T -> U) forall U #

Parallel map. &block is evaluated in a fiber pool.


[View source]
def run(*, fibers : Int32? = nil, &block : T -> _) #

Parallel run. &block is evaluated in a fiber pool. Further processing is not possible except for #wait.


[View source]
def select(*, fibers : Int32? = nil, &block : T -> Bool) #

Parallel select. &block is evaluated in a fiber pool.


[View source]
def serial #

[View source]
def tee(*, fibers : Int32? = nil, &block : T -> _) #

Parallel tee. &block is evaluated in a fiber pool. The original message is passed to the next Stream.


[View source]
def to_a(*args, **options) #

[View source]
def to_a(*args, **options, &) #

[View source]
def wait(*args, **options) #

[View source]
def wait(*args, **options, &) #

[View source]