Rx 操作符二

• concatMap

将 Observable 的元素转换成其他的 Observable，然后将这些 Observables 串连起来 concatMap 操作符将源 Observable 的每一个元素应用一个转换方法，将他们转换成 Observables。然后让这些 Observables 按顺序的发出元素，当前一个 Observable 元素发送完毕后，后一个 Observable 才可以开始发出元素。等待前一个 Observable 产生完成事件后，才对后一个 Observable 进行订阅。

let disposeBag = DisposeBag()

let subject1 = BehaviorSubject(value: "??")
let subject2 = BehaviorSubject(value: "??")

let variable = Variable(subject1)

variable.asObservable()
        .concatMap { $0 }
        .subscribe { print($0) }
        .disposed(by: disposeBag)

subject1.onNext("??")
subject1.onNext("??")

variable.value = subject2

subject2.onNext("I would be ignored")
subject2.onNext("??")

subject1.onCompleted()

subject2.onNext("??")

• connect

通知 ConnectableObservable 可以开始发出元素了

ConnectableObservable 和普通的 Observable 十分相似，不过在被订阅后不会发出元素，直到 connect 操作符被应用为止。这样一来可以等所有观察者全部订阅完成后，才发出元素。

let intSequence = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
    .publish()

_ = intSequence
    .subscribe(onNext: { print("Subscription 1:, Event: \($0)") })

DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
    _ = intSequence.connect()
}

DispatchQueue.main.asyncAfter(deadline: .now() + 4) {
  _ = intSequence
      .subscribe(onNext: { print("Subscription 2:, Event: \($0)") })
}

DispatchQueue.main.asyncAfter(deadline: .now() + 6) {
  _ = intSequence
      .subscribe(onNext: { print("Subscription 3:, Event: \($0)") })
}

• create

通过一个构建函数完整的创建一个 Observable

create 操作符将创建一个 Observable，你需要提供一个构建函数，在构建函数里面描述事件（next，error，completed）的产生过程。

通常情况下一个有限的序列，只会调用一次观察者的 onCompleted 或者 onError 方法。并且在调用它们后，不会再去调用观察者的其他方法。

创建一个 [0, 1, ... 8, 9] 的序列：

let id = Observable<Int>.create { observer in
    observer.onNext(0)
    observer.onNext(1)
    observer.onNext(2)
    observer.onNext(3)
    observer.onNext(4)
    observer.onNext(5)
    observer.onNext(6)
    observer.onNext(7)
    observer.onNext(8)
    observer.onNext(9)
    observer.onCompleted()
    return Disposables.create()
}

• debounce 过滤掉高频产生的元素

debounce 操作符将发出这种元素，在 Observable 产生这种元素后，一段时间内没有新元素产生。

• debug 打印所有的订阅，事件以及销毁信息

let disposeBag = DisposeBag()

let sequence = Observable<String>.create { observer in
    observer.onNext("??")
    observer.onNext("??")
    observer.onCompleted()
    return Disposables.create()
}

sequence
    .debug("Fruit")
    .subscribe()
    .disposed(by: disposeBag)