我有一个 SignalProducer ProducerA,它以不同的时间间隔发出值。我正在尝试收集 SignalProducer 发出的最新 N 值并创建一个新的生产者 ProducerB,它发出包含最新 N 值的数组。
ProducerB 应在 ProducerA 发出前 N 个值时开始发出值,然后每次 ProducerA 发出新值时发出一个新数组。
有人可以帮助我吗?
let (producerA, observerA) = SignalProducer<Int, NoError>.buffer(5)
let n = 3
producerA.take(n).collect()
.takeUntilReplacement(producerA.skip(n).map { [$0] })
.scan([], { $0.suffix(n - 1) + $1 })
.startWithNext {
print($0)
}
observerA.sendNext(1) // nothing printed
observerA.sendNext(2) // nothing printed
observerA.sendNext(3) // prints [1, 2, 3]
observerA.sendNext(4) // prints [2, 3, 4]
observerA.sendNext(5) // prints [3, 4, 5]
我想出了这个代码
extension SignalProducer {
/// Creates a new producer that emits an array that contains the latest N values that were emitted
/// by the original producer as specified in 'capacity'.
@warn_unused_result(message="Did you forget to call `start` on the producer?")
public func latestValues(n:Int) -> SignalProducer<[Value], Error> {
var array: [Value] = []
return self.map {
value in
array.append(value)
if array.count >= n {
array.removeFirst(array.count - n)
}
return array
}
.filter {
$0.count == n
}
}
}
extension SignalProducer {
func latestValues(count: Int) -> SignalProducer<[Value], Error> {
SignalProducer<[Value], Error> { (observer: Signal<[Value], Error>.Observer, lifetime: Lifetime) in
var collectedValues: [Value] = []
let disposable = self
.start(Signal<Value, Error>.Observer(value: { value in
collectedValues.append(value)
if collectedValues.count > count {
collectedValues.removeFirst(collectedValues.count - count)
}
if collectedValues.count == count {
observer.send(value: collectedValues)
}
},
failed: { error in
observer.send(error: error)
},
completed: {
observer.sendCompleted()
},
interrupted: {
observer.sendInterrupted()
}))
lifetime.observeEnded {
disposable.dispose()
}
}
}
}
它发出一个包含最新 N 个值的数组。
测试:
let valuesGenerator = SignalProducer([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
.promoteError(Error.self)
valuesGenerator
.latestValues(count: 3)
.start(Signal.Observer(value: { value in
print("valuesGenerator, date: \(Date()), value: \(value)")
},
failed: { error in
print("valuesGenerator, date: \(Date()), error: \(error)")
},
completed: {
print("valuesGenerator, date: \(Date()), completed")
},
interrupted: {
print("valuesGenerator, date: \(Date()), interrupted")
}))
输出:
valuesGenerator, date: 2024-10-22 11:16:15 +0000, value: [1, 2, 3]
valuesGenerator, date: 2024-10-22 11:16:15 +0000, value: [2, 3, 4]
valuesGenerator, date: 2024-10-22 11:16:15 +0000, value: [3, 4, 5]
valuesGenerator, date: 2024-10-22 11:16:15 +0000, value: [4, 5, 6]
valuesGenerator, date: 2024-10-22 11:16:15 +0000, value: [5, 6, 7]
valuesGenerator, date: 2024-10-22 11:16:15 +0000, value: [6, 7, 8]
valuesGenerator, date: 2024-10-22 11:16:15 +0000, value: [7, 8, 9]
valuesGenerator, date: 2024-10-22 11:16:15 +0000, value: [8, 9, 10]
valuesGenerator, date: 2024-10-22 11:16:21 +0000, completed