rxswift学习-subject

2017-06-08

什么是subject?

subject既是observable也是observer。

他是observable就能够让我们随时调用subscribe方法，让其他Observer接收element，这个接收可以是从中间的，并不像Observable创建操作符那样，每次都从头发送全部elements。

他是observer就能够让我们调用他的on方法来随时动态添加emit的element。

这些特性也告诉了我们什么时候用observable和subject。

PublishSubject

可以先猜想一下是怎么实现的：他是Observable，那么就需要提供subscribe函数，他也是Observer，那么就需要提供on函数。他能subscribe多个observer，那么就肯定会有一个observer数组之类的保存当前所有的observer。当on时，会调用一个循环之类的让所有observer都接收这个element。查看源码后，简单来说也就是这么实现的，不过他考虑的更多，或处理一些同步和dispose的问题。Ok，还是看看怎么用吧

let subject = PublishSubject<String>()
 
 subject.onNext("hello")
 
 subject.subscribe { (event) in
      print("subscriptionOne event:\(event)")
 }.disposed(by: bag)
        
subject.onNext("1")
subject.onNext("2")
//
subject.subscribe { (event) in
    print("subscriptiontwo event:\(event)")
}.disposed(by: bag)
subject.onNext("3")
subject.onNext("4")
subject.onCompleted()
 subject.subscribe { (event) in
            print("subscriptionthree event:\(event)")
 }.disposed(by: bag)
//---
subscriptionOne event:next(1)
subscriptionOne event:next(2)
subscriptionOne event:next(3)
subscriptiontwo event:next(3)
subscriptionOne event:next(4)
subscriptiontwo event:next(4)
subscriptionOne event:completed
subscriptiontwo event:completed
subscriptionthree event:completed

总的来说就是：你subscribe他之后，只能接收到subject之后on的element。如果已经error或者completed，那么会直接收到.error或.completed。

BehaviorSubjects

和PublishSubject类似，只是会重放最后一个.next。查看源文件代码和PublishSubjects基本一样，多了个_element变量用于保存最后一个.next的值，在subscribe时，会对observer调用on方法，让他收到最后一个element。其他代码基本是一样的。

1	observer.on(.next(_element))

let subject = BehaviorSubject<String>.init(value: "初始值")
subject.subscribe { (event) in
    print("subscriptionOne event:\(event)")
}.disposed(by: bag)
subject.onNext("1")
subject.onNext("2")
subject.subscribe { (event) in
    print("subscriptiontwo event:\(event)")
}.disposed(by: bag)
subject.onNext("3")
subject.onNext("4")
subject.onCompleted()
subject.subscribe { (event) in
    print("subscriptionthree event:\(event)")
}.disposed(by: bag)
//-----
subscriptionOne event:next(初始值)
subscriptionOne event:next(1)
subscriptionOne event:next(2)
subscriptiontwo event:next(2)
subscriptionOne event:next(3)
subscriptiontwo event:next(3)
subscriptionOne event:next(4)
subscriptiontwo event:next(4)
subscriptionOne event:completed
subscriptiontwo event:completed
//可以注意到如果sequence已经退出了，此时并没有重放.next。
subscriptionthree event:completed

ReplaySubjects

本以为和上面的类似，只是存储的不是一个element，而是多个。可实际情况却不是这样，ReplaySubject只是一个抽象类。先看看继承关系图：

let subject = ReplaySubject<String>.create(bufferSize: 2)
subject.subscribe { (event) in
    print("subscriptionOne event:\(event)")
}.disposed(by: bag)
subject.onNext("1")
subject.onNext("2")
subject.subscribe { (event) in
    print("subscriptiontwo event:\(event)")
}.disposed(by: bag)
subject.onNext("3")
subject.onNext("4")
subject.onCompleted()
subject.subscribe { (event) in
    print("subscriptionthree event:\(event)")
}.disposed(by: bag)
//----
subscriptionOne event:next(1)
subscriptionOne event:next(2)
subscriptiontwo event:next(1)
subscriptiontwo event:next(2)
subscriptionOne event:next(3)
subscriptiontwo event:next(3)
subscriptionOne event:next(4)
subscriptiontwo event:next(4)
subscriptionOne event:completed
subscriptiontwo event:completed
///和behaviorsubject不同，即使退出了，也能重放之前的.next。
subscriptionthree event:next(3)
subscriptionthree event:next(4)
subscriptionthree event:completed

Variable

是对BehaviorSubject的封装。与BehaviorSubject不同的是他永远不会收到.error而结束sequence。当他deallocated时，会completed sequence。

let subject = Variable<Int>.init(100)
subject.asObservable().subscribe { (event) in
    print("1 ) \(event)")
}.disposed(by: bag)
subject.value = 200
subject.value = 300
subject.asObservable().subscribe { (event) in
    print("2) \(event)")
}.disposed(by: bag)
subject.value = 300
subject.value = 400
//------
1 ) next(100)
1 ) next(200)
1 ) next(300)
2) next(300)
1 ) next(300)
2) next(300)
1 ) next(400)
2) next(400)
1 ) completed
2) completed

AsyncSubject

这个subject只会发送最后一个element。而且只会在subject completed后才发送。

let subject = AsyncSubject<Int>.init()
        
subject.subscribe { (event) in
    print(event)
}.disposed(by: bag)
subject.onNext(100)
subject.onNext(200)
subject.onNext(300)
subject.onCompleted()//如果没有completed，那么不会emit
///结果
next(300)
completed