RxSwift - 过滤操作符系列之 1

/*
 含义: 应用这个操作符会生成一个 Completable, 故只有错误和完成, 所有其他元素均被忽略
 */
exampleOf(msg: "过滤操作符: ignoreElements", action: {
    let strikes = PublishSubject<String>()
    strikes.ignoreElements().subscribe(onCompleted: {
        print("完成")
    }, onError: { error in
        print(error)
    }).disposed(by: $0)
    strikes.onNext("X")
    strikes.onNext("X")
    strikes.onNext("X")
    strikes.onNext("X")
    strikes.onCompleted()
})

/*
 含义: 在一个可观察序列中获取指定下标的元素
 */
exampleOf(msg: "过滤操作符: elementAt", action: {
    let strikes = PublishSubject<String>()
    strikes.elementAt(2).subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("释放")
    }).disposed(by: $0)

    strikes.onNext("X") // 0
    strikes.onNext("Y") // 1
    strikes.onNext("Z") // 2
    strikes.onNext("1") // 3
    strikes.onCompleted()
})

/*
 含义: 在序列中跳过若干个元素后开始取元素直到完成或错误.
 */
exampleOf(msg: "过滤操作符: skip", action: {
    let flow = PublishSubject<String>()
    flow.skip(3).subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("被释放")
    }).disposed(by: $0)
    flow.onNext("X")
    flow.onNext("Y")
    flow.onNext("Z")
    flow.onNext("1")
    flow.onNext("2")
    flow.onCompleted()
})

/*
 一直跳过元素, 直到遇到满足条件的元素就不再执行跳过操作
 */
exampleOf(msg: "过滤操作符: skipWhile", action: { bag in
    let flow = PublishSubject<String>()
    flow.skipWhile({ $0.count > 2 }).subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("被释放")
    }).disposed(by: bag)

    flow.onNext("1234")
    flow.onNext("256")
    flow.onNext("377")
    flow.onNext("4")
    flow.onNext("5")
    flow.onNext("6")
    flow.onNext("7")
    flow.onCompleted()
})

/*
 在另外一个序列没有开始发射时就一直忽略元素, 直到另外一个序列发射后停止忽略.
 */
exampleOf(msg: "过滤操作符: skipUntil", action: { bag in
    let flow = PublishSubject<String>()
    let trigger = PublishSubject<Void>()

    flow.skipUntil(trigger).subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("被释放")
    }).disposed(by: bag)

    flow.onNext("X")
    flow.onNext("X")
    flow.onNext("X")
    trigger.onNext(())
    flow.onNext("Y")
    flow.onNext("Z")
    flow.onCompleted()
})

/*
 取指定个数的元素
 */
exampleOf(msg: "过滤操作符: take", action: { bag in
    Observable.from([1, 2, 3, 4, 5, 6]).take(2).subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("被释放")
    }).disposed(by: bag)
})

/*
 一直取元素, 直到遇到不满足指定条件的元素就终止取元素.
 需要注意的是, 如果序列的第一个元素就不满足条件, 则之后所有的元素都不会取.
 */
exampleOf(msg: "过滤操作符: takeWhile", action: { bag in
    Observable.of(1, 2, 3, 4, 5, 6, 7)
        .takeWhile({ $0 > 3 })
        .subscribe(onNext: {
            print($0)
        }, onError: {
            print($0)
        }, onCompleted: {
            print("完成")
        }, onDisposed: {
            print("被释放")
        }).disposed(by: bag)

    Observable.of(1, 2, 3, 4, 5, 6, 7)
        .enumerated()
        .takeWhile({ index, elem in index < 3 && elem < 2 })
        .subscribe(onNext: {
            print($0)
        }, onError: {
            print($0)
        }, onCompleted: {
            print("完成")
        }, onDisposed: {
            print("被释放")
        }).disposed(by: bag)
})

/*
 在另外一个序列没有发射时一直取元素, 直到另外一个序列发射就停止取
 */
exampleOf(msg: "过滤操作符 takeUntil", action: { bag in
    let flow = PublishSubject<String>()
    let trigger = PublishSubject<Void>()
    flow.takeUntil(trigger).subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("被释放")
    }).disposed(by: bag)
    flow.onNext("X")
    flow.onNext("X")
    flow.onNext("X")
    flow.onNext("X")
    flow.onNext("X")
    trigger.onNext(())
    flow.onNext("Y")
    flow.onNext("Z")
    flow.onNext("1")
    flow.onCompleted()
})

/*
 在连续的元素中, 如果有若干个相邻元素相等, 则只取相邻元素中的第一个, 直到遇到不同的再取.
 */
exampleOf(msg: "过滤操作符: distinctUntilChanged", action: { bag in
    let flow = PublishSubject<String>()
    flow.distinctUntilChanged().subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("被释放")
    }).disposed(by: bag)

    flow.onNext("X")
    flow.onNext("X")
    flow.onNext("Y")
    flow.onNext("X")
    flow.onNext("Z")
    flow.onNext("Z")
    flow.onNext("X")
    flow.onCompleted()
})