RXSwift - 忽略 、 获取事件操作符

---

enum E : Error {
    case demoAError
    case demoBError
}

ignoreElements ：忽略序列中所有的 .next事件。

        let tasks = PublishSubject<String>()
        
        _ = tasks.ignoreElements().subscribe(onCompleted: {
            print("onCompleted")
        }) { (_) in
            print("onError")
        }
        
        tasks.onNext("1")
        tasks.onNext("2")
        tasks.onNext("3")
        tasks.onNext("4")
        tasks.onCompleted()

skip ：忽略特定个数的 .next 事件。

        let tasks = PublishSubject<String>()
        
        _ = tasks.skip(2).subscribe(onNext: { (str) in
            print(str)
        }, onError: { (_) in
            print("onError")
        }, onCompleted: {
            print("onCompleted")
        }, onDisposed: {
            print("onDisposed")
        })
        
        tasks.onNext("1")
        tasks.onNext("2")
        tasks.onNext("3")
        tasks.onNext("4")
        tasks.onCompleted()

skipWhile ：当遇到第一个不满足条件的事件，就开始监听 ， 第一个不满足条件的事件也会打印。

        let tasks = PublishSubject<String>()

        _ = tasks.skipWhile({ (str) -> Bool in
            return str != "2"
        }).subscribe(onNext: { (str) in
            print(str)
        }, onError: { (_) in
            print("onError")
        }, onCompleted: {
            print("onCompleted")
        }, onDisposed: {
            print("onDisposed")
        })
        
        tasks.onNext("1")
        tasks.onNext("2")
        tasks.onNext("3")
        tasks.onNext("4")
        tasks.onCompleted()

skipUntil ：当遇见 另一个observable序列中的事件, 就开始监听自身的事件,只打印自身的事件。

        let tasksTwo = PublishSubject<String>()
        let tasks = PublishSubject<String>()

        _ = tasks.skipUntil(tasksTwo).subscribe(onNext: { (str) in
            print(str)
        }, onError: { (_) in
            print("onError")
        }, onCompleted: {
            print("onCompleted")
        }, onDisposed: {
            print("onDisposed")
        })
        
        tasks.onNext("1")
        tasks.onNext("2")
        tasksTwo.onNext("one")
        tasks.onNext("3")
        tasks.onNext("4")
        tasks.onCompleted()
        tasksTwo.onCompleted()

distinctUntilChange： 忽略序列中连续重复的事件。

        let tasks = PublishSubject<String>()

        _ = tasks.distinctUntilChanged().subscribe(onNext: { (str) in
            print(str)
        }, onError: { (_) in
            print("onError")
        }, onCompleted: {
            print("onCompleted")
        }, onDisposed: {
            print("onDisposed")
        })
        
        tasks.onNext("1")
        tasks.onNext("1")
        tasks.onNext("3")
        tasks.onNext("4")
        tasks.onCompleted()

elementAt ：选择序列中的第n个事件。

let tasks = PublishSubject<String>()
        
        _ = tasks.elementAt(2).subscribe(onNext: { (Str) in
            print(Str)
        }, onError: { (_) in
            print("onError")
        }, onCompleted: {
            print("onCompleted")
        }, onDisposed: {
            print("onDisposed")
        })
        
        tasks.onNext("1")
        tasks.onNext("1")
        tasks.onNext("3")
        tasks.onNext("4")
        tasks.onCompleted()

filter ：通过closure自定义过滤条件。

        let tasks = PublishSubject<String>()
        
       _ = tasks.filter { (str) -> Bool in
            return str == "1"
        }.subscribe(onNext: { (str) in
            print(str)
        }, onError: { (_) in
            print("onError")
        }, onCompleted: {
            print("onCompleted")
        }) {
            print("onDisposed")
        }
        
        
        tasks.onNext("1")
        tasks.onNext("1")
        tasks.onNext("3")
        tasks.onNext("4")
        tasks.onCompleted()

take ：订阅指定位置的事件外，我们还能一次性订阅多个事件。take (2) 表示订阅序列中 的前两个字段。

        let tasks = PublishSubject<String>()
        
        _ = tasks.take(2).subscribe(onNext: { (str) in
            print(str)
        }, onError: { (_) in
            print("onError")
        }, onCompleted: {
            print("onCompleted")
        }) {
            print("onDisposed")
        }
        
        tasks.onNext("1")
        tasks.onNext("2")
        tasks.onNext("3")
        tasks.onNext("4")
        tasks.onCompleted()

takeWhile ：用一个 closure 指定 “只要条件为true”就一直订阅下去。

        let tasks = PublishSubject<String>()

        _ = tasks.takeWhile({ (str) -> Bool in
            return str != "3"
        }).subscribe(onNext: { (str) in
            print(str)
        }, onError: { (_) in
            print("onError")
        }, onCompleted: {
            print("onCompleted")
        }) {
            print("onDisposed")
        }
        
        
        tasks.onNext("1")
        tasks.onNext("2")
        tasks.onNext("3")
        tasks.onNext("4")
        tasks.onCompleted()

takeUntil ： 表示知道里一个序列发生之前，一直订阅着当前序列。如果另一个序列发生事件，订阅结束。

        let tasks = PublishSubject<String>()
        let tasksTwo = PublishSubject<String>()
        
        _ = tasks.takeUntil(tasksTwo).subscribe(onNext: { (str) in
            print(str)
        }, onError: { (_) in
            print("onError")
        }, onCompleted: {
            print("onCompleted")
        }) {
            print("onDisposed")
        }
        
        tasks.onNext("1")
        tasks.onNext("2")
        tasksTwo.onNext("one")
        tasks.onNext("3")
        tasks.onNext("4")
        tasks.onCompleted()
        tasksTwo.onCompleted()