swift - RxSwift，计算发出了多少个相等的连续 Equatable 项

Question

我有一个 Observable，它发出一系列 Equatable 元素。 流可以包含相同元素的连续序列(例如在序列 [1, 1, 1, 18, 2, 2, 0, -1] 中，元素 1 重复 3 次，2 重复 2 次)。 我需要压缩序列，用包含该元素和原始流中存在的重复次数的元组替换每个不同元素:

1, 1, 1, 18, 21, 21, 0, -1, -8, -8, 14, 14, 14...

(1, 3), (18, 1), (21, 2), (0, 1), (-1, 1), (-8, 2), (14, 3)...

我设法使用 scan 运算符对重复次数进行计数，但它发出了不应成为最终序列一部分的所有部分计算:

let numbers = Observable<Int>.from([
    0, 0,
    3, 3, 3, 3, 3,
    2,
    0, 0, 0,
    6, 6,
    ])

let reps = numbers
    .scan((0, 0), accumulator: {
        (prev: (Int, Int), new: (Int)) in
        if prev.0 == new {
            return (new, prev.1 + 1)
        } else {
            return (new, 1)
        }
    })

reps.subscribe(onNext: {
        print("\($0)")
    })

// expected:
// (0, 2), (3, 5), (2, 1), (0, 3), (6, 2)
//
// result:
// (0, 1), (0, 2),
// (3, 1), (3, 2), (3, 3), (3, 4), (3, 5),
// (2, 1),
// (0, 1), (0, 2), (0, 3),
// (6, 1), (6, 2)

Answer 1

您可以编写自己的运算符。这是一个示例实现。我创建了一个新的可观察对象，它订阅 self 的事件。每当 self 有一个新元素时，switch 中的 .next 情况就会被触发，这会保留游程长度预订。每当遇到不同的元素、错误或完成时，就会发出分组。

extension ObservableType where Self.E: Equatable {
    func runLengthEncode() -> Observable<(element: E, count: Int)> {
        var lastGrouping: (element: E, count: Int)? = nil

        return Observable.create { observer in
            return self.subscribe { event in
                switch event {
                case .next(let currentElement):
                    if let currentGrouping = lastGrouping {
                        if currentGrouping.element == currentElement {
                            lastGrouping = (element: currentElement, count: currentGrouping.count + 1)
                        }
                        else { // This run ended, a new element was encountered.
                            lastGrouping = (element: currentElement, count: 1) // start a new grouping
                            observer.on(.next(currentGrouping)) // emit the completed grouping
                        }
                    } else {
                        lastGrouping = (element: currentElement, count: 1)
                    }

                case .error(let error):
                    if let lastGrouping = lastGrouping { observer.on(.next(lastGrouping)) } // Emit the last unemitted grouping.
                    observer.on(.error(error))

                case .completed:
                    if let lastGrouping = lastGrouping { observer.on(.next(lastGrouping)) } // Emit the last unemitted grouping.
                    observer.on(.completed)
                }
            }
        }
    }
}

您还可以实现免费的游程长度解码运算符:

extension ObservableType {
    func runLengthDecode<Element>() -> Observable<Element>
        where Self.E == (element: Element, count: Int) {
        return Observable.create { observer in
            return self.subscribe { event in
                switch event {
                case .next((element: let element, count: let count)):
                    for _ in 1...count {
                        observer.on(.next(element))
                    }


                case .error(let error): observer.on(.error(error))
                case .completed: observer.on(.completed)
                }
            }
        }
    }
}

测试用例:

let numbers = Observable<Int>.from([
    0, 0,
    3, 3, 3, 3, 3,
    2,
    0, 0, 0,
    6, 6,
])

let runLengthEncoded = numbers.runLengthEncode()
runLengthEncoded.subscribe { print($0) }

let runLengthDecoded = runLengthEncoded.runLengthDecode()
runLengthDecoded.subscribe { print($0) }