// A one-shot result that any number of tasks may await.

///|
priv struct Completion {
  done : @ref.Ref[Bool]
  error : @ref.Ref[Error?]
  signal : @async.Queue[Unit]
}

///|
fn Completion::new() -> Completion {
  { done: Ref(false), error: Ref(None), signal: Queue(kind=Unbounded), }
}

///|
fn Completion::complete(self : Completion, error : Error?) -> Unit {
  if self.done.val {
    return
  }
  self.error.val = error
  self.done.val = true
  self.signal.close()
}

///|
async fn Completion::wait(self : Completion) -> Unit {
  if !self.done.val {
    ignore(self.signal.get()) catch {
      @aqueue.QueueAlreadyClosed => ()
      error => raise error
    }
  }
  if self.error.val is Some(error) {
    raise error
  }
}

///|
priv struct DrainControl {
  queue : @async.Queue[async () -> Unit]
  pending : Array[DrainCompletion]
  next_id : @ref.Ref[Int32]
  closed : @ref.Ref[Bool]
}

///|
priv struct DrainCompletion {
  id : Int32
  completion : Completion
}

///|
fn DrainControl::new() -> DrainControl {
  {
    queue: Queue(kind=Unbounded),
    pending: [],
    next_id: Ref(0),
    closed: Ref(false),
  }
}

///|
fn DrainControl::finish(
  self : DrainControl,
  id : Int32,
  completion : Completion,
  error : Error?,
) -> Unit {
  completion.complete(error)
  for i in 0.. Unit {
  if self.closed.val {
    return
  }
  self.closed.val = true
  self.queue.close(clear=true)
  while self.pending.length() > 0 {
    let item = self.pending.remove(self.pending.length() - 1)
    item.completion.complete(Some(error))
  }
}