The Std.Sync.Broadcast module implements a broadcasting primitive for sending values to multiple consumers. It maintains a queue of values and supports both synchronous and asynchronous waiting.

This module is heavily inspired by Std.Sync.Channel as well as tokio’s broadcast implementation.

inductive Std.Broadcast.Error : Type

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

• closed : Error

  Tried to send to a closed broadcast channel.

• alreadyClosed : Error

  Tried to close an already closed broadcast channel.

• notSubscribed : Error

  Tried to unsubscribe a channel that already is not part of it.

@[implicit_reducible]

instance Std.Broadcast.instReprError : Repr Error

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

@[implicit_reducible]

instance Std.Broadcast.instDecidableEqError : DecidableEq Error

@[implicit_reducible]

instance Std.Broadcast.instHashableError : Hashable Error

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

@[implicit_reducible]

instance Std.instToStringBroadcastError : ToString Broadcast.Error

@[implicit_reducible]

instance Std.instMonadLiftBroadcastIO : MonadLift (EIO Broadcast.Error) IO

structure Std.Broadcast (α : Type) : Type

A multi-subscriber broadcast that delivers each message to all current subscribers. Supports only bounded buffering and an asynchronous API; to switch into synchronous mode use Broadcast.sync.

Unlike Std.Channel, each message is received by every subscriber instead of just one. Subscribers only receive messages sent after they have subscribed (unless otherwise specified).

• inner : Std.Bounded✝ α

structure Std.Broadcast.Receiver (α : Type) : Type

A receiver for a Broadcast channel that can asynchronously receive messages. Each receiver gets a copy of every message sent to the broadcast channel after the receiver was created. Multiple receivers can exist for the same broadcast, and each will receive all messages independently.

• inner : Std.Bounded.Receiver✝ α

@[inline]

def Std.Broadcast.new {α : Type} (capacity : Nat := 16) (h : capacity > 0 := by decide) : BaseIO (Broadcast α)

Creates a new broadcast channel.

@[inline]

def Std.Broadcast.trySend {α : Type} (ch : Broadcast α) (v : α) : BaseIO (Option Nat)

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

@[inline]

def Std.Broadcast.subscribe {α : Type} (ch : Broadcast α) : IO (Receiver α)

Subscribes a new Receiver from the Broadcast channel.

@[inline]

def Std.Broadcast.close {α : Type} (ch : Broadcast α) : IO Unit

Closes a Broadcast channel.

@[inline]

def Std.Broadcast.send {α : Type} (ch : Broadcast α) (v : α) : BaseIO (Task (Except IO.Error Nat))

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

@[inline]

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

Try to receive a value from the broadcast receiver, if a message is available right away return some value, otherwise return none without blocking.

@[inline]

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

Receive a value from the broadcast receiver, returning a task that will resolve with the next available message. This will block until a message is available.

@[inline]

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

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

@[inline]

def Std.Broadcast.Receiver.unsubscribe {α : Type} (ch : Receiver α) : IO Unit

Unsubscribes a Receiver from the Broadcast channel.

def Std.Broadcast.Receiver.forAsync {α : Type} (f : α → BaseIO Unit) (ch : Receiver α) (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.

@[implicit_reducible]

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

@[implicit_reducible]

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

@[implicit_reducible]

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

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

A multi-subscriber broadcast that delivers each message to all current subscribers. Supports only bounded buffering and an asynchronous API.

It's the sync version of Broadcast.

def Std.Broadcast.Sync.Receiver (α : Type) : Type

A receiver for a Broadcast channel that can asynchronously receive messages. Each receiver gets a copy of every message sent to the broadcast channel after the receiver was created. Multiple receivers can exist for the same broadcast, and each will receive all messages independently.

It's the sync version of Broadcast.Receiver.

@[inline]

def Std.Broadcast.Sync.new {α : Type} (capacity : Nat := 16) (h : capacity > 0 := by decide) : BaseIO (Sync α)

Creates a new broadcast channel.

@[inline]

def Std.Broadcast.Sync.trySend {α : Type} (ch : Sync α) (v : α) : BaseIO (Option Nat)

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

@[inline]

def Std.Broadcast.Sync.send {α : Type} (ch : Sync α) (v : α) : IO Nat

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

@[inline]

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

Try to receive a value from the broadcast receiver, if a message is available right away return some value, otherwise return none without blocking.

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

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

partial def Std.Broadcast.Sync.Receiver.forIn {α : Type} {m : Type → Type u_1} {β : Type} [Inhabited α] [Monad m] [MonadLiftT BaseIO m] (ch : Receiver α) (f : α → β → m (ForInStep β)) : β → m β

@[implicit_reducible]

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

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