ValueAsync

valueasync.nim

type
  ValueAsync*[T] = ref object
    sigPut, sigPop: SignalAsync
    val: T
    isClosed: bool

func newValueAsync*[T](): ValueAsync[T] =
  ValueAsync[T](
    sigPut: newSignal(),
    sigPop: newSignal(),
    val: nil,
    isClosed: false
  )

proc put*[T](vala: ValueAsync[T], val: T) {.async.} =
  if vala.isClosed:
    raise newValueAsyncClosedError()
  while vala.val != nil:
    await vala.sigPut.waitFor()
  vala.val = val
  vala.sigPop.trigger()

proc pop*[T](vala: ValueAsync[T]): Future[T] {.async.}
  if vala.isClosed:
    raise newValueAsyncClosedError()
  while vala.val == nil:
    await vala.sigPop.waitFor()
  result = vala.val
  vala.val = nil
  vala.sigPut.trigger()

proc close*[T](vala: ValueAsync[T]) {.raises: [].}  =
  if vala.isClosed:
    return
  vala.isClosed = true
  vala.sigPut.close()
  vala.sigPop.close()

when isMainModule:

  func newIntRef(n: int): ref int =
    new result
    result[] = n
  block:
    proc test() {.async.} =
      var q = newValueAsync[ref int]()
      await q.put newIntRef(1)
      doAssert (await q.pop())[] == 1
      await q.put newIntRef(2)
      doAssert (await q.pop())[] == 2
      await q.put newIntRef(3)
      doAssert (await q.pop())[] == 3
      await q.put newIntRef(4)
      doAssert (await q.pop())[] == 4
    waitFor test()
    doAssert not hasPendingOperations()
  block:
    proc test() {.async.} =
      var q = newValueAsync[ref int]()
      proc puts {.async.} =
        await q.put newIntRef(1)
        await q.put newIntRef(2)
        await q.put newIntRef(3)
        await q.put newIntRef(4)
      let puts1 = puts()
      doAssert (await q.pop())[] == 1
      doAssert (await q.pop())[] == 2
      doAssert (await q.pop())[] == 3
      doAssert (await q.pop())[] == 4
      await puts1
    waitFor test()
    doAssert not hasPendingOperations()