通过RACMulticastConnection源码分析理解Re
RACMulticastConnection涉及到热信号，RACDisposable的一些概念。这里对源码进行分析做一个透彻的理解。
美团技术博客关于冷热信号有3篇文章:美团技术博客ReactiveCocoa冷热信号，给出的冷热信号的区别：
1. Hot Observable是主动的，尽管你并没有订阅事件，但是它会时刻推送，就像鼠标移动；而Cold Observable是被动的，只有当你订阅的时候，它才会发布消息。
2. Hot Observable可以有多个订阅者，是一对多，集合可以与订阅者共享信息；而Cold Observable只能一对一，当有不同的订阅者，消息是重新完整发送。
这里从源码层级深入了解一下。下面是一个使用热信号的示例代码一：
// 示例代码一
// <1>创建RACMulticastConnection
RACMulticastConnection *connection = [[RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
[[RACScheduler mainThreadScheduler] afterDelay:1 schedule:^{
[subscriber sendNext:@1];
}];
[[RACScheduler mainThreadScheduler] afterDelay:2 schedule:^{
[subscriber sendNext:@2];
}];
[[RACScheduler mainThreadScheduler] afterDelay:3 schedule:^{
[subscriber sendNext:@3];
}];
[[RACScheduler mainThreadScheduler] afterDelay:4 schedule:^{
[subscriber sendCompleted];
}];
return nil;
}] publish];
// <2> RACMulticastConnection进行connect
[connection connect];
// <3> 对connection.signal进行订阅
RACSignal *signal = connection.signal;
NSLog(@"Signal was created.");
[[RACScheduler mainThreadScheduler] afterDelay:1.1 schedule:^{
[signal subscribeNext:^(id x) {
NSLog(@"Subscriber 1 recveive: %@", x);
}];
}];
示例代码一将冷信号通过RACMulticastConnection转化成了热信号。
步骤<1> 创建RACMulticastConnection
示例代码一的<1>是对一个冷信号发送publish消息创建了一个RACMulticastConnection。RACMulticastConnection的sourceSignal属性保存了要转化的冷信号，signal属性保存了一个新创建的RACSubject，其实RACSubject就是一个热信号。
步骤<2> RACMulticastConnection进行connect
示例代码一的<2>是使用RACMulticastConnection的热信号signal对冷信号sourceSignal进行订阅，
- (RACDisposable *)connect {
BOOL shouldConnect = OSAtomicCompareAndSwap32Barrier(0, 1, &_hasConnected);
if (shouldConnect) {
self.serialDisposable.disposable = [self.sourceSignal subscribe:_signal];
}
return self.serialDisposable;
}
上面代码中的[self.sourceSignal subscribe:_signal]里的subscribe调用的是RACDynamicSignal的subscribe方法，看一下这个方法的源码:
RACDynamic.m
- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
NSCParameterAssert(subscriber != nil);
RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
// <a>
subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];
OSSpinLockLock(&_subscribersLock);
if (_subscribers == nil) {
_subscribers = [NSMutableArray arrayWithObject:subscriber];
} else {
[_subscribers addObject:subscriber];
}
OSSpinLockUnlock(&_subscribersLock);
@weakify(self);
RACDisposable *defaultDisposable = [RACDisposable disposableWithBlock:^{
@strongify(self);
if (self == nil) return;
BOOL stillHasSubscribers = YES;
OSSpinLockLock(&_subscribersLock);
{
// Since newer subscribers are generally shorter-lived, search
// starting from the end of the list.
NSUInteger index = [_subscribers indexOfObjectWithOptions:NSEnumerationReverse passingTest:^ BOOL (id<RACSubscriber> obj, NSUInteger index, BOOL *stop) {
return obj == subscriber;
}];
if (index != NSNotFound) {
[_subscribers removeObjectAtIndex:index];
stillHasSubscribers = _subscribers.count > 0;
}
}
OSSpinLockUnlock(&_subscribersLock);
if (!stillHasSubscribers) {
[self invalidateGlobalRefIfNoNewSubscribersShowUp];
}
}];
[disposable addDisposable:defaultDisposable];
// <b>
if (self.didSubscribe != NULL) {
RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
RACDisposable *innerDisposable = self.didSubscribe(subscriber);
if (innerDisposable != nil) [disposable addDisposable:innerDisposable];
}];
if (schedulingDisposable != nil) [disposable addDisposable:schedulingDisposable];
}
return disposable;
}
上述代码<a>部分依据参数subscriber，创建了一个新的subscriber；参数subscriber代入的是RACMulticastConnection的RACSubject类型的signal属性，返回的subscriber是RACPassthroughSubscriber。初始化源码如下所示:
RACPassthroughSubscriber.m
- (instancetype)initWithSubscriber:(id<RACSubscriber>)subscriber signal:(RACSignal *)signal disposable:(RACCompoundDisposable *)disposable {
NSCParameterAssert(subscriber != nil);
self = [super init];
if (self == nil) return nil;
_innerSubscriber = subscriber;
_signal = signal;
_disposable = disposable;
[self.innerSubscriber didSubscribeWithDisposable:self.disposable];
return self;
}
这个RACPassthroughSubscriber包裹了RACSubject这个subscriber。
上述代码<b>部分意思是如果RACMulticastConnection的冷信号sourceSignal的didSubscribe存在，就去执行didSubscribe(subscriber)。在示例代码一中，冷信号的didSubscribe存在，在didSubscribe执行过程中，比如说“[subscriber sendNext:@1];”这一步，这里的sendNext调用的是RACPassthroughSubscriber的sendNext：
RACPassthroughSubscriber.m
- (void)sendNext:(id)value {
if (self.disposable.disposed) return;
if (RACSIGNAL_NEXT_ENABLED()) {
RACSIGNAL_NEXT(cleanedSignalDescription(self.signal), cleanedDTraceString(self.innerSubscriber.description), cleanedDTraceString([value description]));
}
[self.innerSubscriber sendNext:value];
}
联系<a>部分的分析，"[self.innerSubscriber sendNext:value]"实际上就是对RACMulticastConnection的RACSubject类型的热信号signal发送sendNext消息，
RACSubject.m
- (void)sendNext:(id)value {
[self enumerateSubscribersUsingBlock:^(id<RACSubscriber> subscriber) {
[subscriber sendNext:value];
}];
}
在示例代码一的流程<2>中，RACMulticastConnection的热信号signal并没有订阅者，所以上述sendNext方法没有任何效果。
步骤<3> 对RACMulticastConnection的热信号signal进行订阅
在步骤<2>中，RACMulticastConnection的热信号signal就开始对冷信号sourceSignal进行订阅了，冷信号sourceSignal的didSubscribe就开始执行了，但此时热信号signal没有任何订阅者，所以冷信号的sendNext没有接受者。
步骤<3>对热信号signal进行了订阅，于是此时热信号signal有了订阅者，当didSubscribe中再次sendNext时，上述RACSubject的sendNext方法就会向订阅者发送值了。
总结
根据上述分析，重新对冷热信号进行总结：
1. Hot Observable是主动的，在不断Push信息，不管有没有人订阅；而Cold Observable是被动的，需要去订阅来Pull信息。
2. Hot Observable由于RACSubject的sendNext可以对多个订阅者发内容，所以是一对多，订阅者共享信息；而Cold Observable只能一对一，当有不同的订阅者，消息是重新完整发送。
上述分析涉及到的源码，这里也进行总结：
1. RACMulticastConnection的sourceSignal属性是冷信号，signal属性是热信号，connect方法就是使用热信号对冷信号进行订阅。
2. RACPassthroughSubscriber是对订阅者进行了一层包裹封装，将订阅者保存在了自己的innerSubscriber属性中，而RACPassthroughSubscriber的sendNext又会调用innerSubscriber的sendNext。
所以RACPassthroughSubscriber是一个适配器，适配实现<RACSubscriber>接口的不同类型订阅者。
3. RACSubject实现的sendNext是对其所有订阅者发送sendNext消息；RACSubscriber的sendNext实现的sendNext是执行nextBlock(value)。所以需要RACPassthroughSubscriber这样一个适配器类封装各自实现<RACSubscriber>的类。C
