rxswift-min-value

rxswift-min-operator.swift

struct MyStruct: Comparable {
    var a: Int
    public static func <(lhs: MyStruct, rhs: MyStruct) -> Bool {
        return lhs.a < rhs.a
    }
}

extension ObservableType {
    
    /**
     Takes a sequence of optional elements and returns a sequence of non-optional elements, filtering out any nil values.
     
     - returns: An observable sequence of non-optional elements
     */
    
    public func unwrap<T>() -> Observable<T> where E == T? {
        return self.filter { $0 != nil }.map { $0! }
    }
}

extension ObservableType {
    func min(by areIncreasingOrder: @escaping (Self.E, Self.E) -> Bool) -> Observable<Self.E> {
        return reduce(Optional<E>.none, accumulator: { (min, e) -> Optional<E> in
            if let m = min, areIncreasingOrder(m, e) {
                return m
            }
            return e
        }).unwrap()
    }
}
extension ObservableType where E: Comparable {
    
    private var accumulator: ((Optional<E>, E) -> Optional<E>) {
        return { min, e -> Optional<E> in
            if let m = min, m < e  {
                return m
            }
            return e
        }
    }
    
    func min() -> Observable<Self.E>  {
        return reduce(Optional<E>.none, accumulator: { self.accumulator($0, $1) }).unwrap()
    }
    
    func accumMin() -> Observable<Self.E>  {
        return scan(Optional<E>.none, accumulator: { (min, e) -> Optional<E> in
            if let m = min, m < e {
                return m
            }
            return e
        }).unwrap()
    }
}

let publisher = PublishSubject<MyStruct>.init()
publisher.asObserver().min { $0 < $1 }.debug().subscribe()
publisher.onNext(MyStruct(a: 3))
publisher.onNext(MyStruct(a: 4))
publisher.onNext(MyStruct(a: 1))
publisher.onNext(MyStruct(a: 8))
publisher.onCompleted()