UpdatableArray.swift

import Foundation
import RxSwift
import RxRelay

final class UpdatableArray<T>: ObservableType {
    enum Action {
        case replace([T])
        case append([T])
        case filter((T) -> Bool)

        fileprivate func reduce(previous: @autoclosure () -> [T]) -> [T] {
            switch self {
            case let .replace(newArray):
                return newArray
            case let .filter(condition):
                return previous().filter(condition)
            case let .append(newValue):
                return previous() + newValue
            }
        }
    }

    private let _array: BehaviorRelay<[T]>
    let actionPublisher = PublishRelay<Action>()
    private let disposeBag = DisposeBag()

    init(value: [T] = []) {
        _array = BehaviorRelay(value: value)

        actionPublisher
            .observeOn(SerialDispatchQueueScheduler(qos: .default))
            .subscribe(onNext: { [weak self] action in
                guard let self = self else {
                    return
                }
                let new = action.reduce(previous: self._array.value)
                self._array.accept(new)
            })
            .disposed(by: disposeBag)
    }

    typealias Element = [T]

    func subscribe<Observer>(_ observer: Observer) -> RxSwift.Disposable where Observer: RxSwift.ObserverType, Element == Observer.Element {
        return _array.subscribe(observer)
    }
}