Sampler.swift

extension Signal where Error == NoError {
    public func sample(each timeInterval: DispatchTimeInterval,
                       on scheduler: DateScheduler = QueueScheduler()) -> Signal<[Value], NoError> {
        return Signal<[Value], NoError> { observer in
            let valueHolder : MutableProperty<[Value]> = MutableProperty<Value>([])
            let disposable = CompositeDisposable()
            var sampler: Signal<(), NoError>? = nil
            disposable += timer(interval: until, on: QueueScheduler()).startWithSignal { signal, _ in
                signal = innerSignal.map { _ in () }
            }
            disposable += self.observeValues { value in
                valueHolder.value = valueHolder.value + [value]
            }
            disposable += valueHolder.signal.sample(on: sampler).observe { (event: Event<Value, NoError>) in
                switch event {
                case let .value(value):
                    observer.send(valueHolder.value)
                    valueHolder.value = []
                case .completed:
                    observer.sendCompleted()
                case let .failed(error):
                    observer.send(error: error)
                case .interrupted:
                    observer.sendInterrupted()
                }
            }
            return disposable
        }
    }
}