rxswift学习-Observable操作符

2017-06-08

按照首字母一个个来吧，总共也就60个左右。

amb

将多个observable 传入amb，选择其中最先emit element的那个observable sequence。

//静态方法
static func amb<S: Sequence>(_ sequence: S) -> Observable<Element>
        where S.Iterator.Element == Observable<Element> 
//实例方法
func amb<O2: ObservableType>
        (_ right: O2)
        -> Observable<E> where O2.E == E 
//-----------------
let first = Observable<Int>
            .of(1,2,3)
            .delaySubscription(2, scheduler: ConcurrentDispatchQueueScheduler.init(qos: .default))
let second = Observable<Int>
            .of(4,5,6)
            .delaySubscription(3, scheduler: ConcurrentDispatchQueueScheduler.init(qos: .default))
let third = Observable<Int>
            .of(7,8,9)
            .delaySubscription(1, scheduler: ConcurrentDispatchQueueScheduler.init(qos: .default))
let amb = Observable<Int>.amb([first,second,third])
amb.subscribe { (event) in
    print(event)
}.disposed(by: bag)
//------ third 最先 --
next(7)
next(8)
next(9)
completed

asMaybe

//Maybe是一个包含0个或者1个element的sequence。如果observable超过了1个element，那么会收到.error (RxError.moreThanOneElement)
public func asMaybe() -> Maybe<E>
///-----------
let _ = Observable<Int>
            .of(1,2,3)//超过了1个
            .asMaybe()
            .asObservable()
            .subscribe { (event) in
                print(event)
        }.disposed(by: bag)
//------结果
error(Sequence contains more than one element.)

asSingle

//只包含1个element的observable sequence，如果为0个那么.error(RxError.noElements),如果超过1个那么.error(RxError.moreThanOneElement)
public func asSingle() -> Single<E>
/** 
single和Maybe都是PrimitiveSequence，PrimitiveSequence是对Observable的封装，本身就是一个结构体包含一个observable。
*/

buffer

收集在规定时间内或者达到buffer最大值的所有element，将他们作为一个整体发送。

 func buffer(
   timeSpan: RxTimeInterval, //时间段
   count: Int, //buffer的最大数量
   scheduler: SchedulerType
 )-> Observable<[E]> 
 
 //---------
Observable<Int>
.interval(1, scheduler: ConcurrentDispatchQueueScheduler.init(qos: .default))
.take(5)
.buffer(timeSpan: 2, count: 4, scheduler: MainScheduler.instance)
.subscribe { (event) in
                print(event)
}.disposed(by: bag)
//---------
next([0])
next([1, 2])
next([3, 4])
completed

catchError


func catchError(_ handler: @escaping (Swift.Error) throws -> Observable<E>)
//handler为{ _ in Observable.just(element) }特殊情况
func catchErrorJustReturn(_ element: E)
        -> Observable<E>
        
//--------example---
Observable.of(1,2,3)
.asMaybe()//最多只能有1个element，所以会报error
.asObservable()
.catchError { (error) -> Observable<Int> in//捕获error，将其替换为Observable
            return Observable<Int>.of(3,4,5)
}.subscribe { (event) in
            print(event)
}.disposed(by: bag)

//遍历sequence,如果遇到.error，那么接着下一个Observable emit element,如果observable completed了，那么结束遍历。
static func catchError<S: Sequence>(_ sequence: S) -> Observable<Element> where S.Iterator.Element == Observable<Element> 
//---------- example ----
let first = Observable<Int>.create({ (observer) -> Disposable in
  observer.onNext(1)
  observer.onNext(2)
  observer.onNext(3)
  observer.onError(RxError.unknown)
  return Disposables.create()
})
  let second = Observable<Int>.create({ (observer) -> Disposable in
      observer.onNext(4)
      observer.onNext(5)
      observer.onNext(6)
      observer.onCompleted()
      return Disposables.create()
  })
  let third = Observable<Int>.create({ (observer) -> Disposable in
      observer.onNext(7)
      observer.onNext(8)
      observer.onNext(9)
      observer.onError(RxError.unknown)
      return Disposables.create()
  })
        
      Observable<Int>.catchError([first,second,third]).subscribe { (event) in
        print(event)
      }.disposed(by: bag)
        
