take

abstract fun take(): Kind<F, A>

Takes and removes a value from the Queue, or semantically blocks until a value becomes available.

import arrow.fx.*
import arrow.fx.extensions.fx

//sampleStart
suspend fun main(args: Array<String>): Unit = IO.fx {
  val queue = !Queue.unbounded<Int>()
  val (join, _) = !queue.take().fork()
  !queue.offer(1) // Removing this offer makes, !join block forever.
  val res = !join // Join the blocking take, after we offered a value
  !effect { println(res) }
}.suspended()
//sampleEnd

See Also

peek

tryTake