This module contains the implementation of Std.Channel. Std.Channel is a multi-producer multi-consumer FIFO channel that offers both bounded and unbounded buffering as well as synchronous and asynchronous APIs.

Additionally Std.CloseableChannel is provided in case closing the channel is of interest. The two are distinct as the non-closeable Std.Channel can never throw errors which makes for cleaner code.

inductive Std.CloseableChannel.Error : Type

Errors that may be thrown while interacting with the channel API.

• closed : Error

  Tried to send to a closed channel.

• alreadyClosed : Error

  Tried to close an already closed channel.

instance Std.CloseableChannel.instReprError : Repr Error

def Std.CloseableChannel.instReprError.repr : Error → Nat → Format

instance Std.CloseableChannel.instDecidableEqError : DecidableEq Error

def Std.CloseableChannel.instHashableError.hash : Error → UInt64

instance Std.CloseableChannel.instHashableError : Hashable Error

instance Std.CloseableChannel.instToStringError : ToString Error

instance Std.CloseableChannel.instMonadLiftEIOErrorIO : MonadLift (EIO Error) IO

inductive Std.CloseableChannel.Flavors (α : Type) : Type

This type represents all flavors of channels that we have available.

• unbounded {α : Type} (ch : Std.CloseableChannel.Unbounded✝ α) : Std.CloseableChannel.Flavors α
• zero {α : Type} (ch : Std.CloseableChannel.Zero✝ α) : Std.CloseableChannel.Flavors α
• bounded {α : Type} (ch : Std.CloseableChannel.Bounded✝ α) : Std.CloseableChannel.Flavors α

instance Std.CloseableChannel.instNonemptyFlavors {α✝ : Type} : Nonempty (Flavors α✝)

def Std.CloseableChannel (α : Type) : Type

A multi-producer multi-consumer FIFO channel that offers both bounded and unbounded buffering and an asynchronous API, to switch into synchronous mode use CloseableChannel.sync.

Additionally Std.CloseableChannel can be closed if necessary, unlike Std.Channel. This introduces a need for error handling in some cases, thus it is usually easier to use Std.Channel if applicable.

def Std.CloseableChannel.Sync (α : Type) : Type

A multi-producer multi-consumer FIFO channel that offers both bounded and unbounded buffering and a synchronous API. This type acts as a convenient layer to use a channel in a blocking fashion and is not actually different from the original channel.

Additionally Std.CloseableChannel.Sync can be closed if necessary, unlike Std.Channel.Sync. This introduces the need to handle errors in some cases, thus it is usually easier to use Std.Channel if applicable.

instance Std.instNonemptyCloseableChannel {α : Type} : Nonempty (CloseableChannel α)

instance Std.instNonemptySync {α : Type} : Nonempty (CloseableChannel.Sync α)

def Std.CloseableChannel.new {α : Type} (capacity : Option Nat := none) : BaseIO (CloseableChannel α)

Create a new channel. If:

• capacity is none it will be unbounded (the default)
• capacity is some 0 it will always force a rendezvous between sender and receiver
• capacity is some n with n > 0 it will use a buffer of size n and begin blocking once it is filled

def Std.CloseableChannel.trySend {α : Type} (ch : CloseableChannel α) (v : α) : BaseIO Bool

Try to send a value to the channel. If this can be completed right away without blocking return true; otherwise, don't send the value and return false.

def Std.CloseableChannel.send {α : Type} (ch : CloseableChannel α) (v : α) : BaseIO (Task (Except Error Unit))

Send a value through the channel, returning a task that will resolve once the transmission could be completed. Note that the task may resolve to Except.error if the channel was closed before it could be completed.

def Std.CloseableChannel.close {α : Type} (ch : CloseableChannel α) : EIO Error Unit

Closes the channel, returns Except.ok when called the first time, otherwise Except.error. When a channel is closed:

• no new values can be sent successfully anymore
• all blocked consumers are resolved to none (as no new messages can be sent they will never resolve)
• if there are already values waiting to be received they can still be received by subsequent recv calls

def Std.CloseableChannel.isClosed {α : Type} (ch : CloseableChannel α) : BaseIO Bool

Return true if the channel is closed.

def Std.CloseableChannel.tryRecv {α : Type} (ch : CloseableChannel α) : BaseIO (Option α)

Try to receive a value from the channel, if this can be completed right away without blocking return some value, otherwise return none.

