Non-blocking queue in RAC 3

Enqueue.swift

import ReactiveCocoa

public func enqueue<T>(elements: SignalProducer<T, NoError>) -> SignalProducer<T, NoError> {
  let (poppers, poppersSink) = Signal<SinkOf<Event<T, NoError>>, NoError>.pipe()
  
  elements
    |> zipWith(poppers)
    |> start(next: { element, popper in
      sendNext(popper, element)
      sendCompleted(popper)
    })
  
  let poppersDisposable = ScopedDisposable(ActionDisposable {
    sendCompleted(poppersSink)
  })
  
  return SignalProducer { observer, _ in
    sendNext(poppersSink, observer)
    poppersDisposable
  }
}

Main.swift

let queue = SignalProducer(values: [1, 2]) |> enqueue

// Pop
queue.start(next: println) // => 1

// Pop again
queue.start(next: println) // => 2