//------ 结果 --
next(1)
next(2)
next(3)
next(4)
next(5)
next(6)
completed

retry

如果遇到.error，那么重复observable sequence，知道completed。这有可能产生无限序列。

func retry() -> Observable<E> 
func retry(_ maxAttemptCount: Int)
//-----example ---
var flag = 1
let first = Observable<Int>.create({ (observer) -> Disposable in
  observer.onNext(1)
  observer.onNext(2)
  observer.onNext(3)
  if(flag == 3){
      observer.onCompleted()
  }else{
      flag += 1
      observer.onError(RxError.unknown)
  }
  return Disposables.create()
})
first.retry().subscribe { (event) in
  print(event)
}.disposed(by: bag)
//----- result ----
next(1)
next(2)
next(3)
next(1)
next(2)
next(3)
next(1)
next(2)
next(3)
completed

combineLatest

用指定的函数将多个observable sequence的elment合并成1个element，并emit 他。

//使用指定的函数合并
static func combineLatest<O1: ObservableType, O2: ObservableType>
        (_ source1: O1, _ source2: O2, resultSelector: @escaping (O1.E, O2.E) throws -> E)
//直接合并为一个元组返回
static func combineLatest<O1: ObservableType, O2: ObservableType>
        (_ source1: O1, _ source2: O2)
            -> Observable<(O1.E, O2.E)>
//传入一个Observable集合          
static func combineLatest<C: Collection>(_ collection: C, _ resultSelector: @escaping ([C.Iterator.Element.E]) throws -> Element) -> Observable<Element>
        where C.Iterator.Element: ObservableType            
//不提供resultselector，将返回最近的element数组
static func combineLatest<C: Collection>(_ collection: C) -> Observable<[Element]>
        where C.Iterator.Element: ObservableType, C.Iterator.Element.E == Element
//最多可以合并8个
//-----example ---
let first = PublishSubject<Int>.init()
let second = PublishSubject<Int>.init()
Observable<Int>.combineLatest(first, second) { (o1, o2) -> Int in
  return o1 + o2
}.subscribe { (event) in
  print(event)
}.disposed(by: bag)
first.onNext(1)
second.onNext(5)
first.onNext(2)
first.onNext(3)
second.onNext(6)
first.onNext(4)
second.onNext(7)
first.onCompleted()
second.onNext(8)
second.onCompleted()
//-------
next(6)
next(7)
next(8)
next(9)
next(10)
next(11)
next(12)
completed

concat

将多个observable串联起来, 最好不要用于hot observable。如果遇到.error，那么不会继续concat了。

func concat<O: ObservableConvertibleType>(_ second: O) -> Observable<E> where O.E == E
static func concat<S: Sequence >(_ sequence: S) -> Observable<Element>
        where S.Iterator.Element == Observable<Element>
        
static func concat(_ sources: Observable<Element> ...) -> Observable<Element>
static func concat<S: Collection >(_ collection: S) -> Observable<Element>
//------example ----
let first = Observable<Int>.create { (observer) -> Disposable in
    observer.onNext(1)
    observer.onNext(2)
    observer.onNext(3)
    observer.onCompleted()
    return Disposables.create()
}
let second = Observable<Int>.create { (observer) -> Disposable in
    observer.onNext(4)
    observer.onNext(5)
    observer.onNext(6)
    observer.onCompleted()
    return Disposables.create()
}
let concat = Observable<Int>.concat(first,second)
concat.subscribe { (event) in
    print(event)
}.disposed(by: bag)
//------result ----
next(1)
next(2)
next(3)
next(4)
next(5)
next(6)
completed

debounce

1 2	func debounce(_ dueTime: RxTimeInterval, scheduler: SchedulerType) -> Observable<E>

ifEmpty

