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.

instance Std.Broadcast.instReprError : Repr Error

Equations

• Std.Broadcast.instReprError = { reprPrec := Std.Broadcast.instReprError.repr }

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

Equations

• One or more equations did not get rendered due to their size.
• Std.Broadcast.instReprError.repr Std.Broadcast.Error.closed prec✝ = Repr.addAppParen (Std.Format.nest (if prec✝ ≥ 1024 then 1 else 2) (Std.Format.text "Std.Broadcast.Error.closed")).group prec✝

instance Std.Broadcast.instDecidableEqError : DecidableEq Error

Equations

• Std.Broadcast.instDecidableEqError x✝ y✝ = if h : x✝.ctorIdx = y✝.ctorIdx then isTrue ⋯ else isFalse ⋯

instance Std.Broadcast.instHashableError : Hashable Error

Equations

• Std.Broadcast.instHashableError = { hash := Std.Broadcast.instHashableError.hash }

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

Equations

• Std.Broadcast.instHashableError.hash Std.Broadcast.Error.closed = 0
• Std.Broadcast.instHashableError.hash Std.Broadcast.Error.alreadyClosed = 1
• Std.Broadcast.instHashableError.hash Std.Broadcast.Error.notSubscribed = 2

instance Std.instToStringBroadcastError : ToString Broadcast.Error

Equations

• One or more equations did not get rendered due to their size.

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

Equations

• Std.instMonadLiftBroadcastIO = { monadLift := fun {α : Type} (x : EIO Std.Broadcast.Error α) => EIO.toIO (fun (x : Std.Broadcast.Error) => IO.Error.userError (toString x)) x }

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.

Equations

• Std.Broadcast.new capacity h = do let __do_lift ← Std.Bounded.new✝ capacity h pure { inner := __do_lift }

@[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.

Equations

• ch.trySend v = Std.Bounded.trySend✝ (Std.Broadcast.inner✝ ch) v

@[inline]

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

Subscribes a new Receiver from the Broadcast channel.

Equations

• ch.subscribe = Std.Broadcast.Receiver.mk✝ <$> Std.Bounded.subscribe✝ (Std.Broadcast.inner✝ ch)

@[inline]

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

Closes a Broadcast channel.

Equations

• ch.close = liftM (Std.Bounded.close✝ (Std.Broadcast.inner✝ ch))

@[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.

Equations

• One or more equations did not get rendered due to their size.

@[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.

Equations

• ch.tryRecv = Std.Bounded.Receiver.tryRecv✝ (Std.Broadcast.Receiver.inner✝ ch)

@[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.

Equations

• ch.recv = Std.Bounded.Receiver.recv✝ (Std.Broadcast.Receiver.inner✝ ch)

@[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 that data.

Equations

• ch.recvSelector = Std.Bounded.Receiver.recvSelector✝ (Std.Broadcast.Receiver.inner✝ ch)

@[inline]

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

Unsubscribes a Receiver from the Broadcast channel.

Equations

• ch.unsubscribe = Std.Bounded.Receiver.unsubscribe✝ (Std.Broadcast.Receiver.inner✝ ch)

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.

Equations

• Std.Broadcast.Receiver.forAsync f ch prio = Std.Bounded.Receiver.forAsync✝ f (Std.Broadcast.Receiver.inner✝ ch) prio

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

Equations

• One or more equations did not get rendered due to their size.

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

Equations

• Std.Broadcast.Receiver.instAsyncReadOptionOfInhabited = { read := fun (receiver : Std.Broadcast.Receiver α) => Std.Internal.IO.Async.Async.ofIOTask (liftM receiver.recv) }

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

Equations

• One or more equations did not get rendered due to their size.

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.

Equations

• Std.Broadcast.Sync α = Std.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.

Equations

• Std.Broadcast.Sync.Receiver α = Std.Broadcast.Receiver α

@[inline]

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

Creates a new broadcast channel.

Equations

• Std.Broadcast.Sync.new capacity h = Std.Broadcast.new capacity h

@[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.

Equations

• ch.trySend v = Std.Broadcast.trySend ch v

@[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.

Equations

• ch.send v = do let __do_lift ← liftM (Std.Broadcast.send ch v) IO.ofExcept =<< liftM (IO.wait __do_lift)

@[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.

Equations

• ch.tryRecv = Std.Broadcast.Receiver.tryRecv ch

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.

Equations

• ch.recv = do let __do_lift ← Std.Broadcast.Receiver.recv ch IO.wait __do_lift

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 β

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

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

Equations

• One or more equations did not get rendered due to their size.