class

Concurrent::Stream::SendRecv(B, V, E, SC)

Inherits Concurrent::Stream::Receive / Concurrent::Stream::Base / Reference / Object

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. B=block V=value E=accumulated_error SC=scope

Constructors

new(*, fibers : Int32, dst_vch : Channel(V), dst_ech : Channel(Tuple(Exception, E)) | Nil = nil, parent)
Source

Instance methods

batch(size : Int32, *, flush_interval : Float | Nil = nil, flush_empty : Bool = false)

Groups results in to chunks up to the given size. Runs in a single fiber. Multiple fibers would delay further stream processing.

Source
close
Source
errors(*, fibers : Int32 | Nil = nil, &block : Exception, E -> Nil)
Source
map(*, fibers : Int32 | Nil = nil, &block : V -> U) forall U

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

Source
run(*, fibers : Int32 | Nil = nil, &block : V -> _)

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

Source
scope

Further processing is evaluated within the scope of the returned object.

Source
select(*, fibers : Int32 | Nil = nil, &block : V -> Bool)

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

Source
serial
Source
tee(*, fibers : Int32 | Nil = nil, &block : V -> _)

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

Source
to_a(*args, **options)
Source
to_a(*args, **options, &)
Source