Signal.zipWith

zipWith.swift

private enum Either<T, U> {
	case Left(Box<T>)
	case Right(Box<U>)
}

public func zipWith<T, U, E>(otherSignal: Signal<U, E>)(signal: Signal<T, E>) -> Signal<(T, U), E> {
	return Signal { observer in
		var lock = NSRecursiveLock()
		lock.name = "org.reactivecocoa.ReactiveCocoa.zipWith"

		var buffer: [Either<T, U>] = []
		
		let onNext = { (zipper: () -> (T, U)?) -> () in
			var zipped: (T, U)?
			lock.lock()
			zipped = zipper()
			lock.unlock()

			if let zipped = zipped {
				sendNext(observer, zipped)
			}
		}

		let onError = { sendError(observer, $0) }
		let onCompleted = { sendCompleted(observer) }
		
		let signalDisposable = signal.observe(next: { value in
			onNext {
				switch buffer.first {
				case .Some(.Right(let box)):
					buffer.removeAtIndex(0)
					return (value, box.unbox)
				default:
					buffer.append(.Left(Box(value)))
					return nil
				}
			}
		}, error: onError, completed: onCompleted)
		
		let otherDisposable = otherSignal.observe(next: { value in
			onNext {
				switch buffer.first {
				case .Some(.Left(let box)):
					buffer.removeAtIndex(0)
					return (box.unbox, value)
				default:
					buffer.append(.Right(Box(value)))
					return nil
				}
			}
		}, error: onError, completed: onCompleted)
		
		return CompositeDisposable([ signalDisposable, otherDisposable ])
	}
}