def Std.CloseableChannel.recv {α : Type} (ch : CloseableChannel α) : BaseIO (Task (Option α))

Receive a value from the channel, returning a task that will resolve once the transmission could be completed. Note that the task may resolve to none if the channel was closed before it could be completed.

def Std.CloseableChannel.recvSelector {α : Type} (ch : CloseableChannel α) : Internal.IO.Async.Selector (Option α)

Creates a Selector that resolves once ch has data available and provides that data. In particular if ch is closed while waiting on this Selector and no data is available already this will resolve to none.

partial def Std.CloseableChannel.forAsync {α : Type} (f : α → BaseIO Unit) (ch : CloseableChannel α) (prio : Task.Priority := Task.Priority.default) : BaseIO (Task Unit)

ch.forAsync f calls f for every message received on ch.

Note that if this function is called twice, each message will only arrive at exactly one invocation.

instance Std.CloseableChannel.instAsyncStreamOptionOfInhabited {α : Type} [Inhabited α] : Internal.Async.IO.AsyncStream (CloseableChannel α) (Option α)

instance Std.CloseableChannel.instAsyncReadOptionOfInhabited {α : Type} [Inhabited α] : Internal.Async.IO.AsyncRead (CloseableChannel α) (Option α)

instance Std.CloseableChannel.instAsyncWriteOfInhabited {α : Type} [Inhabited α] : Internal.Async.IO.AsyncWrite (CloseableChannel α) α

@[inline]

def Std.CloseableChannel.sync {α : Type} (ch : CloseableChannel α) : Sync α

This function is a no-op and just a convenient way to expose the synchronous API of the channel.

@[inline]

def Std.CloseableChannel.Sync.new {α : Type} (capacity : Option Nat := none) : BaseIO (Sync α)

Create a new channel. If:

• capacity is none it will be unbounded (the default)
• capacity is some 0 it will always force a rendezvous between sender and receiver
• capacity is some n with n > 0 it will use a buffer of size n and begin blocking once it is filled

@[inline]

def Std.CloseableChannel.Sync.trySend {α : Type} (ch : Sync α) (v : α) : BaseIO Bool

Try to send a value to the channel. If this can be completed right away without blocking return true; otherwise, don't send the value and return false.

def Std.CloseableChannel.Sync.send {α : Type} (ch : Sync α) (v : α) : EIO Error Unit

Send a value through the channel, blocking until the transmission could be completed. Note that this function may throw an error when trying to send to an already closed channel.

@[inline]

def Std.CloseableChannel.Sync.close {α : Type} (ch : Sync α) : EIO Error Unit

Closes the channel, returns Except.ok when called the first time, otherwise Except.error. When a channel is closed:

• no new values can be sent successfully anymore
• all blocked consumers are resolved to none (as no new messages can be sent they will never resolve)
• if there are already values waiting to be received they can still be received by subsequent recv calls

@[inline]

def Std.CloseableChannel.Sync.isClosed {α : Type} (ch : Sync α) : BaseIO Bool

Return true if the channel is closed.

@[inline]

def Std.CloseableChannel.Sync.tryRecv {α : Type} (ch : Sync α) : BaseIO (Option α)

Try to receive a value from the channel, if this can be completed right away without blocking return some value, otherwise return none.

def Std.CloseableChannel.Sync.recv {α : Type} (ch : Sync α) : BaseIO (Option α)

Receive a value from the channel, blocking until the transmission could be completed. Note that the return value may be none if the channel was closed before it could be completed.

instance Std.CloseableChannel.Sync.instForInOfMonadOfMonadLiftTBaseIO {m : Type → Type u_1} {α : Type} [Monad m] [MonadLiftT BaseIO m] : ForIn m (Sync α) α

for msg in ch.sync do ... receives all messages in the channel until it is closed.

structure Std.Channel (α : Type) : Type

A multi-producer multi-consumer FIFO channel that offers both bounded and unbounded buffering and an asynchronous API. To switch into synchronous mode use Channel.sync.

If a channel needs to be closed to indicate some sort of completion event use Std.CloseableChannel instead. Note that Std.CloseableChannel introduces a need for error handling in some cases, thus Std.Channel is usually easier to use if applicable.

• inner : Std.CloseableChannel α

instance Std.instNonemptyChannel {α✝ : Type} : Nonempty (Channel α✝)

def Std.Channel.Sync (α : Type) : Type

