STM.Queue

A transactional first-in, first-out queue, unbounded or bounded, after Haskell’s TQueue and TBQueue.

Popping from an empty queue or pushing onto a full bounded one retries, so these operations block inside STM.atomically until another task’s commit makes them possible. They combine with other transactions, and STM.orElse chooses between queues:

import Async exposing (Outcome(..))
import STM
import STM.Queue as Queue

result = Async.runOutcome {
    urgent = STM.atomically { Queue.unbounded () }
    normal = STM.atomically { Queue.unbounded () }
    STM.atomically { Queue.push normal "report"; Queue.push urgent "page" }
    next = { _ -> STM.atomically { STM.orElse { Queue.pop urgent } { Queue.pop normal } } }
    first = next ()
    second = next ()
    (first, second)
    }
result == Completed ("page", "report")

The queue is two lists in separate variables, so producers and consumers do not conflict while the front holds values; a bounded queue also splits its free-slot count between them. A queue cannot be closed: pair it with an STM.Var Bool and STM.orElse when consumers must learn that no more values will come.

Queue

type Queue aSource

A queue of a values. It may be shared by any number of tasks and has no equality or Show.

bounded

bounded : Int ->{STM} Queue aSource

An empty queue holding at most capacity values; a capacity below 1 is treated as 1.

import Async exposing (Outcome(..))
import STM
import STM.Queue as Queue

result = Async.runOutcome { STM.atomically { queue = Queue.bounded 1; first = Queue.tryPush queue 'a'; (first, Queue.tryPush queue 'b') } }
result == Completed (True, False)

drain

drain : Queue a ->{STM} List aSource

Removes and answers every value, oldest first, without retrying.

import Async exposing (Outcome(..))
import STM
import STM.Queue as Queue

result = Async.runOutcome { STM.atomically { queue = Queue.bounded 3; Queue.push queue 1; Queue.push queue 2; values = Queue.drain queue; (values, Queue.isEmpty queue) } }
result == Completed ([1, 2], True)

isEmpty

isEmpty : Queue a ->{STM} BoolSource

Whether the queue holds no values.

import Async exposing (Outcome(..))
import STM
import STM.Queue as Queue

result = Async.runOutcome { STM.atomically { queue = Queue.unbounded (); Queue.push queue (); Queue.isEmpty queue } }
result == Completed False

peek

peek : Queue a ->{STM} aSource

The oldest value, leaving it in the queue; retries while the queue is empty.

import Async exposing (Outcome(..))
import STM
import STM.Queue as Queue

result = Async.runOutcome { STM.atomically { queue = Queue.unbounded (); Queue.push queue "a"; first = Queue.peek queue; (first, Queue.pop queue) } }
result == Completed ("a", "a")

pop

pop : Queue a ->{STM} aSource

Removes and answers the oldest value; retries while the queue is empty.

import Async exposing (Outcome(..))
import STM
import STM.Queue as Queue

result = Async.runOutcome { STM.atomically { queue = Queue.unbounded (); Queue.push queue 1; Queue.push queue 2; first = Queue.pop queue; (first, Queue.pop queue) } }
result == Completed (1, 2)

push

push : Queue a -> a ->{STM} ()Source

Adds value at the back; retries while a bounded queue is full.

import Async exposing (Outcome(..))
import STM
import STM.Queue as Queue

result = Async.runOutcome { STM.atomically { queue = Queue.unbounded (); Queue.push queue 1; Queue.push queue 2; Queue.drain queue } }
result == Completed [1, 2]

tryPeek

tryPeek : Queue a ->{STM} Maybe aSource

The oldest value if there is one, leaving it in the queue, without retrying.

import Async exposing (Outcome(..))
import Maybe exposing (Maybe(..))
import STM
import STM.Queue as Queue

result = Async.runOutcome { STM.atomically { queue = Queue.unbounded (); Queue.push queue 3; Queue.tryPeek queue } }
result == Completed (Just 3)

tryPop

tryPop : Queue a ->{STM} Maybe aSource

Removes and answers the oldest value if there is one, without retrying.

import Async exposing (Outcome(..))
import Maybe exposing (Maybe(..))
import STM
import STM.Queue as Queue

result = Async.runOutcome { STM.atomically { queue = Queue.unbounded (); Queue.push queue 1; first = Queue.tryPop queue; (first, Queue.tryPop queue) } }
result == Completed (Just 1, Nothing)

tryPush

tryPush : Queue a -> a ->{STM} BoolSource

Adds value at the back unless a bounded queue is full, answering whether it did, without retrying.

import Async exposing (Outcome(..))
import STM
import STM.Queue as Queue

result = Async.runOutcome { STM.atomically { queue = Queue.bounded 2; added = Queue.tryPush queue "x"; (added, Queue.drain queue) } }
result == Completed (True, ["x"])

unbounded

unbounded : () ->{STM} Queue aSource

An empty queue with no capacity limit.

import Async exposing (Outcome(..))
import STM
import STM.Queue as Queue

Async.runOutcome { STM.atomically { Queue.isEmpty (Queue.unbounded ()) } } == Completed True