class Channel[T <: Shareable]

include Enumerable[T]

A typed queue between tasks: all communication and synchronization between tasks goes through channels.

A channel of capacity 0 (the default) is a rendezvous: #send waits for a receiver and #receive for a sender. A larger capacity buffers that many values. Waiting suspends only the calling task. After #close, a send raises Channel::Closed and the buffered values can still be received. Updates made before a #send are visible after the matching #receive.

When every task waits and none can go on, the runtime stops the program with DeadlockError.

ch = Channel[Int].new(capacity: 2)
Task.spawn { [1, 2, 3].each { |n| ch.send(n) }; ch.close }
ch.to_a          # => [1, 2, 3]
ch.receive?      # => nil
ch.receive       # raises Channel::Closed

<prelude>:5929

Classes and modules

Class methods

def self.new(capacity: Int = …) -> Channel[T]

A new open channel buffering up to capacity values (0, the default, buffers none; a negative capacity is taken as 0).

<prelude>:5935

def self.select(&: (Selector) -> Void) -> Void

Performs exactly one of the operations the block registers on the Selector (Selector#receive, Selector#send, Selector#after, Selector#default), the first that can go on, and then runs that operation's block.

When several are ready, the first registered wins. Without Selector#default, it waits until one is ready, suspending only the calling task.

Raises

Channel::Closed when a registered channel is closed (and, for a receive, empty), even with a Selector#default; ArgumentError when the block registers nothing.

jobs = Channel[Int].new(capacity: 1)
jobs.send(7)
got = 0
Channel.select do |s|
  s.receive(jobs) { |n| got = n }
  s.default { got = -1 }
end
got              # => 7

<prelude>:5961

def self.select_loop(&: (Selector) -> Void) -> Void

.select in a loop, until a handler (or the block) calls Selector#stop. When it polls (Selector#default taken) it lets other tasks run after each select, as Task.yield does, so that it does not keep its senders from running; when its cases are ready every time, every 64 selects.

break in a handler leaves only that handler.

nums = Channel[Int].new(capacity: 3)
[1, 2, 3].each { |n| nums.send(n) }
sum = 0
Channel.select_loop do |s|
  s.receive(nums) { |n| sum += n }
  s.default { s.stop }
end
sum              # => 6

<prelude>:5980

Methods

def send(value: T) -> Void

Sends value, waiting while the buffer is full (on a capacity-0 channel, until a receiver takes it).

Raises

Channel::Closed when the channel is closed, or is closed while the task waits.

<prelude>:5989

def receive -> T

The next value, waiting while the channel is empty. The values buffered before #close are still received.

Raises

Channel::Closed when the channel is closed and empty.

<prelude>:5996

def receive? -> T?

The next value, waiting while the channel is empty; nil when it is closed and empty.

<prelude>:5999

def try_send(value: T) -> Bool

Sends value if that needs no wait (there is room in the buffer, or a receiver waits): whether it was sent.

Raises

Channel::Closed when the channel is closed.

<prelude>:6006

def try_receive -> T?

The next value if one is ready without waiting, otherwise nil (also when the channel is closed and empty).

<prelude>:6009

def close -> Void

Closes the channel: later sends raise Channel::Closed, and so do the waiting ones; the receivers get the buffered values and then Channel::Closed (or nil from #receive?).

Closing again does nothing.

<prelude>:6015

def closed? -> Bool

Whether #close has been called.

<prelude>:6017

def empty? -> Bool

Whether no value is buffered.

<prelude>:6019

def size -> Int

The number of values buffered now.

<prelude>:6021

def each(&: (T) -> Void) -> Void

Receives values and yields each until the channel is closed and empty, waiting in between.

<prelude>:6024