A multi-producer multi-consumer FIFO channel that offers both bounded and unbounded buffering and a synchronous API. This type acts as a convenient layer to use a channel in a blocking fashion and is not actually different from the original channel.

If a channel needs to be closed to indicate some sort of completion event use Std.CloseableChannel.Sync instead. Note that Std.CloseableChannel.Sync introduces a need for error handling in some cases, thus Std.Channel.Sync is usually easier to use if applicable.

instance Std.instNonemptySync_1 {α : Type} : Nonempty (Channel.Sync α)

@[inline]

def Std.Channel.new {α : Type} (capacity : Option Nat := none) : BaseIO (Channel α)

Create a new channel. If:

• capacity is none it will be unbounded (the default)
• capacity is some 0 it will always force a rendezvous between sender and receiver
• capacity is some n with n > 0 it will use a buffer of size n and begin blocking once it is filled

@[inline]

def Std.Channel.trySend {α : Type} (ch : Channel α) (v : α) : BaseIO Bool

Try to send a value to the channel. If this can be completed right away without blocking return true; otherwise, don't send the value and return false.

def Std.Channel.send {α : Type} (ch : Channel α) (v : α) : BaseIO (Task Unit)

Send a value through the channel, returning a task that will resolve once the transmission could be completed.

@[inline]

def Std.Channel.tryRecv {α : Type} (ch : Channel α) : BaseIO (Option α)

Try to receive a value from the channel, if this can be completed right away without blocking return some value, otherwise return none.

def Std.Channel.recv {α : Type} [Inhabited α] (ch : Channel α) : BaseIO (Task α)

Receive a value from the channel, returning a task that will resolve once the transmission could be completed. Note that the task may resolve to none if the channel was closed before it could be completed.

def Std.Channel.recvSelector {α : Type} [Inhabited α] (ch : Channel α) : Internal.IO.Async.Selector α

Creates a Selector that resolves once ch has data available and provides that data.

partial def Std.Channel.forAsync {α : Type} [Inhabited α] (f : α → BaseIO Unit) (ch : Channel α) (prio : Task.Priority := Task.Priority.default) : BaseIO (Task Unit)

ch.forAsync f calls f for every message received on ch.

Note that if this function is called twice, each message will only arrive at exactly one invocation.

instance Std.Channel.instAsyncStreamOfInhabited {α : Type} [Inhabited α] : Internal.Async.IO.AsyncStream (Channel α) α

instance Std.Channel.instAsyncReadOfInhabited {α : Type} [Inhabited α] : Internal.Async.IO.AsyncRead (Channel α) α

instance Std.Channel.instAsyncWriteOfInhabited {α : Type} [Inhabited α] : Internal.Async.IO.AsyncWrite (Channel α) α

@[inline]

def Std.Channel.sync {α : Type} (ch : Channel α) : Sync α

This function is a no-op and just a convenient way to expose the synchronous API of the channel.

@[inline]

def Std.Channel.Sync.new {α : Type} (capacity : Option Nat := none) : BaseIO (Sync α)

Create a new channel. If:

• capacity is none it will be unbounded (the default)
• capacity is some 0 it will always force a rendezvous between sender and receiver
• capacity is some n with n > 0 it will use a buffer of size n and begin blocking once it is filled

@[inline]

def Std.Channel.Sync.trySend {α : Type} (ch : Sync α) (v : α) : BaseIO Bool

Try to send a value to the channel. If this can be completed right away without blocking return true; otherwise, don't send the value and return false.

def Std.Channel.Sync.send {α : Type} (ch : Sync α) (v : α) : BaseIO Unit

Send a value through the channel, blocking until the transmission could be completed.

@[inline]

def Std.Channel.Sync.tryRecv {α : Type} (ch : Sync α) : BaseIO (Option α)

Try to receive a value from the channel, if this can be completed right away without blocking return some value, otherwise return none.

def Std.Channel.Sync.recv {α : Type} [Inhabited α] (ch : Sync α) : BaseIO α

Receive a value from the channel, blocking until the transmission could be completed.

instance Std.Channel.Sync.instForInOfInhabitedOfMonadOfMonadLiftTBaseIO {α : Type} {m : Type → Type u_1} [Inhabited α] [Monad m] [MonadLiftT BaseIO m] : ForIn m (Sync α) α

for msg in ch.sync do ... receives all messages in the channel until it is closed.