let observable = Observable<Int>.empty().ifEmpty(default: 100)
observable.subscribe { (event) in
    print(event)
}.disposed(by: bag)
//---result--
next(100)
completed

defer

每次有人subscribe时，通过factory创建一个Observable sequence，并emit这个sequence的element。

delay

1
2
3

//如果是.error不会delay
func delay(_ dueTime: RxTimeInterval, scheduler: SchedulerType)
        -> Observable<E>

delaySubscription

1
2
3

//delay Subscription， 也就是说如果第一个就是.error也会delay.
func delaySubscription(_ dueTime: RxTimeInterval, scheduler: SchedulerType)
        -> Observable<E>

materialize

Observable<Int>.of(1,2,3)
.materialize()
.subscribe { (event) in
    print(event)
}.disposed(by: bag)
//---- result
next(next(1))
next(next(2))
next(next(3))
next(completed)
completed

distinctUntilChanged

1	//去掉连续的重复的element

do

///为每个event注册一个action，当event到达时，调用对应的event。
//和直接调用subscribe方法相比，多了onSubscribe和onSubscribed
func `do`(onNext: ((E) throws -> Void)? = nil, onError: ((Swift.Error) throws -> Void)? = nil, onCompleted: (() throws -> Void)? = nil, onSubscribe: (() -> ())? = nil, onSubscribed: (() -> ())? = nil, onDispose: (() -> ())? = nil)
        -> Observable<E>

elementAt

1	//只发送第n个element。

ignoreElements

1	//忽略所有的"element"，.error和.completed不会忽略

filter

1
2
3

//过滤element，只返回predicate为true的。
func filter(_ predicate: @escaping (E) throws -> Bool)
        -> Observable<E>

groupBy

将一个observable sequence分割成多个sequnce，并将每个sequnce作为一个element emit出去。
func groupBy<K: Hashable>(keySelector: @escaping (E) throws -> K)
        -> Observable<GroupedObservable<K,E>>
//用keySelector来为每个element生成一个唯一标识，将具有相同唯一标识的element分为一组。最后将每组作为一个element emit出去。        
//------example----
let ob = Observable<Int>.of(1,2,3,4,5,6,7,8,9)
let groupe = ob.groupBy { (value) -> Bool in
    return value % 2 == 0 //按照奇偶分组
}
groupe.subscribe { (event) in
    if let observable = event.element {
        let key = observable.key
        observable.subscribe({ (event) in
            print("key \(key) value:\(event)")
        }).disposed(by: self.bag)
    }
}.disposed(by: bag)
//------result---
key false value:next(1)
key true value:next(2)
key false value:next(3)
key true value:next(4)
key false value:next(5)
key true value:next(6)
key false value:next(7)
key true value:next(8)
key false value:next(9)
key false value:completed
key true value:completed

map

对每个element执行map函数，得到一个新的element。

Merge相关

merge

1
2
3

static func merge(_ sources: Observable<E>...) -> Observable<E>
static func merge(_ sources: [Observable<E>]) -> Observable<E>
static func merge<C: Collection>(_ sources: C) -> Observable<E> where C.Iterator.Element == Observable<E>

flatMap

1
2
3

//通过selector将一个element转换为一个observable sequence。然后所有得到的observables执行merge操作，作为最终的结果。
func flatMap<O: ObservableConvertibleType>(_ selector: @escaping (E) throws -> O)
        -> Observable<O.E>

flatMapWithIndex

1
2
3

//和上面的区别在于，传入的selector带有的index参数。
func flatMapWithIndex<O: ObservableConvertibleType>(_ selector: @escaping (E, Int) throws -> O)
        -> Observable<O.E>

flatMapFirst

1	并发的时候，只取当前正在处理的observable，忽略过程中其他observable。

multicast

将source observable sequence传递给某个subject，subject会将element传递给所有subscribe了的observer。初步的感觉就是：将某个observable sequence作为subject的数据源。

//注意返回的是一个ConnectableObservable, 也就是说只有当你调用了connect方法时，才会将source observable sequence的element通过subject转发给所有的observer。
func multicast<S: SubjectType>(_ subject: S)
        -> ConnectableObservable<S.E> where S.SubjectObserverType.E == E
//--------- example ----
let subject = PublishSubject<Int>.init()
let publish = Observable<Int>
.of(1,2,3,4,5)
.multicast(subject)
publish.subscribe { (event) in
    print("first observer \(event)")
}.disposed(by: bag)
publish.subscribe { (event) in
    print("second observer \(event)")
}.disposed(by: bag)
publish.connect().disposed(by: bag)
//----- result ---
first observer next(1)
second observer next(1)
first observer next(2)
second observer next(2)
first observer next(3)
second observer next(3)
first observer next(4)
second observer next(4)
first observer next(5)
second observer next(5)
first observer completed
second observer completed

publish

当subject为publishsubject时的特殊multicast。

1
2
3

func publish() -> ConnectableObservable<E> {
        return self.multicast(PublishSubject())
}

replay

当subject为ReplaySubject时的特殊multicast

func replay(_ bufferSize: Int)
      -> ConnectableObservable<E> {
      return self.multicast(ReplaySubject.create(bufferSize: bufferSize))
}

replayAll

当subject为ReplaySubject时的特殊multicast。ReplaySubject的buff为all。

func replayAll()
        -> ConnectableObservable<E> {
        return self.multicast(ReplaySubject.createUnbounded())
}

refCount

让一个connectable observable表现的像一个普通的observable。我们知道一个connectable observable的主要特点就是只有在调用了connect方法后才会emit element 。而refcount操作符，自动化了connect和disconnect。当第一个observer subscribe了，他会自动connect 对应的connectable observable。他会记录有多少observer subscribe了他，当最后一个observer dispose之后会disconnect。

//注意修饰的是ConnectableObservableType
func refCount() -> Observable<E>
//----- example ---
let subject = PublishSubject<Int>.init()
let publish =
    Observable<Int>
    .interval(1, scheduler: ConcurrentDispatchQueueScheduler.init(qos: .default))
    .take(5)
    .multicast(subject)
    .refCount()
publish.subscribe { (event) in
    print("first observer \(event)")
}.disposed(by: bag)
publish.subscribe { (event) in
    print("second observer \(event)")
}.disposed(by: bag)
///不用调用connect也能够收到element 
//----- result ---
first observer next(0)
second observer next(0)
first observer next(1)
second observer next(1)
first observer next(2)
second observer next(2)
first observer next(3)
second observer next(3)
first observer next(4)
second observer next(4)
first observer completed
second observer completed

never

返回一个不会退出的observable sequnce。

1	static func never() -> Observable<E>

observeOn

1
2
3

//在学习scheduler时已经说过了他的作用：封装source sequnce，让他的observer callback在指定的scheduler上执行。
func observeOn(_ scheduler: ImmediateSchedulerType)
        -> Observable<E>

optional

将某个option值转化为一个observable sequnce。

static func from(optional: E?) -> Observable<E>
static func from(optional: E?, 
scheduler: ImmediateSchedulerType //将element发送到那个scheduler上。
) -> Observable<E>

range

通过起始值和elment数目生成某个范围内的值，并将其封装为一个observable sequence。

1	static func range(start: E, count: E, scheduler: ImmediateSchedulerType = CurrentThreadScheduler.instance) -> Observable<E>

reduce

提供一个初始值和函数，对所有元素操作后的结果作为一个 element。

func reduce<A, R>(
	_ seed: A, 
	accumulator: @escaping (A, E) throws -> A,
	mapResult: @escaping (A) throws -> R
) -> Observable<R> 
//--- example --
//计算1到10的和 将结果作为一个observable<string>返回
Observable<Int>
.range(start: 1, count: 10)
.reduce(0, accumulator: +) { (finalValue) ->String in
    return "\(finalValue)"
}.subscribe { (event) in
	print(event)
}.disposed(by: bag)
//----- result --
next(55)
completed

repeat

无限重复生成某个element

static func repeatElement(
  _ element: E, 
  scheduler: ImmediateSchedulerType = CurrentThreadScheduler.instance
) -> Observable<E>

sample

将一个observable sequence作为一个采样序列，每当采样序列来了一个element，就会将source sequence的最近一个element添加到result sequence中，如果最近elemen是同一个，那么不会重复添加。

scan

将accumulator得到的结果作为sequence。
func scan<A>(
 _ seed: A, 
accumulator: @escaping (A, E) throws -> A
)-> Observable<A>

of

1	static func of(_ elements: E ..., scheduler: ImmediateSchedulerType = CurrentThreadScheduler.instance) -> Observable<E>

from

1
2
3

static func from(_ array: [E], scheduler: ImmediateSchedulerType = CurrentThreadScheduler.instance) -> Observable<E>
static func from<S: Sequence>(_ sequence: S, scheduler: ImmediateSchedulerType = CurrentThreadScheduler.instance) -> Observable<E> where S.Iterator.Element == E

skip

//忽略前n个element
func skip(_ count: Int)  -> Observable<E>
//-----
 Observable<Int>
.range(start: 1, count: 10)
.skip(3)
.subscribe { (event) in
    print(event)
}.disposed(by: bag)
//----
next(4)
next(5)
next(6)
next(7)
next(8)
next(9)
next(10)
completed

skipUntil

一直跳过，直到某个observable 开始emit element了。

skipWhile

startWith

subscribeOn

flatMapLatest

同flatMap，只是只取最近的elemnt。

switchLatest

take相关

//只取前n个
func take(_ count: Int)
//只取前duration内的element
func take(_ duration: RxTimeInterval, scheduler: SchedulerType) -> Observable<E>
//只取最后n个
func takeLast(_ count: Int)-> Observable<E>
//一直take，直到other emit了element。
func takeUntil<O: ObservableType>(_ other: O)-> Observable<E>
//当满足条件就一直take
func takeWhile(_ predicate: @escaping (E) throws -> Bool) -> Observable<E>

throttle

1 2	func throttle(_ dueTime: RxTimeInterval, latest: Bool = true, scheduler: SchedulerType) -> Observable<E>

timeout

规定时间内没有收到element会产生一个error。

interval

每隔一段时间产生一个element

1 2	static func interval(_ period: RxTimeInterval, scheduler: SchedulerType) -> Observable<E>

timer

toArray

将所有element弄成一个数组，作为一个element 。

using

创建一个disposable的resource，只在observable sequence的生命周期内存在，当sequnce dispose时，resouce也dispose。

1	static func using<Resource: Disposable>(_ resourceFactory: @escaping () throws -> Resource, observableFactory: @escaping (Resource) throws -> Observable<E>) -> Observable<E>

window

和buffer类似，不过不是讲多个element弄成一个数组 emit，而是将多个element弄成一个observable sequnce再 emit。

1 2	func window(timeSpan: RxTimeInterval, count: Int, scheduler: SchedulerType) -> Observable<Observable<E>>

withLatestFrom

同combineLatest。只是调用的方式不一样。

1	func withLatestFrom<SecondO: ObservableConvertibleType, ResultType>(_ second: SecondO, resultSelector: @escaping (E, SecondO.E) throws -> ResultType) -> Observable<ResultType>

zip

和comebineLatest类似，不过不是latest，而是严格的1对1.

static func zip<O1: ObservableType, O2: ObservableType>
        (_ source1: O1, _ source2: O2, resultSelector: @escaping (O1.E, O2.E) throws -> E)
        -> Observable<E>
        
static func zip<C: Collection>(_ collection: C, _ resultSelector: @escaping ([C.Iterator.Element.E]) throws -> Element) -> Observable<Element>
        where C.Iterator.Element: ObservableType
        
static func zip<C: Collection>(_ collection: C) -> Observable<[Element]>
        where C.Iterator.Element: ObservableType, C.Iterator.Element.E == Element